问题
前面所讲的延迟队列有一个不足之处，比如现在有一个需求需要延迟半个小时的消息，那么就只有添加一个新的队列。那就意味着，每新增一个不同时间需求，就会新创建一个队列。
解决方案
应该讲消息的时间不要跟队列绑定，应该交给消息的生产者，由发送消息来指定延迟时间。这样就可以定义个通用的队列。
方案图
代码
配置类
package com.xkj.org.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
* TTL队列，配置文件代码
*/
@Configuration
public class TtlQueueConfig {
//普通交换机
public static final String X_EXCHANGE = "X";
//死信交换机
public static final String Y_HEAD_LETTER_EXCHANGE = "Y";
//普通队列
public static final String QUEUE_A = "QA";
public static final String QUEUE_B = "QB";
public static final String QUEUE_C = "QC";
//死信队列
public static final String DEAD_LETTER_QUEUE = "QD";
/**
* 声明普通交换机X,bean的别名xExchange
* @return
*/
@Bean("xExchange")
public DirectExchange xExchange() {
return new DirectExchange(X_EXCHANGE);
}
/**
* 声明死信交换机Y,bean的别名yExchange
* @return
*/
@Bean("yExchange")
public DirectExchange yExchange() {
return new DirectExchange(Y_HEAD_LETTER_EXCHANGE);
}
/**
* 声明普通队列QA
* @return
*/
@Bean("queueA")
public Queue queueA() {
Map<String, Object> arguments = new HashMap<>(3);
//设置死信交换机
arguments.put("x-dead-letter-exchange", Y_HEAD_LETTER_EXCHANGE);
//声明死信的routingey
arguments.put("x-dead-letter-routing-key", "YD");
//设置消息过期时间ttl为10s
arguments.put("x-message-ttl", 10000);
return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
}
/**
* 声明普通队列QB
* @return
*/
@Bean("queueB")
public Queue queueB() {
Map<String, Object> arguments = new HashMap<>(3);
//设置死信交换机
arguments.put("x-dead-letter-exchange", Y_HEAD_LETTER_EXCHANGE);
//声明死信的routingKey
arguments.put("x-dead-letter-routing-key", "YD");
//设置消息过期时间ttl为40s
arguments.put("x-message-ttl", 40000);
return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
}
/**
* 声明普通队列QC
* @return
*/
@Bean("queueC")
public Queue queueC() {
Map<String, Object> arguments = new HashMap<>(2);
//设置死信交换机
arguments.put("x-dead-letter-exchange", Y_HEAD_LETTER_EXCHANGE);
//设置死信的routingKey
arguments.put("x-dead-letter-routing-key", "YD");
//这里不要设置TTL时间
return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();
}
/**
* 声明死信队列QD
* @return
*/
@Bean("queueD")
public Queue queueD() {
return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
}
/**
* 将队列QA绑定到交换机X上，指定routingKey为XA
* @param queueA
* @param xExchange
* @return
*/
@Bean
public Binding queueABindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange")DirectExchange xExchange) {
return BindingBuilder.bind(queueA).to(xExchange).with("XA");
}
/**
* 将队列QB绑定到交换机X上，指定routingKey为XB
* @param queueB
* @param xExchange
* @return
*/
@Bean
public Binding queueBBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") DirectExchange xExchange) {
return BindingBuilder.bind(queueB).to(xExchange).with("XB");
}
/**
* 将队列QC绑定到交换机X上，指定routingKey为XC
* @param queueC
* @param xExchange
* @return
*/
@Bean
public Binding queueCBindingX(@Qualifier("queueC") Queue queueC, @Qualifier("xExchange") DirectExchange xExchange) {
return BindingBuilder.bind(queueC).to(xExchange).with("XC");
}
/**
* 将队列QD绑定到交换机Y上，指定routingKey为YD
* @param queueD
* @param yExchange
* @return
*/
@Bean
public Binding queueDBindingY(@Qualifier("queueD") Queue queueD, @Qualifier("yExchange") DirectExchange yExchange) {
return BindingBuilder.bind(queueD).to(yExchange).with("YD");
}
}
生产者
package com.xkj.org.controller;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;
/**
* 消息生产者
*/
@Slf4j
@RestController
@RequestMapping("/ttl")
@Api(tags = "消息生产者", description = "消息生产者控制器")
public class MessageController {
@Autowired
private RabbitTemplate rabbitTemplate;
@ApiOperation("消息发送测试")
@GetMapping("/sendMsg/{msg}")
public void sendMsg(@ApiParam(value = "发送的消息内容", required = true) @PathVariable("msg") String message) {
log.info("当前时间{}，发送一条消息给两个队列：{}", new Date().toString(), message);
rabbitTemplate.convertAndSend("X", "XA", "ttl=10s的消息：" + message);
rabbitTemplate.convertAndSend("X", "XB", "ttl=40s的消息：" + message);
}
@ApiOperation("发送带过期时间的消息")
@GetMapping("/sendExpiredMsg/{msg}/{ttl}")
public void sendMsgExpired(@ApiParam(value = "消息内容", required = true)@PathVariable("msg") String message,
@ApiParam(value = "ttl时间", required = true)@PathVariable("ttl") String ttlTime) {
log.info("当前时间{}，发送一条消息给队列QC：{}，ttl={}", new Date().toString(), message, ttlTime);
rabbitTemplate.convertAndSend("X", "XC", message, msg -> {
msg.getMessageProperties().setExpiration(ttlTime);
return msg;
});
}
}
消费者
package com.xkj.org.listener;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* 消费者
*/
@Slf4j
@Component
public class DeadLetterQueueConsumer {
@RabbitListener(queues = "QD")
public void receiveD(Message message, Channel channel) throws Exception {
String msg = new String(message.getBody(), "UTF-8");
log.info("当前时间：{}，收到死信队列的消息：{}", new Date().toString(), msg);
}
}
问题
改造代码后，发送两条消息，一条ttl为2s,另一条ttl为10s。ttl为2s的消息也要等到10s后才会收到。
原因
因为队列是先进先出的，消息需要排队，第一条消息10s才会发出去，第二条消息2s发出去，但是由于10s的消息没有发出去，2s的消息就只有等待。
解决方案
使用rabbitmq插件来解决,请看后续博文。
![[网鼎杯 2020 朱雀组]Nmap（详细解读版）](https://i-blog.csdnimg.cn/direct/f8c1ed139c0f405f844029529c49cf4d.png)
