• Springboot 集成 RocketMQ(进阶-消息)

    0. 入门篇

    Springboot 集成 RocketMq（入门）-CSDN博客

    本文主要使用RocketMQTemplate完成生成消息。

    1. 异步消息

    1.1 生产者

    1.     @Resource
    2.     private RocketMQTemplate rocketMqTemplate;
    3.  
    4.     @GetMapping("/send/async/{messageBody}")
    5.     public String sendAsyncMsg(@PathVariable("messageBody") String messageBody) {
    6.         rocketMqTemplate.asyncSend("async", messageBody, new SendCallback() {
    7.             @Override
    8.             public void onSuccess(SendResult sendResult) {
    9.                 log.info("sendAsyncMsg success messageBody:{}", messageBody);
    10.             }
    11.  
    12.             @Override
    13.             public void onException(Throwable throwable) {
    14.                 log.error("sendAsyncMsg fail messageBody:{}", messageBody);
    15.             }
    16.         });
    17.         log.info("sendAsyncMsg operate oK");
    18.         return "OK";
    19.     }

    1.2 消费者

    1. @Component
    2. @RocketMQMessageListener(topic = "async", consumerGroup = "async")
    3. @Slf4j
    4. public class MyAsyncConsumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(String message) {
    8.         // 处理消息的逻辑
    9.         log.info("Received async message: {}", message);
    10.     }
    11.  
    12. }

    通过日志可以发现，主线程结束线程日志先于异步日志打印完成。

    2.广播消息

    2.1 生产者

    1.     @Resource
    2.     private RocketMQTemplate rocketMqTemplate;
    3.  
    4.     @GetMapping("/send/broadcast/{messageBody}")
    5.     public String sendBroadcastMsg(@PathVariable("messageBody") String messageBody) {
    6.         // 单向不可靠消息 void 方法无返回值
    7.         SendResult sendResult = rocketMqTemplate.syncSend("broadcast", messageBody);
    8.         log.info("sendBroadcastMsg msgId:{},sendStatus:{}", sendResult.getMsgId(), sendResult.getSendStatus());
    9.         return "OK";
    10.     }

    2.2 消费者

    消费者监听注解上设置 messageModel = MessageModel.BROADCASTING，默认是（MessageModel.CLUSTERING）。

    1. @Component
    2. @RocketMQMessageListener(topic = "broadcast", consumerGroup = "broadcast1", messageModel = MessageModel.BROADCASTING )
    3. @Slf4j
    4. public class MyBroadcast1Consumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(String message) {
    8.         // 处理消息的逻辑
    9.         log.info("MyBroadcast1Consumer Received broadcast message: {}", message);
    10.     }
    11.  
    12. }
    13.  
    14. @Component
    15. @RocketMQMessageListener(topic = "broadcast", consumerGroup = "broadcast2", messageModel = MessageModel.BROADCASTING )
    16. @Slf4j
    17. public class MyBroadcast2Consumer implements RocketMQListener {
    18.  
    19.     @Override
    20.     public void onMessage(String message) {
    21.         // 处理消息的逻辑
    22.         log.info("MyBroadcast2Consumer Received broadcast message: {}", message);
    23.     }
    24.  
    25. }

    3. 延时、定时消息

    3.1 延时消息

    3.1.1 生产者

    1.     @Resource
    2.     private RocketMQTemplate rocketMqTemplate;
    3.  
    4.     @GetMapping("/send/delayed/{messageBody}")
    5.     public String sendDelayedMsg(@PathVariable("messageBody") String messageBody) {
    6.         // 延时10秒发送
    7.         SendResult sendResult = rocketMqTemplate.syncSendDelayTimeSeconds("delayed", messageBody, 10L);
    8.         log.info("sendDelayedMsg msgId:{},sendStatus:{}", sendResult.getMsgId(), sendResult.getSendStatus());
    9.         return "OK";
    10.     }

    3.1.2 消费者

    1. @Component
    2. @RocketMQMessageListener(topic = "delayed", consumerGroup = "delayed")
    3. @Slf4j
    4. public class MyDelayedConsumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(Message message) {
    8.         // 处理消息的逻辑
    9.         log.info("Received delayed message: {}", message);
    10.     }
    11.  
    12. }

    3.2 定时消息

    3.2.1 生产者

    message.setDeliverTimeMs(System.currentTimeMillis() + 10);

    1.     @Resource
    2.     private RocketMQTemplate rocketMqTemplate;
    3.  
    4.     @GetMapping("/send/scheduled/{messageBody}")
    5.     public String sendScheduledMsg(@PathVariable("messageBody") String messageBody) {
    6.         // 指定时间发送 (当前时间 + 30秒)
    7.         SendResult sendResult = rocketMqTemplate.syncSendDeliverTimeMills("scheduled", messageBody, (System.currentTimeMillis() + (30L * 1000L)));
    8.         log.info("sendDelayedMsg msgId:{},sendStatus:{}", sendResult.getMsgId(), sendResult.getSendStatus());
    9.         return "OK";
    10.     }

    3.2.2 消费者

    1. @Component
    2. @RocketMQMessageListener(topic = "scheduled", consumerGroup = "scheduled")
    3. @Slf4j
    4. public class MyScheduledConsumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(Message message) {
    8.         // 处理消息的逻辑
    9.         log.info("Received scheduled message: {}", message);
    10.     }
    11.  
    12. }

    日志显示时间存在误差，网络等通讯耗时可以忽略。

    4.批量消息

    4.1 生产者

    1.     @Resource
    2.     private RocketMQTemplate rocketMqTemplate;
    3.  
    4.     @GetMapping("/send/batch/{messageBody}")
    5.     public String sendBatchMsg(@PathVariable("messageBody") String messageBody) {
    6.         List> messageList = new ArrayList<>();
    7.         for (int i = 0; i < 10; i++) {
    8.             messageList.add(MessageBuilder.withPayload(messageBody + i).build());
    9.         }
    10.         SendResult sendResult = rocketMqTemplate.syncSend("batch", messageList);
    11.         log.info("sendDelayedMsg msgId:{},sendStatus:{}", sendResult.getMsgId(), sendResult.getSendStatus());
    12.         return "OK";
    13.     }

    4.2 消费者

    1. @Component
    2. @RocketMQMessageListener(topic = "batch", consumerGroup = "batch")
    3. @Slf4j
    4. public class MyBatchConsumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(String message) {
    8.         // 处理消息的逻辑
    9.         log.info("Received batch message: {}", message);
    10.     }
    11.  
    12. }

    5.顺序消息

    5.1 局部有序

    局部消息指的是消费者消费某个topic的某个队列中的消息是顺序的【队列扩容时操作部分数据乱序】。

    5.1.1 生产者

    1.     @Resource
    2.     private RocketMQTemplate rocketMqTemplate;
    3.  
    4.     @GetMapping("/send/partOrder/{targetId}/{messageBody}")
    5.     public String sendPartOrderMsg(@PathVariable("targetId") Integer targetId, @PathVariable("messageBody") String messageBody) {
    6.         SendResult sendResult = rocketMqTemplate.syncSendOrderly("partOrder", (targetId + "_" + messageBody), "0");
    7.         log.info("sendPartOrderMsg msgId:{},sendStatus:{}", sendResult.getMsgId(), sendResult.getSendStatus());
    8.         return "OK";
    9.     }

    5.1.2 消费者

    1. @Component
    2. @RocketMQMessageListener(topic = "partOrder", consumerGroup = "partOrder", consumeMode = ConsumeMode.ORDERLY)
    3. @Slf4j
    4. public class MyPartOrderConsumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(String message) {
    8.         // 处理消息的逻辑
    9.         log.info("Received partOrder message: {}", message);
    10.     }
    11.  
    12. }

    5.2 全局有序

    消费者消费全部消息都是顺序的，只能通过一个某个topic只有一个队列才能实现，这种应用场景较少，且性能较差。【向唯一队列中发送消息，队列无法扩展】。

    5.2.1 生产者

    1.     @Resource
    2.     private RocketMQTemplate rocketMqTemplate;
    3.  
    4.     @GetMapping("/send/overAllOrder/{targetId}/{messageBody}")
    5.     public String sendOverallOrderMsg(@PathVariable("targetId") Integer targetId, @PathVariable("messageBody") String messageBody) {
    6.         SendResult sendResult = rocketMqTemplate.syncSendOrderly("overAllOrder", (targetId + "_" + messageBody), targetId.toString());
    7.         log.info("sendOverallOrderMsg msgId:{},sendStatus:{}", sendResult.getMsgId(), sendResult.getSendStatus());
    8.         return "OK";
    9.     }

    5.2.2 消费者

    1. @Component
    2. @RocketMQMessageListener(topic = "overallOrder", consumerGroup = "overallOrder", consumeMode = ConsumeMode.ORDERLY)
    3. @Slf4j
    4. public class MyOverallOrderConsumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(String message) {
    8.         // 处理消息的逻辑
    9.         log.info("Received overallOrder1 message: {}", message);
    10.     }
    11.  
    12. }

    6.事务消息

    6.1 生产者

    1. @RestController
    2. @Slf4j
    3. public class TransactionProducerController {
    4.  
    5.     @Resource
    6.     private RocketMQTemplate rocketMqTemplate;
    7.  
    8.     @GetMapping("/send/transaction/{targetId}/{messageBody}")
    9.     public String sendTransactionMsg(@PathVariable("targetId") Integer targetId, @PathVariable("messageBody") String messageBody) {
    10.         Message message = MessageBuilder.withPayload(messageBody).build();
    11.         TransactionSendResult transaction = rocketMqTemplate.sendMessageInTransaction("transaction", message, targetId);
    12.         return transaction.getTransactionId();
    13.     }
    14.  
    15. }
    16.  
    17. @Slf4j
    18. @RocketMQTransactionListener
    19. public class TransactionMsgListener implements RocketMQLocalTransactionListener {
    20.  
    21.  
    22.     /** 执行本地事务（在发送消息成功时执行）  */
    23.     @Override
    24.     public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
    25.         //模拟一个处理结果
    26.         int index=2;
    27.         /**
    28.          * 模拟返回事务状态
    29.          */
    30.         switch (index){
    31.             case 1:
    32.                 //处理业务
    33.                 String jsonStr = new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
    34.                 log.info("本地事务回滚，回滚消息，"+jsonStr);
    35.                 //返回ROLLBACK状态的消息会被丢弃
    36.                 return RocketMQLocalTransactionState.ROLLBACK;
    37.             case 2:
    38.                 //返回UNKNOW状态的消息会等待Broker进行事务状态回查
    39.                 log.info("需要等待Broker进行事务状态回查");
    40.                 return RocketMQLocalTransactionState.UNKNOWN;
    41.             default:
    42.                 log.info("事务提交，消息正常处理");
    43.                 //返回COMMIT状态的消息会立即被消费者消费到
    44.                 return RocketMQLocalTransactionState.COMMIT;
    45.         }
    46.     }
    47.     /**
    48.      * 检查本地事务的状态
    49.      * 回查间隔时间：系统默认每隔60秒发起一次定时任务，对未提交的半事务消息进行回查，共持续12小时。
    50.      * 第一次消息回查最快
    51.      */
    52.     @Override
    53.     public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
    54.         String transactionId = message.getHeaders().get("__transactionId__").toString();
    55.         log.info("检查本地事务状态，transactionId：{}", transactionId);
    56.         return RocketMQLocalTransactionState.COMMIT;
    57.     }
    58. }

    6.2 消费者

    1. @Component
    2. @RocketMQMessageListener(topic = "transaction", consumerGroup = "transaction")
    3. @Slf4j
    4. public class MyTransactionConsumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(String message) {
    8.         // 处理消息的逻辑
    9.         log.info("Received transaction message: {}", message);
    10.     }
    11.  
    12. }

    7.单项消息

    单向发送只负责发送消息，不等待RocketMQ服务器返回的发送结果，也不提供回调函数来接收RocketMQ服务器的响应结果，只负责发送至于发送成功还是发送失败并不考虑。通常用于对可靠性要求不高的场景。

    7.1 生产者

    1.     @Resource
    2.     private RocketMQTemplate rocketMqTemplate;
    3.  
    4.     @GetMapping("/send/oneWay/{messageBody}")
    5.     public String sendOneWayMsg(@PathVariable("messageBody") String messageBody) {
    6.         // 单向不可靠消息 void 方法无返回值
    7.         rocketMqTemplate.sendOneWay("oneWay", messageBody);
    8.         return "OK";
    9.     }

    7.2 消费者

    1. @Component
    2. @RocketMQMessageListener(topic = "oneWay", consumerGroup = "oneWay")
    3. @Slf4j
    4. public class MyOneWayConsumer implements RocketMQListener {
    5.  
    6.     @Override
    7.     public void onMessage(String message) {
    8.         // 处理消息的逻辑
    9.         log.info("Received oneWay message: {}", message);
    10.     }
    11.  
    12. }

  • 相关阅读:
    [机缘参悟-66]：怎样才能让别人愿意帮你：利益共享法则、“大道”、“人性”
    三菱PLC FX3U脉冲轴点动功能块(MC_Jog)
    攻防世界题目练习——Crypto密码新手+引导模式（二）（持续更新）
    读书记录 《就算生得暗淡,也要活得光彩》
    LeetCode刷题系列 -- 78. 子集
    TDengine3.0：解决高基数问题的时序数据库设计思路
    nodejs+vue网络课程在线考试系统an7ib
    5.Vue-在Vue框架中实现Vue的增删改查
    网络摄像头（IPC）介绍：类型、供电、镜头、夜视等
    最新版手机软件App下载排行网站源码/App应用商店源码
  • 原文地址：https://blog.csdn.net/qq_34253002/article/details/134243344