全局有序
在RocketMQ中，如果使消息全局有序，可以为Topic设置一个消息队列，使用一个生产者单线程发送数据，消费者端也使用单线程进行消费，从而保证消息的全局有序，但是这种方式效率低，一般不使用。
局部有序
假设一个Topic分配了两个消息队列，生产者在发送消息的时候，可以对消息设置一个路由ID，比如想保证一个订单的相关消息有序，那么就使用订单ID当做路由ID，在发送消息的时候，通过订单ID对消息队列的个数取余，根据取余结果选择消息队列，这样同一个订单的数据就可以保证发送到一个消息队列中，消费者端使用MessageListenerOrderly处理有序消息，这就是RocketMQ的局部有序，保证消息在某个消息队列中有序。
接下来看RoceketMQ源码中提供的顺序消息例子（稍微做了一些修改）：
生产者
public class Producer {
public static void main(String[] args) throws UnsupportedEncodingException {
try {
// 创建生产者
DefaultMQProducer producer = new DefaultMQProducer("生产者组");
// 启动
producer.start();
// 创建TAG
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 100; i++) {
// 生成订单ID
int orderId = i % 10;
// 创建消息
Message msg =
new Message("TopicTest", tags[i % tags.length], "KEY" + i,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
// 获取订单ID
Integer id = (Integer) arg;
// 对消息队列个数取余
int index = id % mqs.size();
// 根据取余结果选择消息要发送给哪个消息队列
return mqs.get(index);
}
}, orderId); // 这里传入了订单ID
System.out.printf("%s%n", sendResult);
}
producer.shutdown();
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
e.printStackTrace();
}
}
}
消费者
public class Consumer {
public static void main(String[] args) throws MQClientException {
// 创建消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("消费者组");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
// 订阅主题
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
// 注册消息监听器，使用的是MessageListenerOrderly
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
context.setAutoCommit(true);
// 打印消息
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
}
从例子中可以看出生产者在发送消息的时候，通过订单ID作为路由信息，将同一个订单ID的消息发送到了同一个消息队列中，保证同一个订单ID相关消息有序发送，接下来就看消费者是如何保证消息的顺序消费的。
定时任务对消息队列加锁
消费者在启动的时候，会对是否是顺序消费进行判断（监听器是否是MessageListenerOrderly类型来判断），如果是顺序消费，会使用ConsumeMessageOrderlyService，并调用它的start方法进行启动，在集群模式模式下，start方法中会启动一个定时加锁的任务，周期性的对该消费者负责的消息队列进行加锁。
为什么集群模式下需要加锁？
因为广播模式下，消息队列会分配给消费者下的每一个消费者，而在集群模式下，一个消息队列同一时刻只能被同一个消费组下的某一个消费者进行，所以在广播模式下不存在竞争关系，也就不需要对消息队列进行加锁，而在集群模式下，有可能因为负载均衡等原因将某一个消息队列分配到了另外一个消费者中，因此在顺序消费情况下，集群模式下需要对消息队列加锁，当某个消息队列被锁定时，其他的消费者不能进行消费。
加锁的具体逻辑如下，首先获取当前消费者负责的所有消息队列MessageQueue，返回数据是一个MAP，key为broker名称，value为broker下的消息队列，接着对MAP进行遍历，处理每一个broker下的消息队列：
（1）根据broker名称查找broker的详细信息；
（2）创建加锁请求，在请求中设置要加锁的消息队列，将请求发送给broker，表示要对这些消息队列进行加锁；
（3）Broker返回请求处理结果，响应结果中包含了加锁成功的消息队列，对于加锁成功的消息队列将消息队列MessageQueue，将其对应的ProcessQueue中的locked属性置为true表示该消息队列已加锁成功，如果响应中未包含某个消息队列的信息，表示此消息队列加锁失败，需要将其对应的ProcessQueue对象中的locked属性置为false表示加锁失