java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > RabbitMQ消费端限流

RabbitMQ实现消费端限流的步骤

作者:思静语

消费者端限流的主要目的是控制消费者每次从 RabbitMQ 中获取的消息数量,从而实现消息处理的流量控制,这篇文章主要介绍了RabbitMQ如何实现消费端限流,需要的朋友可以参考下

概述

在 RabbitMQ 中,可以通过消费者端限流(Consumer Prefetch)来控制消费端处理消息的速度,以避免消费端处理能力不足或处理过慢而导致消息堆积。消费者端限流的主要目的是控制消费者每次从 RabbitMQ 中获取的消息数量,从而实现消息处理的流量控制。
RabbitMQ 提供了一种 QOS(服务质量保证)功能,即在非自动确认消息的前提下,如果一定数目的消息还未被消费确认,则不进行新消息的消费。
RabbitMQ 为我们提供了三种机制
● 对内存和磁盘使用量设置阈值
● 于credit flow 的流控机制
● QoS保证机制
channel.basicQos()
channel.basicQos(int prefetchSize,int prefetchCount,boolean global)
一定要注意的是,如果做限流,那么no_ask是要设置为false,也就是手工签收而不是自动签收的情况下才可以做限流。
参数:
prefetchSize:消息的大小
prefetchCount:会告诉RabbitMQ不要同时给一个消费者推送多于N个消息,即一旦有N个消息还没有ack,则该consumer将block掉,直到有消息ack
global:是否将上面设置应用于channel,简单点说,就是上面限制是channel级别的还是consumer级别
注意,prefetchSize和golobal参数还没有实现。
Channel的详细介绍:
ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。
Connection是RabbitMQ的socket链接,它封装了socket协议相关部分逻辑。
ConnectionFactory如名称,是客户端与broker的tcp连接工厂,负责根据uri创建Connection。
Channel是我们与RabbitMQ打交道的最重要的一个接口,我们大部分的业务操作是在Channel这个接口中完成的,包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。如果每一次访问RabbitMQ都建立一个Connection,在消息量大的时候建立TCP Connection的开销将是巨大的,效率也较低。Channel是在connection内部建立的逻辑连接,如果应用程序支持多线程,通常每个thread创建单独的channel进行通讯,AMQP method包含了channel id帮助客户端和message broker识别channel,所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。
注意,rabbitmq提供了服务质量保障功能,即在非自动确认消息的前提下,如果一定数目的消息未被确认,不进行消费新的消息。也就是说,我们要使用非自动ack

@Configuration
public class DirectRabbitConfig {
    public static final String DEAD_LETTER_EXCHANGE = "dead.latter.exchange";
    public static final String DEAD_LETTER_QUEUE = "dead.latter.queue";
    public static final String DEAD_LETTER_KEY = "dead.latter.key";
    //队列 起名:TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        //实例化队列时各个参数的含义如下:
        //name:队列名称
        // durable:是否持久化,默认是false,持久化队列:会被存储在磁盘上,当消息代理重启时仍然存在,暂存队列:当前连接有效
        // exclusive:默认也是false,只能被当前创建的连接使用,而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:默认是false,是否自动删除,当没有生产者或者消费者使用此队列,该队列会自动删除。
        //一般设置一下队列的持久化就好,其余两个就是默认false(消息和交换机也可以持久化,但是消息持久化的前提是需要和队列,交换机持久化一起使用)
        //为业务队列绑定一个死信交换机,当业务队列里的消息过期了就被转发到死信交换机,再由死信交换机发给死信队列处理
//        Map<String, Object> args = new HashMap<>(2);
//       x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
//        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
//       x-dead-letter-routing-key  这里声明当前队列的死信路由key
//        args.put("x-dead-letter-routing-key", DEAD_LETTER_KEY);
        //5000毫秒
//        args.put("x-message-ttl", 5000);
//        return new Queue("TestDirectQueue",false,false,false,args);
        return new Queue("TestDirectQueue",false,false,false);
    }
    //Direct交换机 起名:TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange",true,false);
    }
    //绑定  将队列和交换机绑定, 并设置用于匹配键:TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
    /**
     * ========================死信队列==================================
     */
    // 声明死信Exchange
    @Bean
    public DirectExchange deadLetterExchange(){
        return new DirectExchange(DEAD_LETTER_EXCHANGE);
    }
    // 声明死信队列A
    @Bean
    public Queue deadLetterQueueA(){
        return new Queue(DEAD_LETTER_QUEUE);
    }
    @Bean
    public Binding deadLetterBindingA(){
        return BindingBuilder.bind(deadLetterQueueA()).to(deadLetterExchange()).with(DEAD_LETTER_KEY);
    }
}
@Component
@RabbitListener(queues = "TestDirectQueue")//监听的队列名称 TestDirectQueue
public class DirectReceiver {
     Integer index = 0;
    @RabbitHandler
    public void process(Channel channel,String msg) throws Exception {
        ++index;
        channel.basicQos(0, 1, false);
        //设置非自动ack
        DefaultConsumer consumer = new DefaultConsumer(channel);
        channel.basicConsume("TestDirectQueue", false,consumer);
        //假设业务处理需要3秒,那么当消费者接受到消息的时候,只处理一条,且要处理3秒,那么在服务器堆积的多条信息就不会疯狂涌入
        Thread.sleep(3000);
        System.out.println(" DirectReceiver消费者收到消息  : " + msg + ",第" +index+ "条"+"======"+ new Date().toString());
    }
}

RabbitMQ 中实现消费端限流的步骤

1.设置消费者端的预取值(Prefetch Count):
在创建消费者时,可以通过设置 basicQos(prefetchCount) 方法来指定消费者端的预取值,即每次从 RabbitMQ 中预取的消息数量。

2.确保消费者端开启手动应答模式:
在设置预取值之前,确保消费者端已经开启了手动应答模式(manual ack mode),这样消费者可以自主控制何时应答消息。

3.消费者端处理消息时进行手动应答:
当消费者端接收到消息后,在处理完消息之后,需要显式地发送应答(ack)给 RabbitMQ,表示该消息已经被消费。这样,消费者才能继续接收下一批消息。
下面是一个简单的 Java 示例代码,演示了如何在 RabbitMQ 中实现消费端限流:

import com.rabbitmq.client.*;
public class Consumer {
    private final static String QUEUE_NAME = "queue_name";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        int prefetchCount = 5; // 设置预取值为 5
        channel.basicQos(prefetchCount);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received: " + message);
            // 模拟消息处理
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // 手动应答消息
        };
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });
    }
}

在上述代码中,我们首先创建了连接和信道,并声明了一个名为 “queue_name” 的队列。然后,通过 channel.basicQos(prefetchCount) 方法设置了消费者端的预取值为 5。接着,我们定义了一个 DeliverCallback 回调函数,在其中处理消息并手动应答。最后,通过 channel.basicConsume() 方法启动消费者端。
通过设置预取值和手动应答,消费者端可以控制自身处理消息的速度,有效地实现消费端的限流。希望这个示例能帮助您理解如何在 RabbitMQ 中实现消费端限流!

到此这篇关于RabbitMQ如何实现消费端限流的文章就介绍到这了,更多相关RabbitMQ消费端限流内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文