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

1
import lombok.extern.slf4j.Slf4j;
2
import org.apache.rocketmq.client.producer.SendCallback;
3
import org.apache.rocketmq.client.producer.SendResult;
4
import org.apache.rocketmq.spring.core.RocketMQTemplate;
5
import org.springframework.messaging.Message;
6
import org.springframework.messaging.support.MessageBuilder;
7
import org.springframework.stereotype.Service;
8

9
import jakarta.annotation.Resource;
10

11
@Service
12
@Slf4j
13
public class RocketMQProducer<T> implements IMQProducer<T> {
14

15
    private final String LOG_SIGN = "生产者消息";
16

17
    @Resource
18
    private RocketMQTemplate rocketMQTemplate;
19

20
    @Override
21
    public void asyncSend(String topic, T msg, int delayLevel) {
22
        try {
23
            Message<T> message = MessageBuilder.withPayload(msg).build();
24
            rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
25
                @Override
26
                public void onSuccess(SendResult sendResult) {
27
                    log.info("【{}】【{}】发送成功, 状态码【{}】", LOG_SIGN, message, sendResult.getSendStatus());
28
                }
29

30
                @Override
31
                public void onException(Throwable throwable) {
32
                    log.warn("【{}】【{}】发送异常", LOG_SIGN, message, throwable);
33
                }
34
            }, 10000, delayLevel);
35
        } catch (Exception e) {
36
            log.error("【{}】【{}】发送异常", LOG_SIGN, msg, e);
37
        }
38
    }
39

40
    @Override
41
    public SendResult syncSend(String topic, T msg, int delayLevel) {
42
        try {
43
            Message<T> message = MessageBuilder.withPayload(msg).build();
44
            SendResult sendResult = rocketMQTemplate.syncSend(topic, message, 10000, delayLevel);
45
            log.info("【{}】【{}】发送结果 状态码【{}】 MsgId【{}】",
46
                    LOG_SIGN, msg, sendResult.getSendStatus(), sendResult.getMsgId());
47
            return sendResult;
48
        } catch (Exception e) {
49
            log.error("【{}】【{}】发送异常", LOG_SIGN, msg, e);
50
        }
51
        return null;
52
    }
53

54
    @Override
55
    public SendResult syncSend(String topic, T msg) {
56
        try {
57
            Message<T> message = MessageBuilder.withPayload(msg).build();
58
            SendResult sendResult = rocketMQTemplate.syncSend(topic, message, 10000);
59
            log.info("【{}】【{}】发送结果 状态码【{}】 MsgId【{}】",
60
                    LOG_SIGN, msg, sendResult.getSendStatus(), sendResult.getMsgId());
61
            return sendResult;
62
        } catch (Exception e) {
63
            log.error("【{}】【{}】发送异常", LOG_SIGN, msg, e);
64
        }
65
        return null;
66
    }
67

68
    @Override
69
    public SendResult syncSendDeliverTimeMills(String topic, T msg, long pushTime) {
70
        try {
71
            Message<T> message = MessageBuilder.withPayload(msg).build();
72
            SendResult sendResult = rocketMQTemplate.syncSendDeliverTimeMills(topic, message, pushTime);
73
            log.info("【{}】【{}】发送定时消息结果 状态码【{}】 MsgId【{}】",
74
                    LOG_SIGN, msg, sendResult.getSendStatus(), sendResult.getMsgId());
75
            return sendResult;
76
        } catch (Exception e) {
77
            log.error("【{}】【{}】发送定时消息异常", LOG_SIGN, msg, e);
78
        }
79
        return null;
80
    }
81

82
    @Override
83
    public SendResult syncSendOrderly(String topic, T msg, String hashKey) {
84
        try {
85
            Message<T> message = MessageBuilder.withPayload(msg).build();
86
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, message, hashKey);
87
            log.info("【{}】【{}】发送结果 状态码【{}】 MsgId【{}】",
88
                    LOG_SIGN, msg, sendResult.getSendStatus(), sendResult.getMsgId());
89
            return sendResult;
90
        } catch (Exception e) {
91
            log.error("【{}】【{}】发送异常", LOG_SIGN, msg, e);
92
        }
93
        return null;
94
    }
95
}

1
// 发送延迟5秒的消息（delayLevel = 2）
2
mqProducer.syncSend("TOPIC", "message", 2);

1
import lombok.extern.slf4j.Slf4j;
2
import org.apache.rocketmq.common.message.MessageExt;
3
import org.apache.rocketmq.spring.annotation.ConsumeMode;
4
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
5
import org.apache.rocketmq.spring.core.RocketMQListener;
6
import org.springframework.context.annotation.Profile;
7
import org.springframework.stereotype.Service;
8

9
import java.nio.charset.StandardCharsets;
10

11
/**
12
 * 并发消费者 - 高吞吐量场景
13
 * 适用于: 日志处理、数据统计、通知发送等对顺序不敏感的场景
14
 */
15
@Profile({"prod"})
16
@Slf4j
17
@Service
18
@RocketMQMessageListener(
19
    consumeThreadMax = 10,
20
    consumeMode = ConsumeMode.CONCURRENTLY,
21
    topic = "CONCURRENT_TOPIC",
22
    consumerGroup = "CONCURRENT_CONSUMER_GROUP"
23
)
24
public class ConcurrentMQConsumer implements RocketMQListener<MessageExt> {
25

26
    @Override
27
    public void onMessage(MessageExt messageExt) {
28
        try {
29
            // 获取消息基本信息
30
            String topic = messageExt.getTopic();
31
            String tags = messageExt.getTags();
32
            String keys = messageExt.getKeys();
33
            String msgId = messageExt.getMsgId();
34
            int reconsumeTimes = messageExt.getReconsumeTimes();
35

36
            // 获取消息体
37
            String messageBody = new String(messageExt.getBody(), StandardCharsets.UTF_8);
38

39
            log.info("并发消费消息 - Topic: {}, Tags: {}, Keys: {}, MsgId: {}, ReconsumeTimes: {}",
40
                    topic, tags, keys, msgId, reconsumeTimes);
41

42
            // 业务处理逻辑
43
            processMessage(messageBody, messageExt);
44

45
            log.info("并发消费成功 - MsgId: {}", msgId);
46

47
        } catch (Exception e) {
48
            log.error("并发消费失败 - MsgId: {}, 错误信息: {}",
49
                    messageExt.getMsgId(), e.getMessage(), e);
50

51
            // 抛出异常会触发重试机制
52
            throw new RuntimeException("消息处理失败", e);
53
        }
54
    }
55

56
    /**
57
     * 业务消息处理
58
     */
59
    private void processMessage(String messageBody, MessageExt messageExt) {
60
        // 根据消息内容进行业务处理
61
        // 例如：数据入库、调用外部API、发送通知等
62
        log.info("处理消息内容: {}", messageBody);
63

64
        // 具体业务逻辑
65
        // ...
66
    }
67
}

1
import lombok.extern.slf4j.Slf4j;
2
import org.apache.rocketmq.common.message.MessageExt;
3
import org.apache.rocketmq.spring.annotation.ConsumeMode;
4
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
5
import org.apache.rocketmq.spring.core.RocketMQListener;
6
import org.springframework.context.annotation.Profile;
7
import org.springframework.stereotype.Service;
8

9
import java.nio.charset.StandardCharsets;
10

11
/**
12
 * 顺序消费者 - 严格顺序场景
13
 * 适用于: 订单状态变更、账户余额变动、状态机流转等需要严格顺序的场景
14
 */
15
@Profile({"prod"})
16
@Slf4j
17
@Service
18
@RocketMQMessageListener(
19
    consumeThreadMax = 10,
20
    consumeMode = ConsumeMode.ORDERLY,
21
    topic = "ORDERLY_TOPIC",
22
    consumerGroup = "ORDERLY_CONSUMER_GROUP"
23
)
24
public class OrderlyMQConsumer implements RocketMQListener<MessageExt> {
25

26
    @Override
27
    public void onMessage(MessageExt messageExt) {
28
        try {
29
            // 获取消息基本信息
30
            String topic = messageExt.getTopic();
31
            String tags = messageExt.getTags();
32
            String keys = messageExt.getKeys();
33
            String msgId = messageExt.getMsgId();
34
            int queueId = messageExt.getQueueId();
35
            long queueOffset = messageExt.getQueueOffset();
36

37
            // 获取消息体
38
            String messageBody = new String(messageExt.getBody(), StandardCharsets.UTF_8);
39

40
            log.info("顺序消费消息 - Topic: {}, Tags: {}, Keys: {}, MsgId: {}, QueueId: {}, QueueOffset: {}",
41
                    topic, tags, keys, msgId, queueId, queueOffset);
42

43
            // 业务处理逻辑
44
            processOrderlyMessage(messageBody, messageExt);
45

46
            log.info("顺序消费成功 - MsgId: {}, QueueId: {}, QueueOffset: {}",
47
                    msgId, queueId, queueOffset);
48

49
        } catch (Exception e) {
50
            log.error("顺序消费失败 - MsgId: {}, QueueId: {}, 错误信息: {}",
51
                    messageExt.getMsgId(), messageExt.getQueueId(), e.getMessage(), e);
52

53
            // 顺序消费失败时，当前队列会被暂停消费
54
            throw new RuntimeException("顺序消息处理失败", e);
55
        }
56
    }
57

58
    /**
59
     * 顺序业务消息处理
60
     */
61
    private void processOrderlyMessage(String messageBody, MessageExt messageExt) {
62
        // 顺序消费的业务处理
63
        // 例如：订单状态流转、账户变动等需要严格顺序的操作
64

65
        String orderKey = messageExt.getKeys();
66
        log.info("处理订单消息 - OrderKey: {}, Content: {}", orderKey, messageBody);
67

68
        // 模拟状态机处理
69
        processOrderStatus(orderKey, messageBody);
70
    }
71

72
    /**
73
     * 订单状态处理示例
74
     */
75
    private void processOrderStatus(String orderKey, String messageBody) {
76
        // 解析消息内容，处理订单状态变更
77
        // 保证同一订单的消息按顺序处理
78
        log.info("订单状态更新完成 - OrderKey: {}", orderKey);
79

80
        // 具体业务逻辑
81
        // ...
82
    }
83
}

1
import lombok.extern.slf4j.Slf4j;
2
import org.apache.rocketmq.common.message.MessageExt;
3
import org.apache.rocketmq.spring.annotation.ConsumeMode;
4
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
5
import org.apache.rocketmq.spring.annotation.SelectorType;
6
import org.apache.rocketmq.spring.core.RocketMQListener;
7
import org.springframework.context.annotation.Profile;
8
import org.springframework.stereotype.Service;
9

10
import java.nio.charset.StandardCharsets;
11

12
/**
13
 * 带TAG过滤的消费者
14
 * 只消费指定TAG的消息
15
 */
16
@Profile({"prod"})
17
@Slf4j
18
@Service
19
@RocketMQMessageListener(
20
    consumeThreadMax = 10,
21
    consumeMode = ConsumeMode.CONCURRENTLY,
22
    topic = "USER_EVENT_TOPIC",
23
    consumerGroup = "USER_EVENT_CONSUMER_GROUP",
24
    selectorType = SelectorType.TAG,
25
    selectorExpression = "LOGIN || LOGOUT || REGISTER"
26
)
27
public class UserEventMQConsumer implements RocketMQListener<MessageExt> {
28

29
    @Override
30
    public void onMessage(MessageExt messageExt) {
31
        try {
32
            String topic = messageExt.getTopic();
33
            String tags = messageExt.getTags();
34
            String msgId = messageExt.getMsgId();
35
            String messageBody = new String(messageExt.getBody(), StandardCharsets.UTF_8);
36

37
            log.info("用户事件消费 - Topic: {}, Tags: {}, MsgId: {}", topic, tags, msgId);
38

39
            // 根据不同TAG处理不同业务
40
            switch (tags) {
41
                case "LOGIN":
42
                    handleUserLogin(messageBody);
43
                    break;
44
                case "LOGOUT":
45
                    handleUserLogout(messageBody);
46
                    break;
47
                case "REGISTER":
48
                    handleUserRegister(messageBody);
49
                    break;
50
                default:
51
                    log.warn("未知的用户事件类型: {}", tags);
52
            }
53

54
            log.info("用户事件处理成功 - MsgId: {}, Tags: {}", msgId, tags);
55

56
        } catch (Exception e) {
57
            log.error("用户事件处理失败 - MsgId: {}, 错误信息: {}",
58
                    messageExt.getMsgId(), e.getMessage(), e);
59
            throw new RuntimeException("用户事件处理失败", e);
60
        }
61
    }
62

63
    private void handleUserLogin(String messageBody) {
64
        log.info("处理用户登录事件: {}", messageBody);
65
        // 用户登录相关业务处理
66
    }
67

68
    private void handleUserLogout(String messageBody) {
69
        log.info("处理用户登出事件: {}", messageBody);
70
        // 用户登出相关业务处理
71
    }
72

73
    private void handleUserRegister(String messageBody) {
74
        log.info("处理用户注册事件: {}", messageBody);
75
        // 用户注册相关业务处理
76
    }
77
}

1
@Service
2
public class MessageService {
3

4
    @Autowired
5
    private IMQProducer<String> mqProducer;
6

7
    public void sendMessage() {
8
        // 同步发送普通消息
9
        SendResult result = mqProducer.syncSend("ORDER_TOPIC", "订单创建消息");
10

11
        // 异步发送延迟消息
12
        mqProducer.asyncSend("ORDER_TOPIC", "延迟处理消息", 3);
13

14
        // 发送定时消息（1小时后执行）
15
        long deliverTime = System.currentTimeMillis() + 3600 * 1000;
16
        mqProducer.syncSendDeliverTimeMills("ORDER_TOPIC", "定时处理消息", deliverTime);
17

18
        // 发送有序消息
19
        mqProducer.syncSendOrderly("ORDER_TOPIC", "订单状态变更", "orderKey123");
20

21
        // 发送带TAG的消息
22
        mqProducer.syncSend("ORDER_TOPIC:CREATE", "订单创建消息");
23
    }
24
}

1
// 并发消费示例
2
@Profile({"prod"})
3
@Slf4j
4
@Service
5
@RocketMQMessageListener(
6
    consumeThreadMax = 10,
7
    consumeMode = ConsumeMode.CONCURRENTLY,
8
    topic = "ORDER_TOPIC",
9
    consumerGroup = "ORDER_CONSUMER_GROUP"
10
)
11
public class OrderMQConsumer implements RocketMQListener<MessageExt> {
12

13
    @Override
14
    public void onMessage(MessageExt messageExt) {
15
        String messageBody = new String(messageExt.getBody(), StandardCharsets.UTF_8);
16
        log.info("处理订单消息: {}", messageBody);
17
        // 业务处理逻辑
18
    }
19
}