RocketMq消息体过大一站式解决方案

• 普及RocketMq消息体大小限制相关知识如下：

1. 消息体大小最大为 4MB, 一般建议发送的消息体在 4kb 之内 ( 性能最佳 )。

2. 消息属性最大为 32kb，一般建议发送的消息属性在 1kb 之内 ( 性能最佳 )。

3. 4MB 这个上限值不能修改，这个会影响全局性能。如果消息体的确很大，建 议侧优化消息体的内容，避免发送大消息或者带有链接地址的消息，或者可 以缩短或者分两条发送。

附送问答链接：https://developer.aliyun.com/ask/314873   （鞠躬致意）

• 解决方案之一——压缩消息体：

 该方法总体思路是在消息生产端对消息体进行压缩，然后在消费端对消息体进行解压缩。

消息生产端消息压缩代码示例如下：

String messageBody = "假装这是一个巨大巨大的消息体，竟然足足有5M之多，超过了4M";
//rocketMQMsgMaxSize（消息体最大限制）：4（兆） = 4*1024*1024（字节）
Integer rocketMQMsgMaxSize = 4 * 1024 * 1024;
Map headers = new HashMap<>();
//如果消息体（字节为单位）的长度大于限定长度，则消息压缩
if (rocketMQMsgMaxSize < messageBody.getBytes().length) {
//压缩消息的方法
messageBody = StringCompressUtils.compress(messageBody);
//为区别压缩消息和非压缩消息，需要在消息上加一个标识，此处选择了在消息的key标识上添加compressed。
// （key标识一般用来标识一个唯一的消息，用于消息检索，这里不是常规用法，但是也可以用）
headers.put(RocketMQHeaders.KEYS, "compressed");
}
//这是rocketmq发消息的方法，发送带key的消息
producerService.sendNormalMessage("我是消息主题",headers, messageBody);

消息消费端消息解压缩代码示例如下：

//获取消息体
String messageBody = new String((byte[]) msg.getPayload());
//获取消息的key标识，用于判断是不是压缩过的消息
String keys = (String)msg.getHeaders().get("rocketmq_KEYS");
//如果是压缩过的消息就解压缩，如果不是就直接转ExaEntity实体（示例实体）用于业务开发
ExaEntity exaEntity = "compressed".equals(keys) ? GsonUtils.toObject(StringCompressUtils.uncompress(messageBody), ExaEntity.class) : GsonUtils.toObject(messageBody, ExaEntity.class);

附上字符压缩与解压缩的方法

public class StringCompressUtils {
/**
* 使用gzip压缩字符串
* @param str 要压缩的字符串
* @return
*/
public static String compress(String str) {
if (str == null || str.length() == 0) {
return str;
}
ByteArrayOutputStream out = new ByteArrayOutputStream();
GZIPOutputStream gzip = null;
try {
gzip = new GZIPOutputStream(out);
gzip.write(str.getBytes());
} catch (IOException e) {
e.printStackTrace();
} finally {
if (gzip != null) {
try {
gzip.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return new sun.misc.BASE64Encoder().encode(out.toByteArray());
}
/**
* 使用gzip解压缩
* @param compressedStr 压缩字符串
* @return
*/
public static String uncompress(String compressedStr) {
if (compressedStr == null) {
return null;
}
ByteArrayOutputStream out = new ByteArrayOutputStream();
ByteArrayInputStream in = null;
GZIPInputStream ginzip = null;
byte[] compressed = null;
String decompressed = null;
try {
compressed = new sun.misc.BASE64Decoder().decodeBuffer(compressedStr);
in = new ByteArrayInputStream(compressed);
ginzip = new GZIPInputStream(in);
byte[] buffer = new byte[1024];
int offset = -1;
while ((offset = ginzip.read(buffer)) != -1) {
out.write(buffer, 0, offset);
}
decompressed = out.toString();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (ginzip != null) {
try {
ginzip.close();
} catch (IOException e) {
}
}
if (in != null) {
try {
in.close();
} catch (IOException e) {
}
}
if (out != null) {
try {
out.close();
} catch (IOException e) {
}
}
}
return decompressed;
}
}

View Code

此方法来自于  https://www.jb51.net/article/206886.htm  （鞠躬致意）

• 解决方案之二——待补充

未完待续...