1. 顺序消息原理图

2. 什么是顺序消息?

  消费消息的顺序要求同发送消息的顺序一致,在RocketMQ中,主要指的是局部顺序,即一类消息为满足顺序性,必须Producer单线程顺序发送,并且发送给到同一队列,这样Consumer就可以按照Producer发送的顺序去消费消息。

  2.1 普通顺序消息

  正常情况下可以保证完全的顺序消息,但是一旦发生通信异常,Broker重启,由于队列总数发生变化,哈希取模后定位的队列会变化,产生短暂的消息顺序不一致。

  如果业务能够容忍在集群异常(如某个Broker宕机或者重启)下,消息短暂乱序,使用普通顺序方式比较合适。

  2.2 严格顺序消息

  无论正常异常都保证顺序,但是牺牲了分布式Failover特性,即Broker集群中只要有一台机器不可用,则整个集群都不可用,服务可用性大大降低。

  如果服务器部署方式为同步双写模式,此缺陷可通过备机自动切换为主避免,不过仍然会存在短暂时间服务不可用。

  目前已知的应用只有数据库binlog同步强依赖严格顺序消息。其他应用绝大部分都可以容忍短暂乱序,推荐使用普通顺序消息。

3. 顺序消息缺陷 

  1. 发送顺序消息无法利用集群FailOver 特性
  2.  消费顺序消息的幵行度依赖亍队列数量
  3.  队列热点问题,个别队列由亍哈希丌均导致消息过多,消费速度跟丌上,产生消息堆积问题
  4.  遇到消息失败的消息,无法跳过,当前队列消费暂停

5. 顺序消息的使用方式

  跟普通消息相比,顺序消息的使用需要在producer的send()方法中添加MessageQueueSelector接口的实现类,并重写select选择使用的队列,因为顺序消息局部顺序,需要将所有消息指定发送到同一队列中。

  Producer实现如下:

package com.wangx.rocketmq.order;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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 org.apache.rocketmq.remoting.exception.RemotingException; import java.util.List; public class OrderProducer { public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer producer = new DefaultMQProducer("my-order-producer-group"); //2. 设置namesrvAddr,集群环境多个nameserver用;分割 producer.setNamesrvAddr("47.105.149.61:9876;47.105.145.123:9876");
//3. 启动
producer.start(); for (int i = 0; i < 4; i++) {
Message message = new Message("OrderTopic", "tabA", ("Hello World " + i).getBytes());
/**send() 方法中的MessageQueueSelector.select()方法用于指定消息队列,
* send() 的最后一个参数为队列下表,一个topic默认初始化四个队列
* select()方法中的第一个参数为队列list,里面存放队列,第二个为send()方法传入的message
* 第三个参数为send()方法传入的队列下标
*
*/
SendResult result = producer.send(message, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
Integer qID = (Integer) o;
MessageQueue queue = list.get(qID);
return queue;
}
},1);
System.out.println(result);
}
producer.shutdown();
}
}

  与普通消息相比,Consumer的使用只需要在注册监听时使用MessageListenerOrderly对象,并重写consumeMessage即可,需要注意的是consumeMessage的返回值与普通消息不同,具体实现如下:

package com.wangx.rocketmq.order;

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.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt; import java.util.List; public class OrderConsumer { public static void main(String[] args) throws MQClientException {
//实例化一个consumer组
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my-order-consumer-group");
//设置setNamesrvAddr,同生产者
consumer.setNamesrvAddr("47.105.149.61:9876;47.105.145.123:9876"); //设置消息读取方式,这里设置的是队尾开始读取
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); //设置订阅主题,第二个参数为过滤tabs的条件,可以写为tabA|tabB过滤Tab,*表示接受所有
consumer.subscribe("OrderTopic", "*"); //注册消息监听
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
try{
MessageExt ext = list.get(0);
int quID = ext.getQueueId();
String body = new String(ext.getBody(),"UTF-8");
System.out.println("quID: " + quID + "接收到消息:" + body);
}catch (Exception e) {
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
}
}

  启动consumer,在启动producer,控制台打印如下:

quID: 1接收到消息:Hello World 0
quID: 1接收到消息:Hello World 1
quID: 1接收到消息:Hello World 2
quID: 1接收到消息:Hello World 3

  可以看到所有消息都在队列1中,且都是按照发送消息的顺序进行消费的。

  那么当向多个消息队列中分别顺序发送多条消息时是什么情况呢?

  分别下下标为1,2,3的三个队列中发送四条消息,代码如下:

package com.wangx.rocketmq.order;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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 org.apache.rocketmq.remoting.exception.RemotingException; import java.util.List; public class OrderProducer { public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
DefaultMQProducer producer = new DefaultMQProducer("my-order-producer-group"); //2. 设置namesrvAddr,集群环境多个nameserver用;分割 producer.setNamesrvAddr("47.105.149.61:9876;47.105.145.123:9876");
//3. 启动
producer.start(); for (int i = 0; i < 4; i++) {
Message message = new Message("OrderTopic", "tabA", ("Hello World " + i).getBytes());
/**send() 方法中的MessageQueueSelector.select()方法用于指定消息队列,
* send() 的最后一个参数为队列下表,一个topic默认初始化四个队列
* select()方法中的第一个参数为队列list,里面存放队列,第二个为send()方法传入的message
* 第三个参数为send()方法传入的队列下标
*
*/
SendResult result = producer.send(message, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
Integer qID = (Integer) o;
MessageQueue queue = list.get(qID);
return queue;
}
},1);
System.out.println(result);
}
for (int i = 0; i < 4; i++) {
Message message = new Message("OrderTopic", "tabA", ("Hello World " + i).getBytes());
/**send() 方法中的MessageQueueSelector.select()方法用于指定消息队列,
* send() 的最后一个参数为队列下表,一个topic默认初始化四个队列
* select()方法中的第一个参数为队列list,里面存放队列,第二个为send()方法传入的message
* 第三个参数为send()方法传入的队列下标
*
*/
SendResult result = producer.send(message, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
Integer qID = (Integer) o;
MessageQueue queue = list.get(qID);
return queue;
}
},2);
System.out.println(result);
}
for (int i = 0; i < 4; i++) {
Message message = new Message("OrderTopic", "tabA", ("Hello World " + i).getBytes());
/**send() 方法中的MessageQueueSelector.select()方法用于指定消息队列,
* send() 的最后一个参数为队列下表,一个topic默认初始化四个队列
* select()方法中的第一个参数为队列list,里面存放队列,第二个为send()方法传入的message
* 第三个参数为send()方法传入的队列下标
*
*/
SendResult result = producer.send(message, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
Integer qID = (Integer) o;
MessageQueue queue = list.get(qID);
return queue;
}
},3);
System.out.println(result);
}
producer.shutdown();
}
}

  消费者不变,控制台打印如下:

quID: 1接收到消息:Hello World 0
quID: 1接收到消息:Hello World 1
quID: 1接收到消息:Hello World 2
quID: 1接收到消息:Hello World 3
quID: 2接收到消息:Hello World 0
quID: 2接收到消息:Hello World 1
quID: 2接收到消息:Hello World 2
quID: 2接收到消息:Hello World 3
quID: 3接收到消息:Hello World 0
quID: 3接收到消息:Hello World 1
quID: 3接收到消息:Hello World 2
quID: 3接收到消息:Hello World 3

  这里很明显的看到,每个队列中所有消息均是按照顺序消费的,这里可能消息队列并不一定为1,2,3这样的方式顺序消费(因为每个队列对应单个线程,多个线程在跑时并不一定能够保证队列的顺序),但是针对单个队列中的消息,确实顺序的,这就是rockemq局部顺序的概念的实现。