文档首页/ 分布式消息服务RocketMQ版/ 开发指南/ Java（TCP协议）/ 消费端限流
更新时间：2024-08-02 GMT+08:00
查看PDF
分享

消费端限流

在分布式消息服务RocketMQ版中，消费者消费消息时，可能会出现消费过快导致下游业务来不及处理的情况，进而影响系统的稳定性。本章节介绍在消费端进行限流的示例代码，以保障系统的稳定。

准备环境

开源的Java客户端支持连接分布式消息服务RocketMQ版，推荐使用的客户端版本为4.9.8。

使用Maven方式引入依赖。 
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.8</version>
</dependency>

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-acl</artifactId>
    <version>4.9.8</version>
</dependency>

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>29.0-jre</version>
</dependency>

示例代码

package org.apache.rocketmq.example.simple;

import java.util.List;
import java.util.concurrent.TimeUnit;

import com.google.common.util.concurrent.RateLimiter;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

public class PushConsumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
        //填入连接地址
        consumer.setNamesrvAddr("192.168.0.1:8100");
        consumer.subscribe("TopicTest", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        RateLimiter rateLimiter = RateLimiter.create(200);
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                if (!rateLimiter.tryAcquire(msgs.size(),3, TimeUnit.SECONDS)) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

父主题： Java（TCP协议）

相关文档