springboot集成RocketMQ过程及使用示例详解

目录

• 1、 说明
• 2、集成过程
• pom文件引用
• yml文件配置
• 消费者
• 生产者
• 测试发送消息接口

1、 说明

Springboot集成RocketMQ时需要特别注意版本问题，否则会出现各类启动报错问题，这里使用的springboot 版本：2.2.7.RELEASE, RocketMq版本：2.2.3

2、集成过程

pom文件引用

这里网上有单独在引用rocketmq-client依赖包的，是不需要的，已经包含在rocketmq-spring-boot-starter依赖包中。

 <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.7.RELEASE</version>
    </parent>
 <dependency>
      <groupId>org.apache.rocketmq</groupId>
      <artifactId>rocketmq-spring-boot-starter</artifactId>
      <version>2.1.1</version>
 </dependency>

yml文件配置

spring:
  # 数据库信息
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/longfamily?useUnicode=true&characterEncoding=utf-8&userSSL=false
    username: root
    password: 123456
##RocketMq配置
rocketmq:
  #nameservice服务器地址（多个以英文逗号隔开）
 # 注意高版本下使用的是name-server，低版本使用的是 nameServer。
  name-server: 10.0.164.31:9876
  producer:
    group: test

这里把spring配置也贴出来，是想说明rocketMq与spring在yaml文件中是同一级别（容易当成spring节点下面的子节点）。否则会启动报错。

消费者

@Component
@RocketMQMessageListener(consumerGroup = \"H\", topic = RocketMQTopic.TOPIC_ORDER)
public class TestMQListener implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println(\"消费者收到信息：===\"+message);
    }
}

消费者是通过RocketMQMessageListener来监听Topc, 通过实现RocketMQListener <T>接口实现其onMessage(T t)方法来处理接收到的消息

生产者

这里同意封装了RocketMQ发送消息的工具类。

@Slf4j
@Component
public class RocketMQUtils {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    /**
     * 异步发送MQ消息
     */
    public void sendMessage(Long id, final String topic, final String context, Boolean isExternal) {
        try {
            log.debug(\"Sending message to MQ topic {}, context {}\", topic, context);
            SendCallback callback = new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    //打印msgId用来以备查验,外部消息发送mq成功则任务置为成功
                    log.info(\"Success sending message to MQ: {}, context: {}, msgId: {}\", topic, context, sendResult.getMsgId());
                }
                @Override
                public void onException(Throwable e) {
                    log.error(\"Failed to send message to MQ {}, msg {}, cause {}\", topic, context, Throwables.getStackTraceAsString(e));
                }
            };
            rocketMQTemplate.asyncSend(topic, context, callback);
        } catch (Exception e) {
            log.error(\"Failed to send message to MQ! message: {}, stackTrace: {}\", context, Throwables.getStackTraceAsString(e));
        }
    }
}

测试发送消息接口

@RestController
@RequestMapping(\"/api/test\")
public class CommonTestController {
    @Autowired
    private RocketMQUtils rocketMQUtils;
    // mq
    @PostMapping(\"/mq/send\")
    public void operateRocketMQ(){
        rocketMQUtils.sendMessage(1L, RocketMQTopic.TOPIC_ORDER,\"firstContext\",true);
    }

运行结果

[2023-02-27 10:17:30.741] [DEBUG] [T:http-nio-8061-exec-3][tid=][Class:c.g.c.u.RocketMQUtils -> sendMessage]|Sending message to MQ topic topic-order, context firstContext
[2023-02-27 10:17:30.760] [INFO] [T:NettyClientPublicExecutor_2][tid=][Class:c.g.c.u.RocketMQUtils -> onSuccess]|Success sending message to MQ: topic-order, context: firstContext, msgId: 0000010169EE18B4AAC2881AB17B0001
消费者收到信息：===firstContext

以上就是springboot集成RocketMQ过程及使用示例详解的详细内容，更多关于springboot集成RocketMQ的资料请关注其它相关文章！