欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

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 中了。
RocketMQ实现顺序消息