网站域名迁移公告,网站开发查找漏洞的工具,网站优化费用报价明细,crm客户关系管理软件优点#xff1a;
批量发送消息可以提高rocketmq的生产者性能和吞吐量。
使用场景:
发送大量小型消息时#xff1b;需要降低消息发送延迟时#xff1b;需要提高生产者性能时#xff1b;
注意事项#xff1a;
消息列表的大小不能超过broker设置的最大消息大小;消息列表…优点
批量发送消息可以提高rocketmq的生产者性能和吞吐量。
使用场景:
发送大量小型消息时需要降低消息发送延迟时需要提高生产者性能时
注意事项
消息列表的大小不能超过broker设置的最大消息大小;消息列表的大小不能超过生产证设置的maxMessageSize 参数此参数默认为 4MB批量发送消息不支持消息事务如果代码在发送消息列表时发生异常则可能会发生部分消息发送成功部分消息发送失败的情况。如果要确保所有消息都已成功发送则需要增加错误处理逻辑和消息重试机制 批量发送消息为什么要限制maxMessageSize❓
消息列表的大小不能超过生产者设置的maxMessageSize参数主要是为了避免消息发送延迟和消息过大导致broker出现性能问题。如果尝试发送大于maxMessageSize的消息RocketMQ会抛出MessageTooLargeException异常并且消息不会被发送到broker。
如果开发者在开发时遇到了消息列表大小超过maxMessageSize的情况可以考虑以下几种处理方式 提升maxMessageSize参数的大小这样可以容纳更大的消息列表。但是需要注意在提升参数大小时要考虑到RocketMQ broker的性能和网络带宽等因素。考虑将消息列表进行拆分然后分批发送。这样可以避免一次发送过多的消息。计算消息的大小并进行压缩。可以使用一些压缩算法如 LZ4、Snappy 等对消息进行压缩以减小消息的大小。对超过 maxMessageSize 的消息进行过滤或其他处理。可以通过业务逻辑对消息进行分组或分类对超过 maxMessageSize 的消息进行过滤或其他处理以避免发送超出限制的消息。 代码实现
package com.resource.sync.rocketmq;import java.util.Iterator;
import java.util.List;/*** description:消息分割在rocketmq中一次性发送消息的长度不可超过4mb此时我们需要进行切割确保消息长度小于4mb**/
public class ListSplitterT implements IteratorListT {/*** 分割数据大小*/private int sizeLimit;/*** 分割数据列表*/private final ListT messages;/*** 分割索引*/private int currIndex;public ListSplitter(int sizeLimit, ListT messages) {this.sizeLimit sizeLimit;this.messages messages;}Overridepublic boolean hasNext() {return currIndex messages.size();}Overridepublic ListT next() {int nextIndex currIndex;int totalSize 0;for (; nextIndex messages.size(); nextIndex) {T t messages.get(nextIndex);totalSize totalSize t.toString().length();if (totalSize sizeLimit) {break;}}ListT subList messages.subList(currIndex, nextIndex);currIndex nextIndex;return subList;}
}private final int maxMessageSize 1024 * 1024 * 4;/*** 消息分割(批量发送)*/private void bulkSendMsg(ListMessageString messageList) {// 限制数据大小ListSplitter splitter new ListSplitter(maxMessageSize, messageList);while (splitter.hasNext()) {ListMessage nextList splitter.next();syncBulkSendMessage(topic, nextList);}}/*** param topic* param list* description:发送实时消息(批量)*/public void syncBulkSendMessage(String topic, ListMessage list) {SendResult sendResult null;try {sendResult rocketMQTemplate.syncSend(topic, list);if (sendResult.getSendStatus() ! SendStatus.SEND_OK) {log.error(BULK_ROCKET_MQ_DISTRIBUTION_ERROR.RESULT_STATUS:{},MSG_ID:{}, sendResult.getSendStatus(), sendResult.getMsgId());}if (sendResult.getSendStatus() SendStatus.SEND_OK) {log.info(BULK_SEND_MSG_SUCCESS.MSG_ID:{}, sendResult.getMsgId());}} catch (Exception e) {log.error(BULK_ROCKET_MQ_DISTRIBUTION_ERROR:{}, e);}}
