使用插件实现延迟队列
因为使用死信队列去多延迟时，存在一个缺陷，就是指当第一个消息进入死信队列时，它的时间如果很大，后面进来的时间少的消息，也必须等待。
去官网下载对应版本的插件，如果版本不对的话，是安装不了的，这里用 rabbitmq_delayed_message_exchange.3.8.0.ez
使用插件后，延迟是由交换机去处理的，基本的处理结构：
生产者----->交换机延迟-------->队列-------->消费者
安装插件后就会出现这种交换机类型
接着可以去准备交换机和队列
@Configuration
public class DelayedQueueConfig {
//交换机
public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
//队列
public static final String DELAYED_QUEUE_NAME = "delayed.queue";
//routingKey
public static final String DELAYED_ROUTING_KEY = "delayed.routingKey";
//队列
@Bean
public Queue delayedQueue(){
return new Queue(DELAYED_QUEUE_NAME);
}
//交换机
@Bean
public CustomExchange delayedExchange(){
Map arguments = new HashMap<>();
arguments.put("x-delayed-type","direct");
return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",
true,false,arguments);
}
//绑定
@Bean
public Binding delayedQueueBindingExchange(@Qualifier("delayedQueue") Queue delayedQueue,
@Qualifier("delayedExchange") CustomExchange delayedExchange){
return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
//自定义的交换机需要加.noargs()
}
}
创建生产者
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
@Autowired
private RabbitTemplate rabbitTemplate;
@GetMapping("/sendDelayedMsg/{message}/{delayTime}")
public void sendMsg2(@PathVariable String message,@PathVariable Integer delayTime){
log.info("当前时间：{}，发送时长{}毫秒的信息给延迟队列delayed.queue:{}",
new Date().toString(),delayTime,message);
rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME,
DelayedQueueConfig.DELAYED_ROUTING_KEY,message, message1 -> {
//发送延迟消息 延迟时长 ms
message1.getMessageProperties().setDelay(delayTime);
return message1;
});
}
}
创建消费者
@Slf4j
@Component
public class DelayQueueConsumer {
//开始监控
@RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
public void receiveDelayQueue(Message message){
String msg = new String(message.getBody());
log.info("当前时间：{}，收到延迟队列消息：{}",new Date().toString(),msg);
}
}
使用效果
都按照设定的情况去运行了没有出错
rabbitmq来做延迟队列可以很好的利用rabbitmq的特性，去做到其他事情，如，消息可靠发送，消息可靠投递，死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过rabbitmq集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延迟队列不可用或消息丢失。
reids的zset 和java的DelayQueue 还有kafka的时间轮也可以选择
发布确认当出现不明原因的问题的时候，处理无法投递的消息
配置文件中增加属性：
publisher-confirm-type: correlated 发布确认模式
交换机和队列继续使用上面的
生产者（我写的时候用了新的交换机和队列，但是是不必要的）
@Slf4j
@RestController
@RequestMapping("/confirm")
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
@GetMapping("/sendMsg/{message}")
public void sendMessage(@PathVariable String message){
CorrelationData correlationData = new CorrelationData("1");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,
ConfirmConfig.CONFIRM_ROUTING_KEY,message,correlationData);
log.info("发送消息：{}",message);
}
}
消费者
@Slf4j
@Component
public class Consumer {
@RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
public void receiveConfirmMessage(Message message){
String s = new String(message.getBody());
log.info("接受到的队列{}消息：{}",ConfirmConfig.CONFIRM_QUEUE_NAME,s);
}
}
回调接口
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
//这个注解比下面的优先级高
@Autowired
private RabbitTemplate rabbitTemplate;
//注入到接口中
@PostConstruct
public void init(){
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : "";
if(ack){
log.info("交换机以及收到id为：{}的消息",id);
}else{
log.info("交换机还未收到id为：{}的消息，原因：{}",id,cause);
}
}
//可以在消息传递过程中不可以达目的地时 将消息返回给生产者
//只有不可达目的地的时候 才回退
@Override
public void returnedMessage(ReturnedMessage returned) {
log.error("消息{}，被交换机{}退回，退回原因：{}，路由key：{}",
returned.getMessage(),returned.getExchange(),returned.getRoutingKey());
}
}
有两种出现错误的情况
一种是发到交换机的时候就出现错误
要在配置文件中再加属性：
publisher-returns: true
比如在生产者中，把交换机的名字写错，就会失败
回调接口中的confirm方法就是处理这个的，处理后
第二种是，交换机以及收到，但是发送给队列时出错
方法returnedMessage就是用于回调的
