RocketMq普通消息，死信队列，消息幂等性（redis）

快来打我* 2024-04-01 08:56 64阅读 0赞

> ## 1 介绍 ##
> 
> [RocketMQ][]作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。

> ### 1.1 RocketMQ 特点 ###
> 
>  *  支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
>  *  在一个[队列][Link 1]中可靠的先进先出（FIFO）和**严格的**顺序传递 （RocketMQ可以保证严格的消息顺序，而[ActiveMQ][]无法保证）
>  *  支持拉（pull）和推（push）两种消息模式
>     
>     > pull其实就是消费者主动从MQ中去拉消息，而push则像rabbit MQ一样，是MQ给消费者推送消息。但是RocketMQ的push其实是基于pull来实现的。  
>     > 它会先由一个业务代码从MQ中pull消息，然后再由业务代码push给特定的应用/消费者。其实底层就是一个pull模式
>  *  单一队列百万消息的堆积能力 （RocketMQ提供亿级消息的堆积能力，这不是重点，重点是堆积了亿级的消息后，依然保持写入低延迟）
>  *  支持多种消息协议，如 JMS、MQTT 等
>  *  [分布式][Link 2]高可用的部署架构,满足至少一次消息传递语义（RocketMQ原生就是支持分布式的，而ActiveMQ原生存在单点性）
>  *  提供 docker 镜像用于隔离测试和云集群部署
>  *  提供配置、指标和监控等功能丰富的 Dashboard

> 新建springboot工程并,导入依赖，本文选用rocketmq2.2.2版本
> 
>     <dependency>
>         <groupId>org.apache.rocketmq</groupId>
>         <artifactId>rocketmq-spring-boot-starter</artifactId>
>         <version>2.2.2</version>
>     </dependency>

> YML配置文件
> 
>     rocketmq:
>       name-server: 192.168.198.129:9876
>       producer:
>         group: group1

> 消息提供者
> 
>     @GetMapping("tset")
>         public String send(){
>             //尝试发送消息
>             rocketMQTemplate.convertAndSend("topic1","smoky哥哥呀");
>     //        log.info("尝试发送同步消息:{}",syncSend);
>             return "success";
>         }

> 消息消费者
> 
>     @Component
>     @RocketMQMessageListener(topic = "topic1",consumerGroup = "group1")
>     @Slf4j
>     public class ReceiveConsumer implements RocketMQListener<String>,RocketMQPushConsumerLifecycleListener{
>     
>         @Autowired
>         StringRedisTemplate stringRedisTemplate;
>     
>         @Override
>         public void onMessage(String s) {
>             log.info("接收到MQ消息:{}",s);
>         }
>     
>         @Override
>         public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
>             defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
>                 MessageExt messageExt = list.get(0);
>                 //尝试获取消息体内容
>                 String body = new String(messageExt.getBody());
>                 //尝试获取消息id
>                 String msgId = messageExt.getMsgId();
>                 log.info("获取到消息id为:{}",msgId);
>                 //尝试获取重试次数
>                 int reconsumeTimes = messageExt.getReconsumeTimes();
>                 log.info("获取到当前消息重试次数:{}",reconsumeTimes);
>                 try {
>                     int i = 10/0;
>                     //接口幂等性
>                     Boolean rocketmq = stringRedisTemplate.opsForHash().hasKey("ROCKETMQ", msgId);
>                     if (!rocketmq) {
>                         log.info("接收到ACK消息:{}", body);
>                         //存入redis日志表,代表消费成功
>                         stringRedisTemplate.opsForHash().put("ROCKETMQ",msgId,"1");
>                     }
>                 } catch (Exception e) {
>                     e.printStackTrace();
>                     log.info("异常捕获，尝试ACK重试机制");
>                     if (reconsumeTimes>=3){
>                         //加入死信队列
>                         log.info("业务逻辑:加入死信队列做,做事后补偿");
>                         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
>                     }
>                     //发起重试
>                     log.info("尝试发起ACK重试");
>                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
>                 }
>                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
>             });
>         }
>     }

[RocketMQ]: https://so.csdn.net/so/search?q=RocketMQ&spm=1001.2101.3001.7020
[Link 1]: https://so.csdn.net/so/search?q=%E9%98%9F%E5%88%97&spm=1001.2101.3001.7020
[ActiveMQ]: https://so.csdn.net/so/search?q=ActiveMQ&spm=1001.2101.3001.7020
[Link 2]: https://so.csdn.net/so/search?q=%E5%88%86%E5%B8%83%E5%BC%8F&spm=1001.2101.3001.7020