• (十九)ATP应用测试平台——springboot集成RocketMQ案例实战

    前言

    本节内容是关于RocketMQ消息中间键的实战内容，主要介绍在springboot项目中如何集成使用RocketMQ消息中间键，包括消息的发送、消息的接收以及RocketMQ的一些配置说明，以及效果说明。话不多说，开始实战内容。

    正文

    • 搭建RocketMQ集群

    参考博客：Docker环境下使用docker-compose一键式搭建RocketMQ（4.5.0版本）集群及其管理工具(外网版)_北溟溟的博客-CSDN博客

    • 引入RocketMQ的pom依赖

    1. <dependency>
    2. 	<groupId>org.apache.rocketmq</groupId>
    3. 	<artifactId>rocketmq-spring-boot-starter</artifactId>
    4. 	<version>2.2.2</version>
    5. </dependency>

    •  在配置文件application.yml引入RocketMQ配置 
    1. #rocketmq配置
    2. rocketmq:
    3.   name-server: 192.168.56.10:9876;192.168.56.10:9877
    4.   producer:
    5.     #生产者组名称
    6.     group: atp-producer
    7.     #命名空间
    8.     namespace: atp
    9.     #异步消息发送失败重试次数，默认是2
    10.     retry-times-when-send-async-failed: 2
    11.     #发送消息超时时间，默认2000ms
    12.     send-message-timeout: 2000
    13.     #消息的最大长度：默认1024 * 1024 * 4(默认4M)
    14.     max-message-size: 40000000
    15.     #压缩消息阈值，超过4k就压缩
    16.     compress-message-body-threshold: 4096
    17.     #是否发送失败，重试另外的broker
    18.     retry-next-server: false
    19.     #是否启用消息追踪
    20.     enable-msg-trace: false
    21.     #默认追踪的主题
    22.     customized-trace-topic: RMQ_SYS_TRACE_TOPIC
    23.     #消息发送失败重试的次数
    24.     retry-times-when-send-failed: 2

    •  创建消息的生产者

    ①创建RocketMQ消息的生产者RocketProducer.java

    1. package com.yundi.atp.platform.rocketmq;
    2.  
    3. import lombok.extern.slf4j.Slf4j;
    4. import org.apache.rocketmq.client.producer.*;
    5. import org.apache.rocketmq.spring.core.RocketMQTemplate;
    6. import org.apache.rocketmq.spring.support.RocketMQHeaders;
    7. import org.springframework.beans.factory.annotation.Autowired;
    8. import org.springframework.messaging.Message;
    9. import org.springframework.messaging.support.MessageBuilder;
    10. import org.springframework.stereotype.Component;
    11.  
    12.  
    13. @Component
    14. @Slf4j
    15. public class RocketProducer {
    16.     @Autowired
    17.     private RocketMQTemplate rocketMQTemplate;
    18.  
    19.     /**
    20.      * 发送同步消息：消息响应后发送下一条消息
    21.      *
    22.      * @param topic 消息主题
    23.      * @param tag   消息tag
    24.      * @param key   业务号
    25.      * @param data  消息内容
    26.      */
    27.     public void sendSyncMsg(String topic, String tag, String key, String data) {
    28.         //消息
    29.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
    30.         //主题
    31.         String destination = topic + ":" + tag;
    32.         SendResult sendResult = rocketMQTemplate.syncSend(destination, message);
    33.         log.info("【RocketMQ】发送同步消息：{}", sendResult);
    34.     }
    35.  
    36.     /**
    37.      * 发送异步消息：异步回调通知消息发送的状况
    38.      *
    39.      * @param topic 消息主题
    40.      * @param tag   消息tag
    41.      * @param key   业务号
    42.      * @param data  消息内容
    43.      */
    44.     public void sendAsyncMsg(String topic, String tag, String key, String data) {
    45.         //消息
    46.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
    47.         //主题
    48.         String destination = topic + ":" + tag;
    49.         rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
    50.             @Override
    51.             public void onSuccess(SendResult sendResult) {
    52.                 log.info("【RocketMQ】发送异步消息：{}", sendResult);
    53.             }
    54.  
    55.             @Override
    56.             public void onException(Throwable e) {
    57.                 log.info("【RocketMQ】发送异步消息异常：{}", e.getMessage());
    58.             }
    59.         });
    60.     }
    61.  
    62.  
    63.     /**
    64.      * 发送单向消息：消息发送后无响应，可靠性差，效率高
    65.      *
    66.      * @param topic 消息主题
    67.      * @param tag   消息tag
    68.      * @param key   业务号
    69.      * @param data  消息内容
    70.      */
    71.     public void sendOneWayMsg(String topic, String tag, String key, String data) {
    72.         //消息
    73.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
    74.         //主题
    75.         String destination = topic + ":" + tag;
    76.         rocketMQTemplate.sendOneWay(destination, message);
    77.     }
    78.  
    79.  
    80.     /**
    81.      * 同步延迟消息
    82.      *
    83.      * @param topic      主题
    84.      * @param tag        标签
    85.      * @param key        业务号
    86.      * @param data       消息体
    87.      * @param timeout    发送消息的过期时间
    88.      * @param delayLevel 延迟等级-----固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    89.      */
    90.     public void sendSyncDelayMsg(String topic, String tag, String key, String data, long timeout, int delayLevel) {
    91.         //消息
    92.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
    93.         //主题
    94.         String destination = topic + ":" + tag;
    95.         SendResult sendResult = rocketMQTemplate.syncSend(destination, message, timeout, delayLevel);
    96.         log.info("【RocketMQ】发送同步延迟消息：{}", sendResult);
    97.     }
    98.  
    99.  
    100.     /**
    101.      * 异步延迟消息
    102.      *
    103.      * @param topic      主题
    104.      * @param tag        标签
    105.      * @param key        业务号
    106.      * @param data       消息体
    107.      * @param timeout    发送消息的过期时间
    108.      * @param delayLevel 延迟等级-----固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    109.      */
    110.     public void sendAsyncDelayMsg(String topic, String tag, String key, String data, long timeout, int delayLevel) {
    111.         //消息
    112.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
    113.         //主题
    114.         String destination = topic + ":" + tag;
    115.         rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
    116.             @Override
    117.             public void onSuccess(SendResult sendResult) {
    118.                 log.info("【RocketMQ】发送异步延迟消息：{}", sendResult);
    119.             }
    120.  
    121.             @Override
    122.             public void onException(Throwable e) {
    123.                 log.info("【RocketMQ】发送异步延迟消息异常：{}", e.getMessage());
    124.             }
    125.         }, timeout, delayLevel);
    126.     }
    127.  
    128.  
    129.     /**
    130.      * 同步顺序消息
    131.      *
    132.      * @param topic 主题
    133.      * @param tag   标签
    134.      * @param key   业务号
    135.      * @param data  消息体
    136.      */
    137.     public void sendSyncOrderlyMsg(String topic, String tag, String key, String data) {
    138.         //消息
    139.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
    140.         //主题
    141.         String destination = topic + ":" + tag;
    142.         SendResult sendResult = rocketMQTemplate.syncSendOrderly(destination, message, key);
    143.         log.info("【RocketMQ】发送同步顺序消息：{}", sendResult);
    144.     }
    145.  
    146.  
    147.     /**
    148.      * 异步顺序消息
    149.      *
    150.      * @param topic 主题
    151.      * @param tag   标签
    152.      * @param key   业务号
    153.      * @param data  消息体
    154.      */
    155.     public void sendAsyncOrderlyMsg(String topic, String tag, String key, String data) {
    156.         //消息
    157.         Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
    158.         //主题
    159.         String destination = topic + ":" + tag;
    160.         rocketMQTemplate.asyncSendOrderly(destination, message, key, new SendCallback() {
    161.             @Override
    162.             public void onSuccess(SendResult sendResult) {
    163.                 log.info("【RocketMQ】发送异步顺序消息：{}", sendResult);
    164.             }
    165.  
    166.             @Override
    167.             public void onException(Throwable e) {
    168.                 log.info("【RocketMQ】发送异步顺序消息异常：{}", e.getMessage());
    169.             }
    170.         });
    171.     }
    172.  
    173.  
    174.     
    175.  
    176. }

    ②创建消息发送的web接口

    1. package com.yundi.atp.platform.module.test.controller;
    2.  
    3. import com.yundi.atp.platform.common.Result;
    4. import com.yundi.atp.platform.rocketmq.RocketConstant;
    5. import com.yundi.atp.platform.rocketmq.RocketProducer;
    6. import io.swagger.annotations.Api;
    7. import org.springframework.beans.factory.annotation.Autowired;
    8. import org.springframework.web.bind.annotation.GetMapping;
    9. import org.springframework.web.bind.annotation.PathVariable;
    10. import org.springframework.web.bind.annotation.RequestMapping;
    11. import org.springframework.web.bind.annotation.RestController;
    12.  
    13. import java.util.UUID;
    14.  
    15.  
    16. @Api(tags = {"Springboot集成RocketMQ测试"})
    17. @RestController
    18. @RequestMapping(value = "/rocket")
    19. public class RocketContoller {
    20.     @Autowired
    21.     private RocketProducer rocketProducer;
    22.  
    23.     @GetMapping(value = "/sendRocketTestMsg/{topic}/{msg}")
    24.     public Result sendKafkaTestMsg(@PathVariable(value = "topic") String topic, @PathVariable(value = "msg") String msg) {
    25.         rocketProducer.sendSyncMsg(topic, RocketConstant.ROCKET_TAG, UUID.randomUUID().toString(), msg);
    26.         return Result.success();
    27.     }
    28. }

    ③创建前端消息发送界面

    2.  
    4.

    • 创建消息的消费者

    ①创建消费者RocketConsumer.java

    1. package com.yundi.atp.platform.rocketmq;
    2.  
    3. import lombok.extern.slf4j.Slf4j;
    4. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    5. import org.apache.rocketmq.spring.core.RocketMQListener;
    6. import org.springframework.stereotype.Component;
    7.  
    8. /**
    9.  * @Description: rocket消费者
    10.  *
    11.  * consumerGroup：消费者组
    12.  * topic：消费主题
    13.  * selectorType: 选则器模式，默认SelectorType.TAG
    14.  * selectorExpression：消费模式的值
    15.  * consumeMode：消费者模式 CONCURRENTLY（默认消费模式）  ORDERLY（顺序消费）
    16.  * messageModel：消息模式 CLUSTERING（集群模式） BROADCASTING（广播模式）
    17.  * consumeThreadNumber：消费者线程数
    18.  * maxReconsumeTimes：最大重复消费次数
    19.  * consumeTimeout：消费过期时间
    20.  * replyTimeout：重试时间
    21.  * enableMsgTrace：是否允许消息追踪
    22.  * customizedTraceTopic：自定义的消息追踪主题
    23.  * nameServer：注册服务器
    24.  * namespace：名称空间
    25.  *
    26.  * @Date: 2021/11/3 18:31
    27.  * @Version: 1.0.0
    28.  */
    29. @Slf4j
    30. @Component
    31. @RocketMQMessageListener(consumerGroup = RocketConstant.ROCKET_CONSUMER_GROUP,
    32.         topic = RocketConstant.ROCKET_TOPIC,
    33.         selectorExpression = RocketConstant.ROCKET_TAG,
    34.         namespace = RocketConstant.ROCKET_NAMESPACE)
    35. public class RocketConsumer implements RocketMQListener {
    36.  
    37.     @Override
    38.     public void onMessage(String message) {
    39.         log.info("message-------------------------------------->:{}", message);
    40.     }
    41. }

    ②消息常量定义RocketConstant.java

    1. package com.yundi.atp.platform.rocketmq;
    2.  
    3. /**
    4.  * @Description: rocketmq常量定义
    5.  * @Date: 2022/10/20 14:24
    6.  * @Version: 1.0.0
    7.  */
    8. public class RocketConstant {
    9.     /**
    10.      * 消费者组
    11.      */
    12.     public final static String ROCKET_CONSUMER_GROUP = "atp-consumer";
    13.     /**
    14.      * 主题
    15.      */
    16.     public final static String ROCKET_TOPIC = "atp";
    17.     /**
    18.      * tag
    19.      */
    20.     public final static String ROCKET_TAG = "app";
    21.     /**
    22.      * 名称空间
    23.      */
    24.     public final static String ROCKET_NAMESPACE = "atp";
    25.  
    26. }

    •  验证结果

    ①启动RocketMQ集群，保证RocketMQ是启动状态

    ②启动后端与前端服务

    ③发送消息测试，控制台有消息，后端也可以消费到消息

     

    结语

    至此，关于springboot集成RocketMQ案例实战，我们下期见。。。

  • 相关阅读:
    STM32_3（GPIO）
    c++面试八股文
    Android音频框架之OpenSL ES、AAudio、Oboe介绍(二)
    DC电源模块对效率有什么要求？
    【云原生】kubernetes中pod的生命周期、探测钩子的实战应用案例解析
    代码管理工具-Git基础介绍及常用技巧
    构建强大的产品级NLP系统：PaddleNLP Pipelines端到端流水线框架解析
    Linux——进程控制之替换
    二次规划的基础知识理论（更新中）
    C#练习题7和8
  • 原文地址：https://blog.csdn.net/yprufeng/article/details/127432380