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

     @Configurationpublic 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 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")//监听的队列名称 TestDirectQueuepublic 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 中实现消费端限流！