RocketMQ笔记-收发消息特性实践

刺骨的言语ヽ痛彻心扉 2022-05-09 09:14 180阅读 0赞

# RocketMQ笔记-收发消息特性实践 #

# 1.批量发送消息 #

其send方法参数是一个Message的列表

public class BatchProducer {
        public static void main(String[] args) throws Exception{
            //Instantiate with a producer group name.
            DefaultMQProducer producer = new DefaultMQProducer("example_group_name");
            //Launch the instance.
            producer.start();
            List<Message> messages = new ArrayList<Message>();
            messages.add(new Message("TopicTest", "TagA", "OrderID001", "Hello world 0".getBytes()));
            messages.add(new Message("TopicTest", "TagA", "OrderID002", "Hello world 1".getBytes()));
            messages.add(new Message("TopicTest", "TagA", "OrderID003", "Hello world 2".getBytes()));
            try {
                producer.send(messages);
            } catch (Exception e) {
                e.printStackTrace();
                //handle the error
            }
            //Shut down once the producer instance is not longer in use.
            producer.shutdown();
        }
    }

# 1.1 发送消息大小限制 #

消息体最大限制为1M，如果列表的大小超过限制，则需要对列表进行分割处理

public class ListSplitter implements Iterator<List<Message>> {
        private final int SIZE_LIMIT = 1000 * 1000;
        private final List<Message> messages;
        private int currIndex;
        public ListSplitter(List<Message> messages) {
                this.messages = messages;
        }
        @Override public boolean hasNext() {
            return currIndex < messages.size();
        }
        @Override public List<Message> next() {
            int nextIndex = currIndex;
            int totalSize = 0;
            for (; nextIndex < messages.size(); nextIndex++) {
                Message message = messages.get(nextIndex);
                int tmpSize = message.getTopic().length() + message.getBody().length;
                Map<String, String> properties = message.getProperties();
                for (Map.Entry<String, String> entry : properties.entrySet()) {
                    tmpSize += entry.getKey().length() + entry.getValue().length();
                }
                tmpSize = tmpSize + 20; //for log overhead
                if (tmpSize > SIZE_LIMIT) {
                    //it is unexpected that single message exceeds the SIZE_LIMIT
                    //here just let it go, otherwise it will block the splitting process
                    if (nextIndex - currIndex == 0) {
                       //if the next sublist has no element, add this one and then break, otherwise just break
                       nextIndex++;  
                    }
                    break;
                }
                if (tmpSize + totalSize > SIZE_LIMIT) {
                    break;
                } else {
                    totalSize += tmpSize;
                }
        
            }
            List<Message> subList = messages.subList(currIndex, nextIndex);
            currIndex = nextIndex;
            return subList;
        }
    }
    //then you could split the large list into small ones:
    ListSplitter splitter = new ListSplitter(messages);
    while (splitter.hasNext()) {
       try {
           List<Message>  listItem = splitter.next();
           producer.send(listItem);
       } catch (Exception e) {
           e.printStackTrace();
           //handle the error
       }
    }

# 2.Tag #

用于消息过滤

*  Producer在发送消息时,一个消息只能有一个Tag
 *  Consumer可以订阅多个Tag,用于接收过滤的消息

//Producer
            for (int i = 0; i < 100; i++){
                Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            }
    
           //consumer
            consumer.subscribe("TopicTest", "TagA || TagC || TagD");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
     ConsumeConcurrentlyContext context) {
                    System.out.printf(Thread.currentThread().getName() +
     " Receive New Messages: " + msgs + "%n");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

# 3.Key #

Key一般用消息在业务层面的唯一标识码，可以方便后续跟踪查询，尽量保证Key的唯一性

Message msg = new Message("TopicTest" ,
                        "TagA" ,
                        ("Hello RocketMQ ").getBytes(RemotingHelper.DEFAULT_CHARSET)
                        /* Message body */
                );
                msg.setKeys("orderId");

# 4.设置Name Server地址 #

有2种方法设置Name Server地址

1.  在环境变量中设置NAMESRV\_ADDR  
    `export NAMESRV_ADDR=localhost:9876`  
    2.在代码中调用setNamesrvAddr方法进行设置,Producer和Consumer都需要设置

DefaultMQProducer producer = new DefaultMQProducer("example_group_name");
            producer.setNamesrvAddr("localhost:9876");
    
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
            consumer.setNamesrvAddr("localhost:9876");

否则会抛异常

Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: No name server address, please set it.
    See http://rocketmq.apache.org/docs/faq/ for further details.
        at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:560)
        at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1069)
        at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1023)
        at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:214)

# 5. 发送延迟消息 #

延迟消息的使用方法是在创建Message对象时，调用setDelayTimeLevel（intlevel）方法设置延迟时间，然后再把这个消息发送出去。目前延迟的时间不支持任意设置，仅支持预设值的时间长度（1s/5s/10s/30s/1m/2m/3m/4m/5m/6m/7m/8m/9m/10m/20m/30m/1h/2h）。比如setDelayTimeLevel（3）表示延迟10s。

public class ScheduledMessageProducer {
    
        public static void main(String[] args) throws Exception {
            // Instantiate a producer to send scheduled messages
            DefaultMQProducer producer = new DefaultMQProducer("example_group_name");
            // Launch producer
            producer.start();
            int totalMessagesToSend = 10;
            for (int i = 0; i < totalMessagesToSend; i++) {
                Message message = new Message("TopicTest", ("Hello scheduled message " + i).getBytes());
                // This message will be delivered to consumer 10 seconds later.
                message.setDelayTimeLevel(3);
                // Send the message
                producer.send(message);
            }
    
            // Shutdown producer after use.
            producer.shutdown();
        }
    
    }

# 6. Consumer线程数设置 #

消费者使用一个 ThreadPoolExecutor 来处理内部的消费，因此您可以通过设置`setConsumeThreadMin`或`setConsumeThreadMax`来更改它。

![1000][]

# 7.Consumer启动消费起点 #

当建立一个新的 Consumer Group 时，需要决定是否需要消费 Broker 中已经存在的历史消息。

*  CONSUME\_FROM\_FIRST\_OFFSET 将消耗 Broker 中存在的所有消息。您还可以使用
 *  CONSUME\_FROM\_LAST\_OFFSET 将忽略历史消息，并消费此后生成的任何内容。
 *  CONSUME\_FROM\_TIMESTAMP 来消费在指定的时间戳之后生成的消息

//consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            //consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
            //consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(
            //        System.currentTimeMillis() - (1000 * 60 * 1)));

# 8.UserProperty #

设置用户自定义键值对数据

//Producer
                Message msg = new Message("TopicTest" ,
                        "TagA" ,
                        ("Hello RocketMQ ").getBytes(RemotingHelper.DEFAULT_CHARSET)
                        /* Message body */
                );
                msg.putUserProperty("test","testValue");
                //Consumer
               String val=msgs.get(0).getUserProperty("test");
               System.out.printf(val);

# 9.批量消费 #

设置consumeMessageBatchMaxSize属性,默认为1

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
            consumer.setNamesrvAddr("localhost:9876");
            consumer.setConsumeMessageBatchMaxSize(10);

[1000]: /images/20220509/ed81686e12b94b109fb0200da8008c2e.png