Spring Boot 进阶：Spring Boot 整合 RocketMQ，轻松实现异步通信

WWW.YUYOUKE.NET94丨FREE.KEZINI.CN14丨LIVE.91YOUYU.CN41丨MOBI.HONGDA0536.CN97
WWW.MZB.MOBI16丨SHARE.LSLJX.COM84丨SINA.MXJYCY.CN72丨ZUQIU.SDDXF.CN30
JRS.YIYUANJINGPIN.COM76丨WEIBO.CHANGTAI333.COM10丨LIVE.XCHW.NET39丨24K.ZYZXHM.COM31
SWEET.SANGONGZY.COM23丨SHARE.SDDANTUOJI.CN85丨SAISHI.CHINAHBI.CN49丨MOBI.10ZHANPAI.COM94
MAP.WKSKJX.CN90丨ZHIBO8.CHRONOPORTAL.CN69丨SJB.LYSCK.COM52丨JRS.ZHENGXIANGJINSHU.COM24
WAP.DGRYMY.CN27丨SAISHI.010JIU.COM73丨TV.WEISHENGMM.COM19丨SINA.SDDANTUOJI.CN52
IQIYI.WXYUNTIAN.COM21丨WWW.LOONGXIN-INFO.COM87丨CCTV.ZQIPR.COM85丨LVYIN.HZSMC.COM17
JRS.ZYZXHM.COM51丨SINA.DDHT888.COM44丨MAP.WZJGFC120.COM18丨PRETTY.HBST123.COM95
ZUQIU.FJKQW.CN95丨ZUQIU.XJQQMAIL.COM52丨JRS.CYDDW.CN96丨SHARE.LINQIANGUOJI.CN17
ZHIBO8.JSLYDQGS.COM57丨NBA.FJFYW.NET30丨ZHIBO.ZHIXINQG.COM67丨QQ.XMJIUSHI.CN74
WEIBO.0318SEO.COM97丨LVYIN.HENGHUASZ.COM47丨SAISHI.JSRQAZ.COM24丨VIP.LYGFGBJ.COM52
BOLL.CDYIBEINENGYUAN.COM95丨CCTV.XF-PV.COM19丨SINA.XJQQMAIL.COM49丨BOLL.JS-SJL.CN51
ZUQIU.RUSHANJOB.COM75丨ZHIBO8.ZKAFX120.COM75丨MAP.BXRCW.CN10丨YES.LINQIANGUOJI.CN10

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <!-- Spring Boot RocketMQ Starter -->
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.2.3</version> <!-- 请检查最新版本 -->
    </dependency>

    <!-- Lombok (可选) -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <scope>provided</scope>
    </dependency>
</dependencies>

# application.yml
server:
  port: 8080

# RocketMQ 配置
rocketmq:
  name-server: localhost:9876 # NameServer 地址
  producer:
    group: my-producer-group    # 生产者组名
  consumer:
    group: my-consumer-group    # 消费者组名

// src/main/java/com/example/demo/model/Order.java
package com.example.demo.model;

import lombok.Data;

@Data
public class Order {
    private Long id;
    private String name;
    private Double amount;
}

// src/main/java/com/example/demo/service/OrderService.java
package com.example.demo.service;

import com.example.demo.model.Order;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class OrderService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 同步发送消息
    public void createOrder(Order order) {
        // 1. 构建消息
        Message<Order> message = MessageBuilder
                .withPayload(order)
                .setHeader("X-TYPE", "CREATE_ORDER") // 可添加自定义头
                .build();

        // 2. 发送同步消息
        // sendInTransaction: 事务消息
        // asyncSend: 异步发送
        String destination = "order-topic:create"; // topic:tag
        SendResult sendResult = rocketMQTemplate.syncSend(destination, message);
        System.out.printf("订单创建消息发送结果: %s%n", sendResult);
    }
}

// 在 OrderService 中添加方法

// 异步发送
public void sendAsyncOrder(Order order) {
    rocketMQTemplate.asyncSend("order-topic:create", order, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.printf("异步发送成功: %s%n", sendResult);
        }

        @Override
        public void onException(Throwable e) {
            System.out.printf("异步发送失败: %s%n", e.getMessage());
        }
    });
}

// 单向发送 (只管发，不关心结果，适用于日志等)
public void sendOneWayOrder(Order order) {
    rocketMQTemplate.sendOneWay("order-topic:create", order);
}

// 延迟消息 (例如，10秒后处理)
public void sendDelayedOrder(Order order) {
    // 延迟等级: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    rocketMQTemplate.syncSend("order-topic:create", order, 3000, 3); // 3代表10s
}

// src/main/java/com/example/demo/controller/OrderController.java
package com.example.demo.controller;

import com.example.demo.model.Order;
import com.example.demo.service.OrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/orders")
public class OrderController {

    @Autowired
    private OrderService orderService;

    @PostMapping
    public String createOrder(@RequestBody Order order) {
        orderService.createOrder(order);
        return "订单创建请求已接收";
    }
}

// src/main/java/com/example/demo/consumer/OrderConsumer.java
package com.example.demo.consumer;

import com.example.demo.model.Order;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Service
@RocketMQMessageListener(
    topic = "order-topic",           // 监听的 Topic
    consumerGroup = "my-consumer-group", // 消费者组，必须唯一
    selectorExpression = "create"    // Tag 过滤，只消费 tag 为 "create" 的消息
)
public class OrderConsumer implements RocketMQListener<Order> {

    private static final Logger log = LoggerFactory.getLogger(OrderConsumer.class);

    @Override
    public void onMessage(Order order) {
        log.info("收到订单消息: {}", order);
        // 在这里处理业务逻辑，例如：
        // 1. 保存订单到数据库
        // 2. 发送邮件/短信通知
        // 3. 更新库存
        // ...
        System.out.println("已处理订单: " + order.getName());
    }
}

@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "my-consumer-group",
    selectorExpression = "*",
    consumeMode = ConsumeMode.CONCURRENTLY, // 并发消费 (默认)
    // consumeMode = ConsumeMode.ORDERLY,   // 顺序消费
    messageModel = MessageModel.CLUSTERING, // 集群模式 (默认)
    // messageModel = MessageModel.BROADCASTING, // 广播模式
    consumeThreadMax = 64, // 最大消费线程数
    consumeTimeout = 30000 // 消费超时时间 (ms)
)