RocketMQ 消费端限流

我会带着你远行 2022-09-04 15:55 473阅读 0赞

## RocketMQ 消费端限流 ##

RocketMQ 消费端限流  
首先我们要明白为什么需要限流？如果不使用限流呢？

通常情况下，当客户端生产的消息很多时，消费者消费消息速度低于生产者消费速度，我们该如何解决？

*  可以增加消费者数量
 *  可以优化消费者程序，提高消费响应速度，从而提高消费者处理能力。
 *  设置消费最大线程数
 *  拉取消息数量

这几种方法虽然能提提高消费速度，但是再特定场合下，生产者生产消息速度指数级上升，消费速度就会远远低于这个速度，就会出现大量的消息积压，导致消费能力下降，消息不停地推送，消费者消费能力有限，会出现什么情况。

在 Apache RocketMQ 中，当消费者去消费消息的时候，无论是通过 pull 的方式还是 push 的方式，都可能会出现大批量的消息突刺。如果此时要处理所有消息，很可能会导致系统负载过高，影响稳定性。但其实可能后面几秒之内都没有消息投递，若直接把多余的消息丢掉则没有充分利用系统处理消息的能力。我们希望可以把消息突刺均摊到一段时间内，让系统负载保持在消息处理水位之下的同时尽可能地处理更多消息，从而起到“削峰填谷”的效果：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]  
那如何来处理这种消息呢，RocketMQ使用 Sentinel来进行消息限流，如果消息生产速度很大， Sentinel就会限制大量的消息，推送一小部门消息给消费者消费，消费完再推送，这种很匀速的推送，不会使得消费端出现系统负载，从而保证系统的稳定。

## Sentinel介绍 ##

Sentinel 是阿里中间件团队开源的，面向分布式服务架构的轻量级流量控制产品，主要以流量为切入点，从流量控制、熔断降级、系统负载保护等多个维度来帮助用户保护服务的稳定性。

Sentinel 专门为这种场景提供了匀速器的特性，可以把突然到来的大量请求以匀速的形式均摊，以固定的间隔时间让请求通过，以稳定的速度逐步处理这些请求，起到“削峰填谷”的效果，从而避免流量突刺造成系统负载过高。同时堆积的请求将会排队，逐步进行处理；当请求排队预计超过最大超时时长的时候则直接拒绝，而不是拒绝全部请求。

可以查看官方文档  
[https://github.com/alibaba/Sentinel/wiki/Sentinel-%E4%B8%BA-RocketMQ-%E4%BF%9D%E9%A9%BE%E6%8A%A4%E8%88%AA][https_github.com_alibaba_Sentinel_wiki_Sentinel-_E4_B8_BA-RocketMQ-_E4_BF_9D_E9_A9_BE_E6_8A_A4_E8_88_AA]  
[https://github.com/alibaba/Sentinel/wiki/%E4%B8%BB%E9%A1%B5][https_github.com_alibaba_Sentinel_wiki_E4_B8_BB_E9_A1_B5]

比如在 RocketMQ 的场景下配置了匀速模式下请求 QPS 为 5，则会每 200 ms 处理一条消息，多余的处理任务将排队；同时设置了超时时间为 5 s，预计排队时长超过 5s 的处理任务将会直接被拒绝。示意图如下图所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]

Sentinel工作原理

topic中是我们生产的消息，消费者端从消费消息  
Sentinel会根据消息的qps 以及设置消费者每秒的处理速度，从topic中推送一部分消息到本地队列，消费者从本地队列消费消息，超过指定时间直接放弃消费，消费完再从topic重新获取一部分消息继续消费。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]  
这样在大量消息来到时候，消费端始终是按照一定得消费速度匀速消费消息，始终不会对消费端造成系统压力。

## RocketMQ Client 接入 Sentinel 的示例 ##

首先分别引入 RocketMQ和Sentinel依赖

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.5.1</version>
            </dependency>
    
            
            <dependency>
                <groupId>com.alibaba.csp</groupId>
                <artifactId>sentinel-core</artifactId>
                <version>1.7.2</version>
            </dependency>

新建一个生产者 发送1000条消息

public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException, MQBrokerException { 
            // 实例化生产者，并指定生产组名称
            DefaultMQProducer producer = new DefaultMQProducer("myproducer_group_topic_name_sentinle_01");
    
    
            //设置实例名称，一个jvm中有多个生产者可以根据实例名区分
            //默认default
            producer.setInstanceName("topic_sentinle");
    
            // 指定nameserver的地址
            producer.setNamesrvAddr("localhost:9876");
    
            //设置同步重试次数
            producer.setRetryTimesWhenSendFailed(2);
            //设置异步发送次数
            //producer.setRetryTimesWhenSendAsyncFailed(2);
    
    
            // 初始化生产者
            producer.start();
    
            for (int i = 0; i < 10000; i++) { 
                Message message = new Message("topic_name_sentinle", ("key=" + i).getBytes("utf-8"));
                // 1 同步发送 如果发送失败会根据重试次数重试
                SendResult send = producer.send(message);
                SendStatus sendStatus = send.getSendStatus();
                System.out.println(sendStatus.toString());
    
            }
    
            // 关闭生产者
            producer.shutdown();
    
        }

使用Sentinel来控制消息速度匀速消费

// 消费组名称
        private static final String GROUP_NAME = "consumer_grp_sentinle_01";
        // 主题名称
        private static final String TOPIC_NAME = "topic_name_sentinle";
        // consumer_grp_13_01:tp_demo_13
        private static final String KEY = String.format("%s:%s", GROUP_NAME, TOPIC_NAME);
        // 使用map存放主题每个MQ的偏移量
        private static final Map<MessageQueue, Long> OFFSET_TABLE = new HashMap<MessageQueue, Long>();
    
        // 具有固定大小的线程池
        private static final ExecutorService pool = Executors.newFixedThreadPool(32);
    
        private static final AtomicLong SUCCESS_COUNT = new AtomicLong(0);
        private static final AtomicLong FAIL_COUNT = new AtomicLong(0);
    
        public static void main(String[] args) throws MQClientException { 
            // 初始化哨兵的流控
            initFlowControlRule();
    
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(GROUP_NAME);
            consumer.setNamesrvAddr("localhost:9876");
            consumer.start();
    
            Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(TOPIC_NAME);
            for (MessageQueue mq : mqs) { 
                System.out.printf("Consuming messages from the queue: %s%n", mq);
    
                SINGLE_MQ:
                while (true) { 
                    try { 
                        PullResult pullResult =
                                consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
                        if (pullResult.getMsgFoundList() != null) { 
                            for (MessageExt msg : pullResult.getMsgFoundList()) { 
                                doSomething(msg);
                            }
                        }
    
                        long nextOffset = pullResult.getNextBeginOffset();
                        // 将每个mq对应的偏移量记录在本地HashMap中
                        putMessageQueueOffset(mq, nextOffset);
                        consumer.updateConsumeOffset(mq, nextOffset);
                        switch (pullResult.getPullStatus()) { 
                            case NO_NEW_MSG:
                                break SINGLE_MQ;
                            case FOUND:
                            case NO_MATCHED_MSG:
                            case OFFSET_ILLEGAL:
                            default:
                                break;
                        }
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }
    
            consumer.shutdown();
        }
    
        /** * 对每个收到的消息使用一个线程提交任务 * * @param message */
        private static void doSomething(MessageExt message) { 
            pool.submit(() -> { 
                Entry entry = null;
                try { 
                    // 应用流控规则
                    ContextUtil.enter(KEY);
                    entry = SphU.entry(KEY, EntryType.OUT);
    
                    // 在这里处理业务逻辑，此处只是打印
                    System.out.printf("[%d][%s][消费成功: %d] 接收新消息 : %s %n", System.currentTimeMillis(),
                            Thread.currentThread().getName(), SUCCESS_COUNT.addAndGet(1), new String(message.getBody()));
                } catch (BlockException ex) { 
                    // Blocked.
                    System.out.println("Blocked: " + FAIL_COUNT.addAndGet(1));
                } finally { 
                    if (entry != null) { 
                        entry.exit();
                    }
                    ContextUtil.exit();
                }
            });
        }
    
        private static void initFlowControlRule() { 
            FlowRule rule = new FlowRule();
            // 消费组名称:主题名称 字符串
            rule.setResource(KEY);
            // 根据QPS进行流控
            rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
            // 1表示QPS为1，请求间隔1000ms。
            // 如果是5，则表示每秒5个消息，请求间隔200ms
            rule.setCount(5);
            rule.setLimitApp("default");
    
            // 调用使用固定间隔。如果qps为1，则请求之间间隔为1s
            rule.setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER);
            // 如果请求太多，就将这些请求放到等待队列中
            // 该队列有超时时间。如果等待队列中请求超时，则丢弃
            // 此处设置超时时间为5s
            rule.setMaxQueueingTimeMs(2 * 1000);
            // 使用流控管理器加载流控规则
            FlowRuleManager.loadRules(Collections.singletonList(rule));
        }
    
        // 获取指定MQ的偏移量
        private static long getMessageQueueOffset(MessageQueue mq) { 
            Long offset = OFFSET_TABLE.get(mq);
            if (offset != null) { 
                return offset;
            }
    
            return 0;
        }
    
        // 在本地HashMap中记录偏移量
        private static void putMessageQueueOffset(MessageQueue mq, long offset) { 
            OFFSET_TABLE.put(mq, offset);
        }

可以看到 消费一定数量消息时候，其他是处在阻塞状态，消费完了后再继续消费

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]: /images/20220829/dfaabfa6430441db9e3ed13f2a79ed1f.png
[https_github.com_alibaba_Sentinel_wiki_Sentinel-_E4_B8_BA-RocketMQ-_E4_BF_9D_E9_A9_BE_E6_8A_A4_E8_88_AA]: https://github.com/alibaba/Sentinel/wiki/Sentinel-%E4%B8%BA-RocketMQ-%E4%BF%9D%E9%A9%BE%E6%8A%A4%E8%88%AA
[https_github.com_alibaba_Sentinel_wiki_E4_B8_BB_E9_A1_B5]: https://github.com/alibaba/Sentinel/wiki/%E4%B8%BB%E9%A1%B5
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]: /images/20220829/ab2f721bde6c4adc9d1ddc7e5f09e4d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]: /images/20220829/092b51dfce79471581648ab6fa7341a1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 3]: /images/20220829/b0574c23d0ef407b960ae9b4ac711045.png