首页 > 其他分享 >RocketMq消息体过大一站式解决方案

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

时间:2022-11-21 16:57:26浏览次数:83  
标签:String 一站式 messageBody 消息 new null 体过 RocketMq out

  • 普及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

相关文章

  • HM-RocketMQ2.3【SpringBoot整合Dubbo】
    1前置条件安装依赖包下载dubbo-spring-boot-starter依赖包将dubbo-spring-boot-starter安装到本地仓库mvninstall-Dmaven.skip.test=true注意springboot整......
  • 不科学,RocketMQ生产者在一个应用服务竟然不能向多个NameServer发送消息
    前言目前有两套RocketMQ集群,集群A包含​​topic​​名称为​​cluster_A_topic​​,集群B包含​​topic​​名称为​​cluster_B_topic​​,在应用服务​​OrderApp​​上通过......
  • “一站式”家校服务系统---开发日志1
    一、虚拟机搭建(CentOS)1、自行下载安装VirtualBox。visualBox进行安装需要cpu开启虚拟化,在开机启动的时候设置主板,CPUconfiguration,然后点击IntelVitualizationTec......
  • RocketMQ 重试机制详解及最佳实践
    作者:斜阳引言本文主要介绍在使用RocketMQ时为什么需要重试与兜底机制,生产者与消费者触发重试的条件和具体行为,如何在RocketMQ中合理使用重试机制,帮助构建弹性,高可用......
  • 解析 RocketMQ 多样消费功能-消息过滤
    作者:徒钟什么是消息过滤在消息中间件的使用过程中,一个主题对应的消费者想要通过规则只消费这个主题下具备某些特征的消息,过滤掉自己不关心的消息,这个功能就叫消息过滤。......
  • 初识RocketMQ基础概念(一)
    20年入职ryx公司后,刚好接触到一个线上问题,遂开了一个研究课题,一场针对ApacheRocketMQ的无烟战争正式打开。一开始刚接触这玩意,只是通过百度搜索简单了解下,后来发现很多文......
  • RocketMQ 5.0 API 与 SDK 的演进
    作者:艾阳坤RocketMQ5.0SDK采用了全新的API,使用gRPC作为通信层的实现,并在可观测性上做了很大幅度的提升。全新统一的API此处的API并不单单只是接口上的定义,同......
  • 一站式解决Node项目中遇到的 诸如sharp Command failed.或Building fresh packages...
    一站式解决Node项目中遇到的诸如sharp:Commandfailed.或Buildingfreshpackages...始终执行问题1.问题的形式形式一:Buildingfreshpackages...1.[-/5]⠐waiti......
  • 让数据流动起来,RocketMQ Connect 技术架构解析
    作者:周波WhyRocketMQConnect在业务系统,或者大数据系统中不同数据源之间的数据同步是十分常见的,传统的点对点的数据同步工具,在面临越来越多的数据源点对点的数据同步会......
  • rocketMq springboot2 发送广播消息
    广播消息:一个点发送,所有有监听订阅的程序都能收到消息。应用场景:一个配置更新了,其他点都需要知道配置更新需加载。 mq创建主要是创建组时与队列有点区别mqadminup......