RocketMQ如何保证消息被有序消费
消费者端如何接收有序消息
队列消费的两种模式
并发消费模式
当同一类消息被送入不同队列，且这些消息在处理上并不需要按时序消费时，可以考虑使用并发消费模式。
并发消费模式生产者会将消息轮询发送到不同的队列当中，这些队列会和消费者实例建立多个连接（线程），将消息并发送入到不同的消费者，因为消费者处理速度有快慢，所以并不能保证物流数据会按1~9的顺序依次消费。
并发消费模式处理效率很高，但无法保证有序性。
有序消费模式
有序消息是指生产者在产生数据的时候，根据Hash规则指定让消息放入哪个队列，在消费者消费时会保证不同消费者针对每一个队列只有唯一的连接（线程）用于消费指定队列。
有序消费模式可以保证消息按队列FIFO顺序依次被消费，但因此失去并发性能，有序消费模式只有在业务要求必须按顺序消费的场景下才允许使用。
RocketMQ如何实现有序消息
要实现RocketMQ有序消息需要两点调整：
- 生产者端要求按id等唯一标识分配消息队列
- 消费者端采用专用的监听器保证对队列的单线程应用
下面咱们来看一下代码：
生产者端
SequenceMessageProvider核心代码是在向Broker发送消息时附加MessageQueueSelector对象，在实现select方法时指定存放到哪个队列中。
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import java.nio.charset.StandardCharsets;
import java.util.List;
@Slf4j
public class SequenceMessageProvider {
public static void main(String[] args) {
// 前置准备代码
DefaultMQProducer producer = new DefaultMQProducer("producer-group");
producer.setNamesrvAddr("127.0.0.1:9876");
try {
producer.start();
// 模拟10笔订单
for (Integer orderId = 1; orderId <= 10; orderId++) {
// 每笔订单要发3条消息：（1）创建订单 （2）订单库存扣减 （3）增加积分
for (int i = 0; i < 3; i++) {
String data = "";
switch (i % 3) {
case 0:
data = orderId + "号创建订单";
break;
case 1:
data = orderId + "号订单减少库存";
break;
case 2:
data = orderId + "号订单增加积分";
break;
}
// 创建消息对象 topic="order",tags="order",key=orderId
Message message = new Message("order", "order", orderId.toString(), data.getBytes(StandardCharsets.UTF_8));
// 发送消息，实现MessageQueueSelector接口
SendResult result = producer.send(message, new MessageQueueSelector() {
// select方法决定向broker哪一个队列发送消息
@Override
public MessageQueue select(List<MessageQueue> list, Message msg, Object arg) {
int orderId = Integer.parseInt(msg.getKeys());
int index = orderId % list.size();
MessageQueue messageQueue = list.get(index);
log.info("id:{}，data:{}，queue:{}", orderId, new String(msg.getBody()), messageQueue);
return messageQueue;
}
}, null);
}
}
} catch (Exception exception) {
exception.printStackTrace();
} finally {
try {
producer.shutdown();
log.warn("连接已关闭");
} catch (Exception exception) {
exception.printStackTrace();
}
}
}
}
消费者端
消费者端最大的变化是registerMessageListener监听器要实例化MessageListenerOrdery对象，用于为每一个队列分配唯一的连接（线程）进行消费。
每一批消息从Broker投递给消费者都会触发consumeMessage()方法实现对消息的消费。
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
@Slf4j
public class SequenceMessageConsumer {
public static void main(String[] args) throws Exception {
// 声明并初始化一个consumer
// 需要一个consumer group名字作为构造方法的参数
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
// 同样也要设置NamesrvAddr地址，须要与提供者的地址列表保持一致
consumer.setNamesrvAddr("127.0.0.1:9876");
// 设置consumer所订阅的Top 和 Tag，*代表全部的Tag
consumer.subscribe("order", "*");
// 注册消息监听者，消费者端要增加MessageListenerOrderly监听器，用于实现有序队列
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext context) {
// 遍历输出
list.forEach(msg -> {
log.info("{}，{}，{}", msg.getKeys(), new String(msg.getBody()), context.getMessageQueue());
});
return ConsumeOrderlyStatus.SUCCESS;
}
});
// 启动消费者
consumer.start();
}
}
如何实现消息全局顺序消费？
只需要再生产者固定将所有消息发往到0号队列即可保证全局有序，这也意味着全局采用单线程消费，执行效率极差。
@Override
public MessageQueue select(List<MessageQueue> list, Message msg, Object arg) {
MessageQueue messageQueue = list.get(0);
return messageQueue;
}
有序消费有什么使用限制吗？
有序消费模式只支持集群模式（CLUSTERING），不支持广播模式（BROADCASTING），采用广播模式会无法接收到数据。
// 设置为集群模式
consumer.setMessageModel(MessageModel.CLUSTERING); //支持有序消息，默认模式
consumer.setMessageModel(MessageModel.BROADCASTING); //不支持有序消息
