🔄 RocketMQ 顺序消费者详解

@Profile({"prod"}) @Slf4j @Service @RocketMQMessageListener( consumeThreadMax = 10, consumeMode = ConsumeMode.ORDERLY, topic = "ORDER_TOPIC", // 与生产者Topic对应 consumerGroup = "ORDER_CONSUMER_GROUP" ) public class OrderlyMQConsumer implements RocketMQListener<MessageExt> { @Override public void onMessage(MessageExt messageExt) { try { // 获取生产者发送的完整消息信息 String topic = messageExt.getTopic(); // ORDER_TOPIC String tags = messageExt.getTags(); // TagA/TagB/TagC 或 null String keys = messageExt.getKeys(); // orderId_1001/orderId_1002 String msgId = messageExt.getMsgId(); int queueId = messageExt.getQueueId(); long queueOffset = messageExt.getQueueOffset(); // 获取消息体：订单创建/订单支付/订单完成 String messageBody = new String(messageExt.getBody(), StandardCharsets.UTF_8); log.info("顺序消费消息 - Topic: {}, Tags: {}, Keys: {}, QueueId: {}, QueueOffset: {}", topic, tags, keys, queueId, queueOffset); // 业务处理逻辑 processOrderlyMessage(messageBody, messageExt); log.info("顺序消费成功 - MsgId: {}", msgId); } catch (Exception e) { log.error("顺序消费失败 - MsgId: {}, 错误信息: {}", messageExt.getMsgId(), e.getMessage(), e); // 顺序消费失败时，当前队列会被暂停消费 throw new RuntimeException("顺序消息处理失败", e); } } private void processOrderlyMessage(String messageBody, MessageExt messageExt) { String orderKey = messageExt.getKeys(); // orderId_1001 String tag = messageExt.getTags(); // TagA/TagB/TagC 或 null log.info("处理订单消息 - OrderKey: {}, Tag: {}, Content: {}", orderKey, tag, messageBody); // 根据消息内容处理不同业务状态 switch (messageBody) { case "订单创建": handleOrderCreate(orderKey, tag); break; case "订单支付": handleOrderPay(orderKey, tag); break; case "订单完成": handleOrderComplete(orderKey, tag); break; } } }