1 介绍
RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。
1.1 RocketMQ 特点
- 支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
- 在一个队列中可靠的先进先出（FIFO）和严格的顺序传递 （RocketMQ可以保证严格的消息顺序，而ActiveMQ无法保证）
- 支持拉（pull）和推（push）两种消息模式
pull其实就是消费者主动从MQ中去拉消息，而push则像rabbit MQ一样，是MQ给消费者推送消息。但是RocketMQ的push其实是基于pull来实现的。
它会先由一个业务代码从MQ中pull消息，然后再由业务代码push给特定的应用/消费者。其实底层就是一个pull模式
- 单一队列百万消息的堆积能力 （RocketMQ提供亿级消息的堆积能力，这不是重点，重点是堆积了亿级的消息后，依然保持写入低延迟）
- 支持多种消息协议，如 JMS、MQTT 等
- 分布式高可用的部署架构,满足至少一次消息传递语义（RocketMQ原生就是支持分布式的，而ActiveMQ原生存在单点性）
- 提供 docker 镜像用于隔离测试和云集群部署
- 提供配置、指标和监控等功能丰富的 Dashboard
新建springboot工程并,导入依赖，本文选用rocketmq2.2.2版本
1. <dependency> 2. <groupId>org.apache.rocketmq</groupId> 3. <artifactId>rocketmq-spring-boot-starter</artifactId> 4. <version>2.2.2</version> 5. </dependency>
YML配置文件
1. rocketmq: 2. name-server: 192.168.198.129:9876 3. producer: 4. group: group1
消息提供者
1. @GetMapping("tset") 2. public String send(){ 3. //尝试发送消息 4. rocketMQTemplate.convertAndSend("topic1","smoky哥哥呀"); 5. // log.info("尝试发送同步消息:{}",syncSend); 6. return "success"; 7. }
消息消费者
1. @Component 2. @RocketMQMessageListener(topic = "topic1",consumerGroup = "group1") 3. @Slf4j 4. public class ReceiveConsumer implements RocketMQListener<String>,RocketMQPushConsumerLifecycleListener{ 5. 6. @Autowired 7. StringRedisTemplate stringRedisTemplate; 8. 9. @Override 10. public void onMessage(String s) { 11. log.info("接收到MQ消息:{}",s); 12. } 13. 14. @Override 15. public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) { 16. defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> { 17. MessageExt messageExt = list.get(0); 18. //尝试获取消息体内容 19. String body = new String(messageExt.getBody()); 20. //尝试获取消息id 21. String msgId = messageExt.getMsgId(); 22. log.info("获取到消息id为:{}",msgId); 23. //尝试获取重试次数 24. int reconsumeTimes = messageExt.getReconsumeTimes(); 25. log.info("获取到当前消息重试次数:{}",reconsumeTimes); 26. try { 27. int i = 10/0; 28. //接口幂等性 29. Boolean rocketmq = stringRedisTemplate.opsForHash().hasKey("ROCKETMQ", msgId); 30. if (!rocketmq) { 31. log.info("接收到ACK消息:{}", body); 32. //存入redis日志表,代表消费成功 33. stringRedisTemplate.opsForHash().put("ROCKETMQ",msgId,"1"); 34. } 35. } catch (Exception e) { 36. e.printStackTrace(); 37. log.info("异常捕获，尝试ACK重试机制"); 38. if (reconsumeTimes>=3){ 39. //加入死信队列 40. log.info("业务逻辑:加入死信队列做,做事后补偿"); 41. return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 42. } 43. //发起重试 44. log.info("尝试发起ACK重试"); 45. return ConsumeConcurrentlyStatus.RECONSUME_LATER; 46. } 47. return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 48. }); 49. } 50. }
