批量消息

批量發(fā)送消息能顯著提高傳遞小消息的性能。限制是這些批量消息應(yīng)該有相同的topic，相同的waitStoreMsgOK，而且不能是延時(shí)消息。此外，這一批消息的總大小不應(yīng)超過4MB。

發(fā)送批量消息

如果您每次只發(fā)送不超過4MB的消息，則很容易使用批處理，樣例如下：

package com.leon.mq.rocketmq.batch;import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendStatus; import org.apache.rocketmq.common.message.Message;import java.util.ArrayList; import java.util.List; import java.util.concurrent.TimeUnit;public class Producer {public static void main(String[] args) throws Exception {//1.創(chuàng)建消息生產(chǎn)者producer，并制定生產(chǎn)者組名DefaultMQProducer producer = new DefaultMQProducer("group1");//2.指定Nameserver地址producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//3.啟動(dòng)producerproducer.start();List<Message> msgs = new ArrayList<Message>();//4.創(chuàng)建消息對(duì)象，指定主題Topic、Tag和消息體/*** 參數(shù)一：消息主題Topic* 參數(shù)二：消息Tag* 參數(shù)三：消息內(nèi)容*/Message msg1 = new Message("BatchTopic", "Tag1", ("Hello World" + 1).getBytes());Message msg2 = new Message("BatchTopic", "Tag1", ("Hello World" + 2).getBytes());Message msg3 = new Message("BatchTopic", "Tag1", ("Hello World" + 3).getBytes());msgs.add(msg1);msgs.add(msg2);msgs.add(msg3);//5.發(fā)送消息SendResult result = producer.send(msgs);//發(fā)送狀態(tài)SendStatus status = result.getSendStatus();System.out.println("發(fā)送結(jié)果:" + result);//線程睡1秒TimeUnit.SECONDS.sleep(1);//6.關(guān)閉生產(chǎn)者producerproducer.shutdown();}}

如果消息的總長(zhǎng)度可能大于4MB時(shí)，這時(shí)候最好把消息進(jìn)行分割

public class ListSplitter implements Iterator<List<Message>> {private final int SIZE_LIMIT = 1024 * 1024 * 4;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; // 增加日志的開銷20字節(jié)if (tmpSize > SIZE_LIMIT) {//單個(gè)消息超過了最大的限制//忽略,否則會(huì)阻塞分裂的進(jìn)程if (nextIndex - currIndex == 0) {//假如下一個(gè)子列表沒有元素,則添加這個(gè)子列表然后退出循環(huán),否則只是退出循環(huán)nextIndex++;}break;}if (tmpSize + totalSize > SIZE_LIMIT) {break;} else {totalSize += tmpSize;}}List<Message> subList = messages.subList(currIndex, nextIndex);currIndex = nextIndex;return subList;} } //把大的消息分裂成若干個(gè)小的消息 ListSplitter splitter = new ListSplitter(messages); while (splitter.hasNext()) {try {List<Message> listItem = splitter.next();producer.send(listItem);} catch (Exception e) {e.printStackTrace();//處理error} } package com.leon.mq.rocketmq.batch;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.common.message.MessageExt;import java.util.List;public class Consumer {public static void main(String[] args) throws Exception {//1.創(chuàng)建消費(fèi)者Consumer，制定消費(fèi)者組名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2.指定Nameserver地址consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//3.訂閱主題Topic和Tagconsumer.subscribe("BatchTopic", "*");//4.設(shè)置回調(diào)函數(shù)，處理消息consumer.registerMessageListener(new MessageListenerConcurrently() {//接受消息內(nèi)容@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//5.啟動(dòng)消費(fèi)者consumerconsumer.start();System.out.println("消費(fèi)者啟動(dòng)");} }

?

總結(jié)

以上是生活随笔為你收集整理的批量消息发送的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。