优点:
批量发送消息可以提高rocketmq的生产者性能和吞吐量。
使用场景:
- 发送大量小型消息时;
- 需要降低消息发送延迟时;
- 需要提高生产者性能时;
注意事项:
- 消息列表的大小不能超过broker设置的最大消息大小;
- 消息列表的大小不能超过生产证设置的maxMessageSize 参数,此参数默认为 4MB;
- 批量发送消息不支持消息事务;
- 如果代码在发送消息列表时发生异常,则可能会发生部分消息发送成功,部分消息发送失败的情况。如果要确保所有消息都已成功发送,则需要增加错误处理逻辑和消息重试机制;
批量发送消息为什么要限制maxMessageSize❓
消息列表的大小不能超过生产者设置的maxMessageSize参数,主要是为了避免消息发送延迟和消息过大导致broker出现性能问题。如果尝试发送大于maxMessageSize的消息,RocketMQ会抛出MessageTooLargeException异常,并且消息不会被发送到broker。文章来源:https://www.toymoban.com/news/detail-773690.html
如果开发者在开发时遇到了消息列表大小超过maxMessageSize的情况,可以考虑以下几种处理方式:文章来源地址https://www.toymoban.com/news/detail-773690.html
-
- 提升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 ListSplitter<T> implements Iterator<List<T>> {
/**
* 分割数据大小
*/
private int sizeLimit;
/**
* 分割数据列表
*/
private final List<T> messages;
/**
* 分割索引
*/
private int currIndex;
public ListSplitter(int sizeLimit, List<T> messages) {
this.sizeLimit = sizeLimit;
this.messages = messages;
}
@Override
public boolean hasNext() {
return currIndex < messages.size();
}
@Override
public List<T> 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;
}
}
List<T> subList = messages.subList(currIndex, nextIndex);
currIndex = nextIndex;
return subList;
}
}
private final int maxMessageSize = 1024 * 1024 * 4;
/**
* 消息分割(批量发送)
*/
private void bulkSendMsg(List<Message<String>> messageList) {
// 限制数据大小
ListSplitter splitter = new ListSplitter(maxMessageSize, messageList);
while (splitter.hasNext()) {
List<Message> nextList = splitter.next();
syncBulkSendMessage("topic", nextList);
}
}
/**
* @param topic
* @param list
* @description:发送实时消息(批量)
*/
public void syncBulkSendMessage(String topic, List<Message> 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);
}
}
到了这里,关于RocketMQ如何安全的批量发送消息❓的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!