解锁新姿势 | 如何用配置中心实现全局动态流控?

本文涉及的产品
Serverless 应用引擎 SAE,800核*时 1600GiB*时
云原生网关 MSE Higress,422元/月
注册配置 MSE Nacos/ZooKeeper,118元/月
简介: 当资源成为瓶颈时,服务框架需要对消费者做限流,启动流控保护机制。流量控制有多种策略,比较常用的有:针对访问速率的静态流控、针对资源占用的动态流控、针对消费者并发连接数的连接控制和针对并行访问数的并发控制。在分布式架构中,应用和应用之间的调用类型分为以下两种,流控方式也略有不同。

当资源成为瓶颈时,服务框架需要对消费者做限流,启动流控保护机制。流量控制有多种策略,比较常用的有:针对访问速率的静态流控、针对资源占用的动态流控、针对消费者并发连接数的连接控制和针对并行访问数的并发控制。在实践中,各种流量控制策略需要综合使用才能起到较好的效果。

在分布式架构中,应用和应用之间的调用类型分为以下两种,流控方式也略有不同。

同步RPC类调用,比如RESTful,Dubbo,HSF等都属于该类。对于该类同步调用,通常限流方式为两种:针对服务提供者的并发全局流控,或针对服务消费者的并发局部流控。两种的控制手段类似,都是通过限制服务端或客服端并发调用数来进行限制。

异步MQ类调用,典型如RocketMQ, Kafka,等。对于该类异步调用,通常限流方式是在订阅端限流。限流方式为两种:针对消息订阅者的并发流控,或针对消息订阅者的消费延时流控。

针对消息订阅者的消费延时流控基本原理是,在每次客户端消费时,可以增加一个延时来控制消费速度,这样理论消费并发最快速度为:

MaxRate = 1 / ConsumInterval * ConcurrentThreadNumber

比如如果消息并发消费线程为20,延时为100ms,则理论上可以将并发消费控制在200以下。具体公式如下:

200 = 1 / 0.1 * 20

相比并发线程数流控,消费延时流控优点在于实现相对简单,对MQ类客户端包依赖较少,不需要客户端提供控制并发线程数的动态调整接口。

以上各种流量控制方法,在分布式架构下,如果要做到全局动态控制,一个简单的技术方法是依赖配置中心,即通过配置中心来进行流控参数的下发。

下面章节详细介绍如何基于配置中心来实现异步消息消费的全局动态流控。使用的例子为阿里云上的 MQ (消息队列)和 ACM (应用配置管理)两款产品。

注:之所以用MQ为示例是因为在本文撰写之时,正好MQ Consumer Client SDK并不支持动态调整现成并发数,因此通过基于ACM来动态调整消费延迟的方法正好可以解决MQ消费流控动态的问题。

基于消费延时流控的基本原理

基本原理如下。其中,管理员或应用程序通过ACM控制台发布消费延时配置(RCV_INTERVAL_TIME),所有MQ消费程序订阅该配置。理论上,该配置从发布到下发所有客户端,可以在1秒内完成(取决于网络延时)。

代码示例

该章节基于配置中心来实现异步消息消费的全局动态流控的代码示例。使用的例子为阿里云上的MQ(消息队列)和ACM(应用配置管理)两款产品,基于Java语言。关于SDK的详细介绍,可参见两款产品的官方文档。

在ACM上创建消费延时的参数,截屏如下。

设置全局消费延时变量

首先,设置消费接收延时的全局变量, 如下。

     // 初始化消息接收延时参数,单位为millisecond

        static int RCV_INTERVAL_TIME = 10000;

        // 初始化配置服务,控制台通过示例代码自动获取下面参数

        ConfigService.init("acm.aliyun.com", /*租户ID*/"xxx", /*AK*/"xxx", /*SK*/"yyy");    

        // 主动获取配置

        String content = ConfigService.getConfig("app.mq.qos", "DEFAULT_GROUP", 6000);

        Properties p = new Properties();

        try {

            p.load(new StringReader(content));

            RCV_INTERVAL_TIME = Integer.valueOf(p.getProperty("RCV_INTERVAL_TIME"));

        } catch (IOException e) {

            e.printStackTrace();

        }

其次,设置ACM listener,确保当配置被修改时,即使更新 RCV_INTERVAL_TIME 参数, 如下。

     // 初始化的时候,给配置添加监听,配置变更会回调通知

        ConfigService.addListener("app.mq.qos", "DEFAULT_GROUP", new ConfigChangeListener() {

            public void receiveConfigInfo(String configInfo) {

                Properties p = new Properties();

                try {

                    p.load(new StringReader(configInfo));

                    RCV_INTERVAL_TIME = Integer.valueOf(p.getProperty("RCV_INTERVAL_TIME"));

                } catch (IOException e) {

                    e.printStackTrace();

                }

            }

        });

设置 MQ 消费延时逻辑

完整实例如下。

注:这里 RCV_INTERVAL_TIME 参数的访问是故意没有加锁的,读者可以自行思考原因。Aliyun ONS Client不提供动态线程并发数,默认并发为20。因此这里正好使用消费延时参数来动态调节QoS。

     //以下代码可直接贴在Main()函数里

    Properties properties = new Properties();

    properties.put(PropertyKeyConst.ConsumerId, "CID_consumer_group");

    properties.put(PropertyKeyConst.AccessKey,"xxx");

    properties.put(PropertyKeyConst.SecretKey, "yyy");

    properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");

    // 设置 TCP 接入域名(此处以公共云生产环境为例)

    properties.put(PropertyKeyConst.ONSAddr,

      "http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet");

    Consumer consumer = ONSFactory.createConsumer(properties);

    consumer.subscribe(/*Topic*/"topic-name", /*Tag*/null, new MessageListener() 

    {

        public Action consume(Message message, ConsumeContext context) {

            // MQ Subscribe QoS logical start, 

            // Each consuming process will sleep for RCV_INTERVAL_TIME seconds with 100 ms sleeping cycle.

            // Within each cycle, the thread will check RCV_INTERVAL_TIME in case it's set to a smaller value. 

            // RCV_INTERVAL_TIME <= 0 means no sleeping.

            int rcvIntervalTimeLeft = RCV_INTERVAL_TIME;

            while (rcvIntervalTimeLeft > 0) {

                if (rcvIntervalTimeLeft > RCV_INTERVAL_TIME) {

                    rcvIntervalTimeLeft = RCV_INTERVAL_TIME;

                }

                try {

                    if (rcvIntervalTimeLeft >= 100) {

                        rcvIntervalTimeLeft -= 100;

                        Thread.sleep(100);

                    } else {

                        Thread.sleep(rcvIntervalTimeLeft);

                        rcvIntervalTimeLeft = 0;

                    }

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

            }

            // MQ Subscribe interval logical ends

            System.out.println("Receive: " + message);

            /*

             * Put your business logic here.

             */

            doSomething();

            return Action.CommitMessage;

        }

    });

    consumer.start();

运行结果

单机运行consumer进行消费,假设queue内的消息无限多,不存在消费万的情况,分三段测试,分别运行约5分钟,通过ACM配置推送来达到以下效果。

RCV_INTERVAL_TIME = 100 ms

RCV_INTERVAL_TIME = 5000 ms

RCV_INTERVAL_TIME = 1000 ms

结果如下,在单MQ消费业务处理耗时约100ms情况下的,单机并发20线程的测试结果。

RCV_INTERVAL_TIME = 100 ms:平均消费性能约为 9000 tpm 左右

RCV_INTERVAL_TIME = 5000 ms:平均消费性能被限制到了 200 tpm 左右

RCV_INTERVAL_TIME = 1000 ms:平均消费性能回升到到了 1100 tpm 左右

以上结果基本达到消费和 tpm 成反比的预期,最关键的是整个过程中,应用不中断,流控推送结果秒级生效到分布式集群。单机性能结果如下所示。

相关产品详情请参见:

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
域名解析 Kubernetes Java
图文详述Nacos配置中心使用:应用间配置共享、扩展配置文件加载优先级、新老版本差异
图文详述Nacos配置中心使用:应用间配置共享、扩展配置文件加载优先级、新老版本差异
4015 1
图文详述Nacos配置中心使用:应用间配置共享、扩展配置文件加载优先级、新老版本差异
|
2月前
|
微服务
微服务多机房部署大揭秘:全局单一实例、全局多实例,一文让你彻底解锁!
【8月更文挑战第25天】本文探讨了微服务架构中的多机房部署策略,包括全局单一与多实例、区域及机房多实例等方法,分析了它们在可用性、容错性、扩展性和成本上的差异。示例展示了如何利用AWS CloudFormation实现跨不同机房的微服务部署。这为实际应用场景提供了有价值的参考和指导。
48 2
|
4月前
|
运维 Serverless API
Serverless 应用引擎产品使用合集之在创建或修改熔断规则时,没看到地方可以选行,是什么原因
阿里云Serverless 应用引擎(SAE)提供了完整的微服务应用生命周期管理能力,包括应用部署、服务治理、开发运维、资源管理等功能,并通过扩展功能支持多环境管理、API Gateway、事件驱动等高级应用场景,帮助企业快速构建、部署、运维和扩展微服务架构,实现Serverless化的应用部署与运维模式。以下是对SAE产品使用合集的概述,包括应用管理、服务治理、开发运维、资源管理等方面。
|
5月前
|
移动开发 前端开发
基于flowable没有规则的并发网关流程跳转记录分析
基于flowable没有规则的并发网关流程跳转记录分析
95 0
|
5月前
|
Kubernetes 算法 NoSQL
动态扩缩容下的全局流水号设计
该文介绍了在动态扩缩容场景下如何使用雪花算法生成全局流水号。雪花算法生成的ID由时间戳、工作机器ID和序列号组成。在K8s环境中,通过Redis存储当前workerId的最大值,每次生成时加1并取模,确保workerId在0-1023范围内。文中提供了实现雪花算法的`SnowflakeIdWorker`类示例,并展示了两种动态获取workerId的方法:一是利用Redis incr操作;二是通过Nacos服务发现获取IP和端口信息计算。此外,还提到了其他获取workId和dataCenterId的策略,如使用本地IP和主机名。
102 1
|
12月前
|
Java 数据库 开发者
自动配置要点解读
自动配置要点解读
|
监控 Dubbo Linux
微服务 热点流控 规则-授权 系统规则 自定义返回
微服务 热点流控 规则-授权 系统规则 自定义返回
93 0
|
负载均衡 网络协议 NoSQL
Envoy架构概览(10):热启动,动态配置,初始化,排水,脚本
Envoy架构概览(10):热启动,动态配置,初始化,排水,脚本
|
Prometheus Kubernetes Cloud Native
Flagger(应用自动发布)介绍和原理剖析
## 简介 [Flagger](https://github.com/weaveworks/flagger)是一个能使运行在k8s体系上的应用发布流程全自动(无人参与)的工具, 它能减少发布的人为关注时间, 并且在发布过程中能自动识别一些风险(例如:RT,成功率,自定义metrics)并回滚. ## 主要特性 ![features](https://intranetproxy.ali
4443 0
|
Java 开发者 Spring
项目启动加载限流规则代码|学习笔记
快速学习项目启动加载限流规则代码