RocketMQ实现顺序消息
程序员文章站
2022-07-14 23:52:14
...
大部分业务场景不需要消息的产生和消费的顺序相同(顺序消费)。但在某些场景(比如 订单的生成,付款 ,下单 这3个消息必须按顺序处理)。解决思路是:(1)在发送端,要把统一业务ID 发送到同一个MessageQueue. (2) 在消费端,同一MessageQueue 的消息不要被并发处理了。
针对(1) , 我们可以使用RocketMQ 提供的 MesssageQueueSelector 接口以及默认的3种实现(例如SelectMessageQueueByHash, 你也可以自己实现这个接口),控制消息向哪个MessageQueue 发送。代码如下:
public class Producer {
public static void main(String[] args) throws MQClientException {
DefaultMQProducer producer = new DefaultMQProducer("orderProducerGroup");
producer.setNamesrvAddr("10.137.38.61:9876");
producer.start();
int orderId = 1; // 同一个订单号
for (int i = 0; i < 20; i++) {
try {
Message msg = new Message("TopicA", "TagA", "Key" + i ,("OrderId" + orderId + ", Msg body" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg, new SelectMessageQueueByHash(), orderId);
System.out.println(sendResult.toString());
} catch (Exception e) {
System.out.println(e);
}
}
producer.shutdown();
}
}
针对(2) 每一个 RocketMQ 提供了 MessageListenerOrderly 接口 作为一个RegisterMessageListener。 这种消息监听保证每次消费ConsumerQueue时会对每个ConsumerQueue 加锁,从而保证同一个ConsumerQueue 的消息不会多个Consumer 并发消费,又同样保证了不同ConsumerQueue 的消息可以被并发消费。这样消费一个ConsumerQueue里的消息就保证了顺序性。代码如下:
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderConsumerGroup");
consumer.setNamesrvAddr("10.137.38.61:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
consumer.subscribe("TopicA", "TagA");
consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
MessageExt msg = msgs.get(0);
System.out.println(msg.toString());
return ConsumeOrderlyStatus.SUCCESS;
});
consumer.start();
System.out.println("Consumer start.");
}
}
如下图, consumer 端的 日志, 显示 所用的消息orderId=1 的消息都发送到了queueId=1 的 MessageQueue 中了。
上一篇: call() 和 apply() 的相关介绍和应用场景
下一篇: Date Format