RabbitMQ真延时队列实现消息提醒功能

一、需求场景

用户可以制定多个计划，同时可给该计划设置是否需要到点提醒，且中途可以取消提醒或修改提醒时间。

二、需要解决的问题

学习过rabbitmq的同学们都知道，通过TTL+死信队列可以实现延时队列的效果，

图片描述

TTL+死信队列实现延时队列示意图

但是这个延时队列有个弊端，即里面的消息死亡并非是异步的，举个例子：

消息1设置的死亡时间是5分钟，消息2设置的死亡时间是10分钟，当消息2比消息1先进入队列时，消息2没有死亡即使消息1已经到达了死亡时间也会被消息2所阻塞，导致无法被消费。这样就无法满足上述需求：每条消息的死亡相互独立这种场景了。

三、解决方法

那有没有既需要延时触发、也可以满足延时时间不一样的场景的方法呢？

有！

那就是rabbitmq的插件大法，安装方法跟添加管理台插件时大同小异。

插件名字叫rabbitmq_delayed_message_exchange 翻译过来就是延时消息交换机

开启该插件后，就会在原来的交换机类型上又多加了一种类型的交换机：x-delayed-message交换机

之后的消息的延时触发都会交给该交换机完成，而无需再使用两个交换机路，两个队列。

1、如何安装？

我是用docker安装rabbitmq的，所以这里只介绍docker安装插件的方法。

这是插件的下载地址 https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

小编用的是手动安装的方式，将安装包拷贝到rabbitmq映射的文件目录下的plugins文件夹

docker cp rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez rabbitmq(rabbitmq的容器名称):/plugins
rm rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez
docker exec -it rabbitmq bin/bash
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2、如何开发？

按照上面的需求，我们结合实际的业务开发，因为涉及一些数据库操作的实体，为避免篇幅过长，推荐配合源码食用。

1）git拉取项目

项目地址：https://github.com/CaiCaiXian/rabbitmq-plan.git

2) 数据库配置

不含创建数据库语句

计划表sql：

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;-- ----------------------------
-- Table structure for qk_plan
-- ----------------------------
DROP TABLE IF EXISTS `qk_plan`;
CREATE TABLE `qk_plan`  (`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '唯一标识',`user_id` bigint(20) NOT NULL COMMENT '用户id',`type_id` int(11) NOT NULL COMMENT '类型id',`title` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL DEFAULT '' COMMENT '标题',`content` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL DEFAULT '' COMMENT '内容',`location` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL DEFAULT '' COMMENT '地点',`start_time` datetime(0) NOT NULL COMMENT '开始时间',`end_time` datetime(0) NOT NULL COMMENT '结束时间',`create_time` datetime(0) NOT NULL COMMENT '创建时间',`version` int(11) NULL DEFAULT NULL COMMENT '版本号',`status` int(11) NULL DEFAULT NULL COMMENT '是否生效',PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 1349648474282508291 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;SET FOREIGN_KEY_CHECKS = 1;

计划类型sql：

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;-- ----------------------------
-- Table structure for qk_plantype
-- ----------------------------
DROP TABLE IF EXISTS `qk_plantype`;
CREATE TABLE `qk_plantype`  (`id` bigint(11) NOT NULL AUTO_INCREMENT,`name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;SET FOREIGN_KEY_CHECKS = 1;

注意一个小细节：

我在计划表中加入了version（版本号）和status（是否生效）两个字段，是为了满足需求场景中提到的 消息可以被修改和取消 的需求。

3）配置交换机，绑定队列

@Configuration
public class DelayMQConfig {/*** 交换机名称*/public final static String DELAY_EXCHANGE_NAME = "delay_exchange";/*** 队列名称*/public final static String DELAY_QUEUE_NAME = "delay_queue";/*** 路由key 不是topic不能使用通配符*/public final static String DELAY_ROUTE_KEY = "delay.notify";/*** 延迟交换机*/@Bean("delayExchange")public CustomExchange delayExchange(){Map<String,Object> args = new HashMap<>();args.put("x-delayed-type","direct");return new CustomExchange(DELAY_EXCHANGE_NAME,"x-delayed-message",true,true,args);}/*** 延迟队列*/@Bean("delayQueue")public Queue delayQueue(){return new Queue(DELAY_QUEUE_NAME,true,false,false);}@Beanpublic Binding bindingDelayExchangeQueue(@Qualifier("delayExchange") Exchange exchange,@Qualifier("delayQueue") Queue queue){return BindingBuilder.bind(queue).to(exchange).with(DELAY_ROUTE_KEY).noargs();}
}

在这里定义一个类型为x-delayed-message的交换机，注意这里返回的是CustomExchange意思是自定义交换机，交换机名称为delay_exchange。

args.put(“x-delayed-type”,“direct”) 将属性 x-delayed-type 设为direct交换机。

同时我们给该交换机绑定一个队列名称为：delay_queue，路由KEY是 delay.notify

4）配置消费者，绑定队列

@Component
public class NotifyListener {@AutowiredPlanService planService;//此处绑定要监听的队列@RabbitListener(queues = DelayMQConfig.DELAY_QUEUE_NAME)@RabbitHandlerpublic void onMessage(PlanDTO msg, Message message, Channel channel) throws IOException {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {//通知用户planService.notifyUser(msg.getId(),msg.getVersion());channel.basicAck(deliveryTag, false);//System.out.println("消息被确认！");} catch (IOException e) {channel.basicNack(deliveryTag, false, false);//System.out.println("消息被否定确认！");}}
}

我们通过队列名给刚刚配置好的队列绑定上消费者，这样就实现了交换机—>队列—>消费者的模型了。

开发者只需要完善notifyUser方法的代码，如发送到邮箱，就可以实现发送提醒的效果了。

注意：

channel.basicAck(deliveryTag, false);中，是否多条确认要设置为false，不能一次确认多条消息，否则会把时间还没到的消息也一块确认了。

5）notifyUser（发送消息提醒）方法

因为计划可能被修改过，也可能被取消，所以我们在发送提醒时要确保版本号和我们消息中的计划的版本号要一致，且消息并没有被取消提醒才可以发送提醒。通过上述提到的version和status字段判断即可。

  @Overridepublic void notifyUser(Long planId,Integer version) {PlanDTO planDTO = planDao.selectPlanById(planId);SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");String now = sdf.format(new Date());//保证计划是生效的且和通知时是同一个版本if(planDTO != null && PlanConstant.CAN_USE.equals(planDTO.getStatus()) && planDTO.getVersion().equals(version)){//发送通知System.out.println(now + " 你消息提醒："+planDTO.toString());}else{System.out.println(now + " " + planDTO.getTitle()+": 该消息已取消提醒");}}

6）编写生产者

无论是更新还是新增计划，我们都需要重新向队列中发送一条消息。

   @Overridepublic boolean saveOrUpdatePlan(PlanDTO planDTO) {PlanEntity planEntity = new PlanEntity();//dto拷贝给数据库操作实体BeanUtil.copyProperties(planDTO,planEntity);try {//判断是更新还是新增Long id = planEntity.getId();if(ObjectUtil.isNotNull(id)){//更新,将数据库的版本号加一Integer old = planDao.selectVersion(id);planEntity.setVersion(old + 1);updateById(planEntity);}else {//新增 初始化版本号planEntity.setVersion(0);save(planEntity);}//获取数据库中最新数据PlanDTO newPlanDTO = planDao.selectPlanById(planEntity.getId());//判断开始时间是不是比当前时间大且消息提醒是生效long x = (planDTO.getStartTime().getTime() - System.currentTimeMillis());if( x >= 0 && planDTO.getStatus().equals(PlanConstant.CAN_USE)){//计划加入消息队列rabbitTemplate.convertAndSend(DelayMQConfig.DELAY_EXCHANGE_NAME, DelayMQConfig.DELAY_ROUTE_KEY,newPlanDTO, msg->{//设置延迟msg.getMessageProperties().setDelay((int)x);return msg;});}return true;}catch (Exception e){e.printStackTrace();return false;}}

其中这段为生产者的核心代码，rabbitTemplate.convertAndSend（交换机名称，路由Key，传递的消息实体，消息配置器）

用lamdba表达式的方式设置消息的延迟时间，单位为毫秒。

rabbitTemplate.convertAndSend(DelayMQConfig.DELAY_EXCHANGE_NAME, DelayMQConfig.DELAY_ROUTE_KEY,newPlanDTO, msg->{//设置延迟msg.getMessageProperties().setDelay((int)x);return msg;
});