RocktMQ顺序消费

梦里梦外; 2023-06-08 06:41 22阅读 0赞

rocketmq以Topic来管理不同应用的消息。对于生产者而言，发送消息时，需要指定消息的Topic，对于消费者而言，在启动后，需要订阅相应的Topic，然后可以消费响应的消息。在物理实现上，一个Topic由多个Queue组成，采用多个Queue的好处是，可以将Broker存储分布式化，提供系统性能。默认情况下一个Topic下会有4个队列。

RocketMQ中，producer将消息发送给broker时，需要指定发送到哪一个队列中，默认情况下，producer会轮询的将消息发送到每个队列中（所有Broker下的Queue合并成一个list去轮询）。也就是说正常情况下，如果你发送了四条消息，就会分配到4个队列里（我们就认为是均匀分配了）。

而消费者从四个队列里消费，这个时候4个消息不知道是不是按照顺序进入4个队列的，如果是按顺序进入队列，我们要想顺序消费，就是等于说消费的时候按照队列顺序，一个队列一个队列的排队消费。明确说，这个是做不到的。如果不是按顺序进入队列，那更别想进行顺序消费了。

所以我们需要换一个思路。要想实现顺序消费。我们消费的信息就不能随机的乱进队列，我们需要把他们放进一个队列里。这样在一个队列的所有消息，是可以实现一个一个消费的。

SendResult result = producer.send(message);//普通的推送消息  
这个是很普通的推送方式，会随机分发到不同的队列中去的，是无法实现消息顺序消费的。

SendResult result = producer.send(message, new MessageQueueSelector() \{  
    public MessageQueue select(List<MessageQueue> list, Message message, Object o) \{  
        Integer id = (Integer) o;  
        int index = id % list.size();  
        return list.get(index);  
    \}  
\},0);  //指定到第0个队列

应该使用这个方法，这个方法是选择的消息队列了，这样就完成了讲消息放在了一个队列中。如何顺序消费，在consumer代码中也有区别。

rocketmq消费者注册监听有两种模式，有序消费MessageListenerOrderly和并发消费MessageListenerConcurrently，这两种模式返回值不同。

MessageListenerOrderly正确消费返回ConsumeOrderlyStatus.SUCCESS，

稍后消费返回ConsumeOrderlyStatus.SUSPEND\_CURRENT\_QUEUE\_A\_MOMENT

MessageListenerConcurrently正确消费返回ConsumeConcurrentlyStatus.CONSUME\_SUCCESS

稍后消费返回ConsumeConcurrentlyStatus.RECONSUME\_LATER

所以要进行顺序消费我们使用，有序消费的注册监听模式MessageListenerOrderly

代码如下：

import java.io.UnsupportedEncodingException;  
import java.util.List;

import org.apache.commons.lang3.time.StopWatch;  
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.common.RemotingHelper;  
import org.apache.rocketmq.remoting.exception.RemotingException;

public class Server \{  
    DefaultMQProducer producer;  
      
    public static void main(String\[\] args) throws Exception \{  
        Server server =new Server();  
        server.producer = defaultMQProducer();  
          
          
        for (int i = 0; i < 100; i++) \{  
            int orderId = i % 10;  
              
            server.sendInOrder(orderId, "TopicTest", "TagA", "ooooooooo"+orderId+"----"+i);  
        \}  
          
          
        server.producer.shutdown();  
    \}  
      
      
    public static DefaultMQProducer  defaultMQProducer() \{

// 生产者的组名  
        DefaultMQProducer producer = new DefaultMQProducer("TopicTestGroup");  
        // 指定NameServer地址，多个地址以 ; 隔开  
        producer.setNamesrvAddr("127.0.0.1:9876");  
        producer.setVipChannelEnabled(false);  
        try \{  
            producer.start();  
            System.out.println("-------->:producer启动了");  
        \} catch (MQClientException e) \{  
            e.printStackTrace();  
        \}  
        return producer;  
          
          
    \}

/\*\*  
     \* 同步发送  
     \*   
     \* @param topic  
     \* @param tags  
     \* @param body  
     \* @return  
     \* @throws InterruptedException  
     \* @throws RemotingException  
     \* @throws MQClientException  
     \* @throws MQBrokerException  
     \* @throws UnsupportedEncodingException  
     \*/  
    public String send(String topic, String tags, String body) throws InterruptedException, RemotingException,  
            MQClientException, MQBrokerException, UnsupportedEncodingException \{  
        Message message = new Message(topic, tags, body.getBytes(RemotingHelper.DEFAULT\_CHARSET));  
        StopWatch stop = new StopWatch();  
        stop.start();  
        SendResult result = producer.send(message);  
        System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());  
        stop.stop();  
        return "\{\\"MsgId\\":\\"" + result.getMsgId() + "\\"\}";  
    \}

/\*\*  
     \* 按照订单id顺序发送  
     \*   
     \* @param orderId  
     \* @param topic  
     \* @param tags  
     \* @param body  
     \* @return  
     \* @throws MQClientException  
     \* @throws RemotingException  
     \* @throws MQBrokerException  
     \* @throws InterruptedException  
     \* @throws UnsupportedEncodingException  
     \*/  
    public String sendInOrder(int orderId, String topic, String tags, String body) throws MQClientException,  
            RemotingException, MQBrokerException, InterruptedException, UnsupportedEncodingException \{

Message msg = new Message(topic, tags, body.getBytes(RemotingHelper.DEFAULT\_CHARSET));  
        SendResult sendResult = producer.send(msg, new MessageQueueSelector() \{  
            @Override  
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) \{  
                Integer id = (Integer) arg;  
                int index = id % mqs.size();  
                return mqs.get(index);  
            \}  
        \}, 1);//这个1是指定队列的下标  
          
        return "\{\\"MsgId\\":\\"" + sendResult.getMsgId() + "\\"\}";  
    \}

消费端：

import java.io.UnsupportedEncodingException;  
import java.util.List;

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.message.MessageExt;  
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class OrderedConsumerX \{

public static void main(String\[\] args) throws MQClientException \{

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please\_rename\_unique\_group\_name");

consumer.setNamesrvAddr("127.0.0.1:9876");

consumer.subscribe("TopicTest", "TagA");

//顺序消费，注意这里的监听接口  
        consumer.registerMessageListener(new MessageListenerOrderly() \{

@Override  
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) \{  
                MessageExt messageExt = list.get(0);  
                try \{  
                    byte\[\] body = messageExt.getBody();  
                    String tags = messageExt.getTags();  
                    String topic = messageExt.getTopic();  
                    String keys = messageExt.getKeys();  
                    System.out.println("body:"+new String(body, RemotingHelper.DEFAULT\_CHARSET)+" tags:"+tags+" topic:"+topic+" keys:"+keys);  
                \} catch (UnsupportedEncodingException e) \{  
                    e.printStackTrace();  
                    return ConsumeOrderlyStatus.SUSPEND\_CURRENT\_QUEUE\_A\_MOMENT;  
                \}  
                return ConsumeOrderlyStatus.SUCCESS;  
            \}  
        \});

consumer.start();  
        System.out.printf("Consumer Started.%n");  
    \}  
\}  
代码可以参考： [https://github.com/hsn999/start-cloud/tree/master/base-framwork-rocketmq][https_github.com_hsn999_start-cloud_tree_master_base-framwork-rocketmq]

[https_github.com_hsn999_start-cloud_tree_master_base-framwork-rocketmq]: https://github.com/hsn999/start-cloud/tree/master/base-framwork-rocketmq