RocketMQ实践之有序消息（三）

叁歲伎倆 2022-09-15 06:10 215阅读 0赞

**发送消息：**

--------------------

package my.demo;
    
    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 java.util.List;
    
    public class OrderedProducer { 
        public static void main(String[] args) throws Exception { 
            DefaultMQProducer producer = new DefaultMQProducer("p_group5");
            producer.setNamesrvAddr("localhost:9876");
            producer.start();
            String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD", "TagE"};
            for (int i = 0; i < 100; i++) { 
                int orderId = i % 10;
                Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).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);
                }
                }, orderId);//此处orderId会被传入回调方法select中，对应arg参数
    
                System.out.printf("%s%n", sendResult);
            }
            producer.shutdown();
        }
    }

**接收消息：**

--------------------

package my.demo;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    
    
    public class OrderedConsumer { 
        public static void main(String[] args) throws Exception { 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("c_group10");
    
            consumer.setNamesrvAddr("localhost:9876");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            consumer.subscribe("TopicTestjjjj", "TagA || TagC || TagD");
    
            consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> { 
                msgs.stream().forEach(msg -> System.out.println("currentThread:" + Thread.currentThread().getName() + "queueId:" + msg.getQueueId() + "content:" + new String(msg.getBody())));
                return ConsumeOrderlyStatus.SUCCESS;
            });
    
            consumer.start();
    
            System.out.printf("Consumer Started.%n");
        }
    }

**效果展示：**

--------------------

D:\dev\Java\jdk1.8.0_261\bin\java.exe "-javaagent:D:\dev\IntelliJ IDEA 2021.2\lib\idea_rt.jar=54937:D:\dev\IntelliJ IDEA 2021.2\bin" -Dfile.encoding=UTF-8 -classpath D:\dev\Java\jdk1.8.0_261\jre\lib\charsets.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\deploy.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\access-bridge-64.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\cldrdata.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\dnsns.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\jaccess.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\jfxrt.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\localedata.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\nashorn.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\sunec.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\sunjce_provider.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\sunmscapi.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\sunpkcs11.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\zipfs.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\javaws.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\jce.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\jfr.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\jfxswt.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\jsse.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\management-agent.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\plugin.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\resources.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\rt.jar;D:\dev\code\koala\rocket-mq\target\classes;C:\Users\Administrator\.m2\repository\org\apache\rocketmq\rocketmq-client\4.9.1\rocketmq-client-4.9.1.jar;C:\Users\Administrator\.m2\repository\org\apache\rocketmq\rocketmq-common\4.9.1\rocketmq-common-4.9.1.jar;C:\Users\Administrator\.m2\repository\org\apache\rocketmq\rocketmq-remoting\4.9.1\rocketmq-remoting-4.9.1.jar;C:\Users\Administrator\.m2\repository\com\alibaba\fastjson\1.2.76\fastjson-1.2.76.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-all\4.1.65.Final\netty-all-4.1.65.Final.jar;C:\Users\Administrator\.m2\repository\org\apache\rocketmq\rocketmq-logging\4.9.1\rocketmq-logging-4.9.1.jar;C:\Users\Administrator\.m2\repository\commons-validator\commons-validator\1.7\commons-validator-1.7.jar;C:\Users\Administrator\.m2\repository\commons-beanutils\commons-beanutils\1.9.4\commons-beanutils-1.9.4.jar;C:\Users\Administrator\.m2\repository\commons-digester\commons-digester\2.1\commons-digester-2.1.jar;C:\Users\Administrator\.m2\repository\commons-logging\commons-logging\1.2\commons-logging-1.2.jar;C:\Users\Administrator\.m2\repository\commons-collections\commons-collections\3.2.2\commons-collections-3.2.2.jar;C:\Users\Administrator\.m2\repository\org\apache\commons\commons-lang3\3.4\commons-lang3-3.4.jar;C:\Users\Administrator\.m2\repository\commons-codec\commons-codec\1.9\commons-codec-1.9.jar my.demo.OrderedConsumer
    Consumer Started.
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:2
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_1queueId:2content:Hello RocketMQ orderId:6
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:1
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:5
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8
    currentThread:ConsumeMessageThread_3queueId:1content:Hello RocketMQ orderId:9
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:0
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:3
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:4
    currentThread:ConsumeMessageThread_2queueId:3content:Hello RocketMQ orderId:7
    currentThread:ConsumeMessageThread_4queueId:0content:Hello RocketMQ orderId:8

顺序消费同样的currentThread一定对应同样的queueId，而queue的选择是按照orderId对队列数取余的，可以看出我实验中Topic：TopicTestjjjj对应的队列数是4个，那就是orderId对4取余，结果没有问题的。