消息队列深度实践

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

┌──────────┐    1.send(topic, msg)    ┌──────────┐
│ Producer │ ──────────────────────▶ │  Broker  │
│ (订单服务)│                          │  Node-1  │
└──────────┘                          │ ┌──────┐ │
                                      │ │TopicA│ │
┌──────────┐    4.pull(topic, queue)  │ │Q0 Q1 │ │
│ Consumer │ ◀────────────────────── │ │Q2 Q3 │ │
│(库存服务) │                          │ └──────┘ │
└──────────┘                          └──────────┘
                                      ┌──────────┐
                  2.register ◀───────  │NameServer│
                  3.update ────────▶  │  集群    │
                                      └──────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

  Producer 写入路径
  ────────────────
  msg1 ─┐
  msg2 ─┤
  msg3 ─┼──▶ Broker ──▶ CommitLog(顺序写,append only)
  msg4 ─┤                 (1G/片,文件名=起始偏移)
  msg5 ─┘                       │
                                │ 每条 msg 落盘时异步构建
                                ▼
                        ┌─────────────────┐
                        │  ConsumeQueue    │ ◀── Consumer 拉取入口
                        │  (TopicA/Queue0) │     (类似"目录")
                        │  offset/size/tag │
                        └─────────────────┘
                                │
                                │ 按 msgKey 反查时
                                ▼
                        ┌─────────────────┐
                        │   IndexFile     │ ◀── 按 key 查询
                        │  (Hash 500w槽)  │
                        └─────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

spring:
  application:
    name: chat-service
  cloud:
    stream:
      rocketmq:
        binder:
          name-server: rocketmq-nameserver:9876
          access-key: ${ROCKETMQ_AK}
          secret-key: ${ROCKETMQ_SK}
          vip-channel-enabled: false
        bindings:
          output-chat-message:
            producer:
              group: chat-message-producer
              sync: true
              delivery-timeout: 5000
              max-message-size: 4194304
              message-orderly: false
              transactional-message-enabled: true
          input-chat-message:
            consumer:
              group: chat-message-consumer
              orderly: false
              max-attempts: 3
              pull-batch-size: 32
              push-orderly: false
      bindings:
        output-chat-message:
          destination: CHAT_MESSAGE_TOPIC
          content-type: application/json
        input-chat-message:
          destination: CHAT_MESSAGE_TOPIC
          group: chat-message-consumer
          consumer:
            concurrency: 8
            max-attempts: 3

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

package com.saas.chat.mq;

import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.apache.rocketmq.client.producer.LocalTransactionChecker;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessagingException;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@Slf4j
@Component
public class ChatMessageProducer {

    @Autowired
    private StreamBridge streamBridge;

    /**
     * 普通消息发送:同步等待 Broker ACK,3 秒内未确认抛异常走重试
     */
    public boolean sendNormal(ChatMessageDTO msg) {
        Message<ChatMessageDTO> message = MessageBuilder
                .withPayload(msg)
                .setHeader("msgId", msg.getMsgId())
                .setHeader("tenantId", msg.getTenantId())
                .build();
        try {
            streamBridge.send("output-chat-message", message);
            return true;
        } catch (MessagingException e) {
            log.error("sendNormal failed, msgId={}", msg.getMsgId(), e);
            return false;
        }
    }

    /**
     * 事务消息发送:先发 half 消息,本地事务执行后 commit/rollback
     * RocketMQ Broker 会在 unknown 状态回调 checkLocalTransactionState
     */
    public boolean sendTransactional(ChatMessageDTO msg, LocalBizExecutor executor) {
        Map<String, Object> headers = new HashMap<>();
        headers.put("msgId", msg.getMsgId());
        headers.put(MessageConst.PROPERTY_TRANSACTION_KEY, msg.getMsgId());
        Message<ChatMessageDTO> message = MessageBuilder
                .withPayload(msg)
                .copyHeaders(headers)
                .build();
        try {
            streamBridge.send("output-tx-chat-message", message, executor);
            return true;
        } catch (MessagingException e) {
            log.error("sendTransactional failed, msgId={}", msg.getMsgId(), e);
            return false;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

package com.saas.chat.mq.tx;

import org.apache.rocketmq.client.producer.LocalTransactionChecker;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;

@Slf4j
@Component
@RocketMQTransactionListener
public class ChatTxMessageChecker implements RocketMQLocalTransactionListener {

    @Resource
    private ChatMessageTxLogService txLogService;

    /**
     * 本地事务执行器
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        String msgId = (String) message.getHeaders().get("msgId");
        try {
            LocalBizExecutor executor = (LocalBizExecutor) arg;
            boolean ok = executor.doBiz();
            if (ok) {
                txLogService.markSuccess(msgId);
                return RocketMQLocalTransactionState.COMMIT;
            } else {
                txLogService.markRollback(msgId);
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e) {
            log.error("executeLocalTransaction failed, msgId={}", msgId, e);
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    /**
     * Broker 回调:消息状态 UNKNOWN 时触发,反查本地事务表
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        String msgId = messageExt.getProperty("msgId");
        boolean exists = txLogService.checkExists(msgId);
        if (exists) {
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

package com.saas.chat.mq.consumer;

import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.beans.factory.annotation.Autowired;
import com.saas.chat.service.ChatService;
import com.saas.chat.entity.ChatMessageDTO;
import com.saas.chat.mq.ack.AckManualHelper;
import lombok.extern.slf4j.Slf4j;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.common.message.MessageConst;

@Slf4j
@Component
public class ChatMessageConsumer {

    @Autowired
    private ChatService chatService;

    @Autowired
    private IdempotentChecker idempotentChecker;

    @Autowired
    private AckManualHelper ackHelper;

    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 接收消息 + 幂等校验 + 手动 ACK
     * ackMode=MANUAL 时必须显式调用 ackHelper.ack()
     */
    @StreamListener(value = Sink.INPUT, condition = "headers['rocketmq_flag']=='chat_msg'")
    public void onMessage(@Payload ChatMessageDTO msg,
                          org.springframework.messaging.MessageHeaders headers) {
        String msgId = (String) headers.get("msgId");
        String tenantId = (String) headers.get("tenantId");
        try {
            // 1. 幂等校验:已处理过直接 ACK 跳过
            if (idempotentChecker.isProcessed(msgId)) {
                log.debug("msg already processed, skip. msgId={}", msgId);
                ackHelper.ack();
                return;
            }
            // 2. 业务处理
            chatService.handleMessage(msg, tenantId);
            // 3. 写入幂等表(用 msgId 主键,带过期时间 7 天)
            idempotentChecker.markProcessed(msgId, 7 * 24 * 3600);
            // 4. 手动 ACK
            ackHelper.ack();
        } catch (Exception e) {
            log.error("onMessage failed, msgId={}", msgId, e);
            // 5. 异常:不 ACK,Broker 会在 30s 后重新投递
            //    max-attempts=3 控制最大重试次数,超 3 次进死信队列
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

package com.saas.chat.mq.partition;

import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class ConversationPartitioner implements PartitionSelectorStrategy {

    /**
     * 根据会话 ID 算分区号,保证同一会话的消息进同一 Queue
     * partitionCount 必须和 Topic 的 Queue 数一致
     */
    @Override
    public int selectPartition(Message<?> message) {
        Object convId = message.getHeaders().get("conversationId");
        if (convId == null) {
            return 0;
        }
        int hash = Math.abs(convId.hashCode());
        int partitionCount = 8;  // 与 Broker 配置 readQueueNums=8 一致
        return hash % partitionCount;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

spring:
  cloud:
    stream:
      bindings:
        output-orderly-message:
          destination: CHAT_ORDERLY_TOPIC
          partitioned: true
          partition-key-expression: headers['conversationId']
          producer:
            partition-count: 8
        input-orderly-message:
          destination: CHAT_ORDERLY_TOPIC
          group: chat-orderly-consumer
          consumer:
            partitioned: true
            concurrency: 8

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

apiVersion: v1
kind: Namespace
metadata:
  name: rocketmq
---
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: rocketmq-ssd
provisioner: diskplugin.csi.alibabacloud.com
parameters:
  type: cloud_essd
  performanceLevel: PL1
  fsType: ext4
reclaimPolicy: Retain
volumeBindingMode: WaitForFirstConsumer
allowVolumeExpansion: true

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rocketmq-nameserver
  namespace: rocketmq
spec:
  replicas: 2
  selector:
    matchLabels: { app: rocketmq-nameserver }
  template:
    metadata:
      labels: { app: rocketmq-nameserver }
    spec:
      containers:
      - name: nameserver
        image: apache/rocketmq:5.3.0
        command: ["sh", "mqnamesrv"]
        ports:
        - containerPort: 9876
        env:
        - name: JAVA_OPT_EXT
          value: "-Xms2g -Xmx2g -Xmn1g"
        resources:
          requests: { cpu: "500m", memory: "2Gi" }
          limits:   { cpu: "2",    memory: "4Gi" }
        livenessProbe:
          tcpSocket: { port: 9876 }
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          tcpSocket: { port: 9876 }
          initialDelaySeconds: 15
          periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: rocketmq-nameserver
  namespace: rocketmq
spec:
  selector: { app: rocketmq-nameserver }
  ports:
  - port: 9876
    targetPort: 9876
  type: ClusterIP

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: rocketmq-broker
  namespace: rocketmq
spec:
  serviceName: rocketmq-broker-headless
  replicas: 3
  selector:
    matchLabels: { app: rocketmq-broker }
  template:
    metadata:
      labels: { app: rocketmq-broker }
    spec:
      containers:
      - name: broker
        image: apache/rocketmq:5.3.0
        command: ["sh", "mqbroker", "-c", "/etc/rocketmq/broker.conf"]
        ports:
        - { containerPort: 10909, name: vip }
        - { containerPort: 10911, name: broker }
        - { containerPort: 10912, name: ha }
        env:
        - name: NAMESRV_ADDR
          value: "rocketmq-nameserver:9876"
        - name: JAVA_OPT_EXT
          value: "-Xms8g -Xmx8g -Xmn4g -XX:+UseG1GC"
        resources:
          requests: { cpu: "2", memory: "10Gi" }
          limits:   { cpu: "8", memory: "16Gi" }
        volumeMounts:
        - { name: store, mountPath: /home/rocketmq/store }
        - { name: conf,  mountPath: /etc/rocketmq }
        livenessProbe:
          tcpSocket: { port: 10911 }
          initialDelaySeconds: 90
          periodSeconds: 15
          failureThreshold: 4
        readinessProbe:
          tcpSocket: { port: 10911 }
          initialDelaySeconds: 60
          periodSeconds: 10
      volumes:
      - name: conf
        configMap: { name: rocketmq-broker-conf }
  volumeClaimTemplates:
  - metadata: { name: store }
    spec:
      accessModes: [ "ReadWriteOnce" ]
      storageClassName: rocketmq-ssd
      resources:
        requests: { storage: 100Gi }

1
2
3
4
5
6
7
8

brokerClusterName=DefaultCluster
brokerName=broker-dledger-1
enableDLegerCommitLog=true
dLegerGroup=broker-dledger-1
dLegerPeers=n0-rocketmq-broker-0.rocketmq-broker-headless:40911;n1-rocketmq-broker-1.rocketmq-broker-headless:40911;n2-rocketmq-broker-2.rocketmq-broker-headless:40911
dLegerSelfId=n0
storePathRootDir=/home/rocketmq/store
flushDiskType=SYNC_FLUSH

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rocketmq-proxy
  namespace: rocketmq
spec:
  replicas: 2
  selector:
    matchLabels: { app: rocketmq-proxy }
  template:
    metadata:
      labels: { app: rocketmq-proxy }
    spec:
      containers:
      - name: proxy
        image: apache/rocketmq:5.3.0
        command: ["sh", "mqproxy", "-n", "rocketmq-nameserver:9876"]
        ports:
        - { containerPort: 8080, name: remoting }
        - { containerPort: 8081, name: grpc }
        resources:
          requests: { cpu: "1", memory: "4Gi" }
          limits:   { cpu: "4", memory: "8Gi" }
        livenessProbe:
          tcpSocket: { port: 8081 }
          initialDelaySeconds: 45
          periodSeconds: 10
        readinessProbe:
          tcpSocket: { port: 8081 }
          initialDelaySeconds: 30
          periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: rocketmq-proxy
  namespace: rocketmq
spec:
  selector: { app: rocketmq-proxy }
  ports:
  - { name: remoting, port: 8080 }
  - { name: grpc,     port: 8081 }
  type: ClusterIP

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: rocketmq-controller
  namespace: rocketmq
spec:
  serviceName: rocketmq-controller-headless
  replicas: 3
  selector:
    matchLabels: { app: rocketmq-controller }
  template:
    metadata:
      labels: { app: rocketmq-controller }
    spec:
      containers:
      - name: controller
        image: apache/rocketmq:5.3.0
        command: ["sh", "mqcontroller", "-c", "/etc/rocketmq/controller.conf"]
        ports:
        - { containerPort: 9878 }
        resources:
          requests: { cpu: "500m", memory: "2Gi" }
          limits:   { cpu: "2",    memory: "4Gi" }
        livenessProbe:
          tcpSocket: { port: 9878 }
          initialDelaySeconds: 30
          periodSeconds: 10
        volumeMounts:
        - { name: data, mountPath: /home/rocketmq/controller }
  volumeClaimTemplates:
  - metadata: { name: data }
    spec:
      accessModes: [ "ReadWriteOnce" ]
      storageClassName: rocketmq-ssd
      resources: { requests: { storage: 10Gi } }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 1. 所有 Pod Running 且 Ready
kubectl -n rocketmq get pod -o wide
# 预期:nameserver 2/2、broker 3/3、proxy 2/2、controller 3/3 均 1/1 Running

# 2. 集群拓扑列表(Master/Slave 角色 + Dledger 主从)
kubectl -n rocketmq exec rocketmq-broker-0 -- sh mqadmin clusterList -n rocketmq-nameserver:9876
# 预期:看到 3 个 Master,BID=0,InSyncReplicas=3

# 3. 创建测试 Topic 并验证消息收发
kubectl -n rocketmq exec rocketmq-broker-0 -- sh mqadmin updateTopic \
  -n rocketmq-nameserver:9876 -c DefaultCluster -t TestTopic -r 8 -w 8
# 预期:8 读 8 写队列分配到 3 个 broker,无 ERROR 返回

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# broker.conf 核心配置
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
# 同步刷盘:消息写入 CommitLog 后才返回 Producer 成功
flushDiskType=SYNC_FLUSH
# 同步双写:Master 等 Slave 同步刷盘成功才返回成功
brokerRole=SYNC_MASTER
# 主从异步复制会让 Slave 落后 Master 50-200ms
# 同步复制 SLA 99.99%,但吞吐下降 30%

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// Redis 去重表 + Lua 原子写入
public class RedisIdempotentChecker {
    private static final String LUA_SCRIPT =
        "if redis.call('SETNX', KEYS[1], '1') then " +
        "  return redis.call('EXPIRE', KEYS[1], 86400) " +
        "else return 0 end";

    public boolean isProcessed(String msgId) {
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(LUA_SCRIPT, Long.class),
            List.of("mq:processed:" + msgId)
        );
        return result == null || result == 0;
    }
}

1
2
3
4
5
6
7
8
9

spring:
  cloud:
    stream:
      rocketmq:
        bindings:
          output:
            producer:
              # 同一 orderId 的消息路由到同一 Queue
              messageQueueSelector: orderIdHashSelector

1
2
3
4

# K8s 快速扩容(平时 3 个副本,临时到 12 个)
kubectl scale deployment notification-consumer --replicas=12
# 队列数锁死了并行度上限,扩到队列数=4 时已无意义
# 4 个队列,扩 4 个以上实例不会提升吞吐

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

spring:
  cloud:
    stream:
      rocketmq:
        bindings:
          input:
            consumer:
              # 默认单条拉取,改批量
              pullBatchSize: 1000
              consumeMessageBatchMaxSize: 1000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// 第一阶段:发送半消息(Half Message),对消费者不可见
rocketMQTemplate.sendMessageInTransaction(
    "order-tx-topic",
    MessageBuilder.withPayload(order).build(),
    order  // 业务参数,传递给本地事务检查器
);

// 本地事务检查器 - 第二阶段
@RocketMQTransactionListener
public class OrderTxListener implements RocketMQLocalTransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            orderService.createOrder((Order) arg);
            return LocalTransactionState.COMMIT_MESSAGE;  // 提交
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;  // 回滚
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // MQ Broker 端反查:60s 还没收到 COMMIT/ROLLBACK 就回调这里
        Order order = orderService.findById(msg.getKeys());
        return order != null ? COMMIT_MESSAGE : ROLLBACK_MESSAGE;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

rocketMQTemplate.syncSendOrderly(
    "order-status-topic",
    MessageBuilder.withPayload(status).build(),
    orderId,  // hashKey,相同 orderId 进同一队列
    3000      // 超时
);

// 或用 MessageQueueSelector 自己控制
rocketMQTemplate.getProducer().send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> queues, Message msg, Object arg) {
        long orderId = (long) arg;
        int idx = (int) (orderId % queues.size());
        return queues.get(idx);
    }
}, orderId);

1
2
3
4
5

Message msg = MessageBuilder.withPayload(order)
    .setHeader("KEYS", orderId)
    .build();
msg.setDelayTimeLevel(3);  // 10 秒
rocketMQTemplate.getProducer().send(msg);

1
2
3
4
5
6

// 替代旧的 reconsumeLater,直接拉历史消息
List<MessageExt> messages = popConsumer.reconsumeFromTimestamp(
    "order-topic",
    startTimestamp,  // 3 天前的时间戳
    1000             // 最多 1000 条
);