我们知道Topic的有序消息已经成为mq的标配。而RocketMQ中是这样区分消息类型的, 普通消息也叫做无序消息,简单来说就是没有顺序的消息,而有序消息就是按照一定的先后顺序的消息类型。举个例子,producer 依次发送 order id 为 1、2、3 的消息到 broker,consumer 接到的消息顺序也就是 1、2、3 ,而不会出现普通消息那样的 2、1、3 等情况。
那有序消息该如何实现呢?
理论上:我们都知道消息首先由 producer 到 broker,再从 broker 到 consumer,分这两步走。那么要保证消息的有序,势必这两步都是要保证有序的,即要保证消息是按有序发送到 broker,broker 也是有序将消息投递给 consumer,两个条件必须同时满足,缺一不可。
实现方式呢?
全局有序消息:由于一个 topic 只有一个 queue ,即使我们有多个 producer 实例和 consumer 实例也很难提高消息吞吐量。就好比过独木桥,大家只能一个挨着一个过去,效率低下。
局部有序消息:常见做法就是将 order id 进行处理,将 order id 相同的消息发送到 topicB 的同一个 queue,假设我们 topicB 有 2 个 queue,那么我们可以简单的对 id 取余,奇数的发往 queue0,偶数的发往 queue1,消费者按照 queue 去消费时,就能保证 queue0 里面的消息有序消费,queue1 里面的消息有序消费。
科普一下RocketMQ的topic
topic 只是消息的逻辑分类,内部实现其实是由 queue 组成。当 producer 把消息发送到某个 topic 时,默认是会消息发送到具体的 queue 上。由于一个 topic 可以有多个 queue,所以在性能比全局有序高得多。假设 queue 数是 n,理论上性能就是全局有序的 n 倍,当然 consumer 也要跟着增加才行。在实际情况中,这种局部有序消息是会比全局有序消息用的更多。
public class Producer {
public static void main(String[] args) throws UnsupportedEncodingException {
try {
// 声明并初始化一个producer
// 需要一个producer group名字作为构造方法的参数,这里为ordered_producer
DefaultMQProducer orderedProducer = new DefaultMQProducer("ordered_producer");
// 设置NameServer地址,此处应改为实际NameServer地址,多个地址之间用;分隔
//NameServer的地址必须有,但是也可以通过环境变量的方式设置,不一定非得写死在代码里
orderedProducer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
// 调用start()方法启动一个producer实例
orderedProducer.start();
// 自定义一个tag数组
String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
// 发送10条消息到Topic为TopicTestOrdered,tag为tags数组按顺序取值,
// key值为“KEY”拼接上i的值,消息内容为“Hello RocketMQ”拼接上i的值
for (int i = 0; i < 10; i++) {
int orderId = i % 10;
Message msg =
new Message("TopicTestOrdered", tags[i % tags.length], "KEY" + i,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
// 这里设置需要延时的等级即可
msg.setDelayTimeLevel(3);
SendResult sendResult = orderedProducer.send(msg, new MessageQueueSelector() {
// 选择发送消息的队列
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
// arg的值其实就是orderId
Integer id = (Integer) arg;
// mqs是队列集合,也就是topic所对应的所有队列
int index = id % mqs.size();
// 这里根据前面的id对队列集合大小求余来返回所对应的队列
return mqs.get(index);
}
}, orderId);
System.out.println(sendResult);
}
orderedProducer.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
}
public class Consumer {
public static void main(String[] args) throws MQClientException {
//声明并初始化一个consumer
//需要一个consumer group名字作为构造方法的参数,这里为concurrent_consumer
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ordered_consumer");
//同样也要设置NameServer地址
consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
//这里设置的是一个consumer的消费策略
//CONSUME_FROM_LAST_OFFSET 默认策略,从该队列最尾开始消费,即跳过历史消息
//CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费,即历史消息(还储存在broker的)全部消费一遍
//CONSUME_FROM_TIMESTAMP 从某个时间点开始消费,和setConsumeTimestamp()配合使用,默认是半个小时以前
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//设置consumer所订阅的Topic和Tag
consumer.subscribe("TopicTestOrdered", "TagA || TagC || TagD");
//设置一个Listener,主要进行消息的逻辑处理
//注意这里使用的是MessageListenerOrderly这个接口
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
//返回消费状态
//SUCCESS 消费成功
//SUSPEND_CURRENT_QUEUE_A_MOMENT 消费失败,暂停当前队列的消费
return ConsumeOrderlyStatus.SUCCESS;
}
});
//调用start()方法启动consumer
consumer.start();
System.out.println("Consumer Started.");
}
}
延时消息呢?
延时消息,简单来说就是当 producer 将消息发送到 broker 后,会延时一定时间后才投递给 consumer 进行消费。RcoketMQ的延时等级为:1s,5s,10s,30s,1m,2m,3m,4m,5m,6m,7m,8m,9m,10m,20m,30m,1h,2h。level=0,表示不延时。level=1,表示 1 级延时,对应延时 1s。level=2 表示 2 级延时,对应5s,以此类推。这种消息一般适用于消息生产和消费之间有时间窗口要求的场景。比如说我们网购时,下单之后是有一个支付时间,超过这个时间未支付,系统就应该自动关闭该笔订单。那么在订单创建的时候就会就需要发送一条延时消息(延时15分钟)后投递给 consumer,consumer 接收消息后再对订单的支付状态进行判断是否关闭订单。