- 普及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<String, Object> 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 (鞠躬致意)
- 解决方案之二——待补充
未完待续...
标签:String,一站式,messageBody,消息,new,null,体过,RocketMq,out From: https://www.cnblogs.com/gongjing999/p/16911890.html