美文网首页
RocketMQ笔记-收发顺序消息

RocketMQ笔记-收发顺序消息

作者: 兴浩 | 来源:发表于2018-07-23 11:08 被阅读116次

    1.消息发送端(Producer)

    public class OrderedProducer {
        public static void main(String[] args) throws Exception {
            //Instantiate with a producer group name.
            MQProducer producer = new DefaultMQProducer("example_group_name");
            //Launch the instance.
            producer.start();
    
            String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
            for (int i = 0; i < 10; i++) {
                int orderId = i % 10;
                //Create a message instance, specifying topic, tag and message body.
                Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
    
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, i);
    
                System.out.printf("%s%n", sendResult);
            }
            //server shutdown
            producer.shutdown();
        }
    }
    

    1.1 MessageQueueSelector

    MessageQueueSelector如其名,消息队列选择器, DefaultMQProducer 的send方法最后一个参数,会传入MessageQueueSelector的select回调方法中去

    Topic创建后默认会有4个MessageQueue

    要保证部分消息有序,需要发送端和消费端配合处理。 在发送端, 要做到把同一业务 ID的消息发送到同一个Message Queue; 在消费过程中,要做到从同一个 MessageQueue读取的消息不被并发处理, 这样才能达到部分有序。发送端使用MessageQueueSelector类来控制把消息发往哪个Message Queue

    2.消息接收端(Consumer)

    订阅MessageListenerOrderly消息

    public class OrderedConsumer {
        public static void main(String[] args) throws Exception {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
    
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            consumer.subscribe("TopicTest", "TagA || TagC || TagD");
    
            consumer.registerMessageListener(new MessageListenerOrderly() {
    
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                           ConsumeOrderlyContext context) {
                    System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
    
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }
    }
    

    3.全局顺序消息

    参考:
    RocketMq基础认知
    https://help.aliyun.com/document_detail/49323.html

    相关文章

      网友评论

          本文标题:RocketMQ笔记-收发顺序消息

          本文链接:https://www.haomeiwen.com/subject/nxvdmftx.html