RabbitMQ+springboot用延迟插件实现延迟消息的发送

延迟队列：其实就是死信队列中消息过期的特殊情况

延迟队列应用场景：

可以用死信队列来实现，不过死信队列要等上一个消息消费成功，才会进行下一个消息的消费，这时候就需要用到延迟插件了，不过要线在docker上装一个插件

安装过程（Linux【Docker】）

前置条件是在Docker中部署过RabbitMq。

1、打开你的远程工具，首先查看docker中已有的容器，主要是为了查看rabbitmq的容器ID

2、将本地下载好的压缩包传到服务器某文件夹下，然后将其复制到Docker中的RabbitMq容器中的plugins文件夹下。

docker cp /home/rabbitmq_delayed_message_exchange-3.8.0.ez a687ef46141b:/plugins

3、进入容器查看该目录下是否有该压缩包。

进入容器命令：（通过容器号或者容器名）

docker exec -it a687ef46141b bash

4、同样在容器中的命令行执行一下命令添加插件。

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

5、退出容器，重启该容器。

6、在管理端即同样可以看到新增了一种交换机模式。

总结：以上就是RabbitMQ的延迟插件的安装过程！

基于插件的延迟队列DEMO

成功安装RabbitMQ的延迟插件之后，我们就可以尝试写一个延迟队列来验证一下是否可以解决上述问题。

首先我们的测试环境是在一个Springboot的框架下完成！

1、最先写配置类

/*** 定义延迟交换机*/
@Configuration
public class RabbitMQDelayedConfig {//队列private static final String DELAYQUEUE = "delayedqueue";//交换机private static final String DELAYEXCHANGE = "delayedExchange";@Beanpublic Queue delayqueue(){return new Queue(DELAYQUEUE);}//自定义延迟交换机@Beanpublic CustomExchange delayedExchange(){Map<String, Object> arguments = new HashMap<>();arguments.put("x-delayed-type","direct");/*** 1、交换机名称* 2、交换机类型* 3、是否需要持久化* 4、是否需要自动删除* 5、其他参数*/return new CustomExchange(DELAYEXCHANGE,"x-delayed-message",true,false,arguments);}//绑定队列和延迟交换机@Beanpublic Binding delaybinding(){return BindingBuilder.bind(delayqueue()).to(delayedExchange()).with("sectest").noargs();}
}

2、先写生产者

/*** 基于插件的延迟队列* 消息生产者*/
@Service
@Slf4j
public class DelayMQSender {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendmsg(String message,Integer delaytime){log.info("当前时间：{},发送时长{}信息给延迟队列：{}",new Date().toString(),delaytime,message);rabbitTemplate.convertAndSend("delayedExchange","sectest",message,msg->{//设置发送消息的延长时间 单位：msmsg.getMessageProperties().setDelay(delaytime);return msg;});}
}

3、再写消费者

/*** 基于插件的延迟队列* 消息的消费者*/
@Service
@Slf4j
public class DelayMQReceiver {@RabbitListener(queues = "delayedqueue")public void receivemsg(Message messages){String msg = new String(messages.getBody());log.info("当前时间：{},接收时长信息给延迟队列：{}",new Date().toString(),msg);}
}

4、进行测试

将模拟请求放在了一个简易的网页上，点击后输出如下结果，证明当先发送了20s延时的消息，再发送2s延时的消息，在2s后消息正常被消费，基于插件的延迟队列完美解决了问题。

**【思考】：**如果在实际业务场景中使用延迟队列的话，那就需要服务端在消息被消费之后主动告诉前端消费的结果，如果是这样的话，那么Ajxs的通信方式是单双工通信，只能前端主动访问后端并返回结果，后端无法主动发送消息，应该使用Websocket来进行通信才可，websocket是长连接，不同于http的短连接，可以实现全双工通信，前后端都可以主动发送消息。