快捷導(dǎo)航

springboot集成RocketMQ過程及使用示例詳解

更新時(shí)間：2023年07月12日 11:25:55 作者：Geroge1226

這篇文章主要為大家介紹了springboot集成RocketMQ過程及使用示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

1、說明

Springboot集成RocketMQ時(shí)需要特別注意版本問題，否則會出現(xiàn)各類啟動報(bào)錯(cuò)問題，這里使用的springboot 版本：2.2.7.RELEASE, RocketMq版本：2.2.3

2、集成過程

pom文件引用

這里網(wǎng)上有單獨(dú)在引用rocketmq-client依賴包的，是不需要的，已經(jīng)包含在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:
  # 數(shù)據(jù)庫信息
  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服務(wù)器地址（多個(gè)以英文逗號隔開）
 # 注意高版本下使用的是name-server，低版本使用的是 nameServer。
  name-server: 10.0.164.31:9876
  producer:
    group: test

這里把spring配置也貼出來，是想說明rocketMq與spring在yaml文件中是同一級別（容易當(dāng)成spring節(jié)點(diǎn)下面的子節(jié)點(diǎn)）。否則會啟動報(bào)錯(cuò)。

消費(fèi)者

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

消費(fèi)者是通過RocketMQMessageListener來監(jiān)聽Topc, 通過實(shí)現(xiàn)RocketMQListener <T>接口實(shí)現(xiàn)其onMessage(T t)方法來處理接收到的消息

生產(chǎn)者

這里同意封裝了RocketMQ發(fā)送消息的工具類。

@Slf4j
@Component
public class RocketMQUtils {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    /**
     * 異步發(fā)送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用來以備查驗(yàn),外部消息發(fā)送mq成功則任務(wù)置為成功
                    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));
        }
    }
}

測試發(fā)送消息接口

@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);
    }

運(yùn)行結(jié)果

[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
消費(fèi)者收到信息：===firstContext

以上就是springboot集成RocketMQ過程及使用示例詳解的詳細(xì)內(nèi)容，更多關(guān)于springboot集成RocketMQ的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: