526互联

rabbitmq不公平分发

发布时间 2023-09-01 20:15:10作者: 自学Java笔记本

前言

对于mq来说，默认采用的是轮询的消息发送模式，例如一个生产者对接多个消费者，那么发送消息时，消费者1接收一个消息后就轮到了消费者2接收，依次类推大致的算法就是取模的方式。
但是在某种场景下这种策略并不是 很好，比方说有两个消费者在处理任务，其中有个消费者 1 处理任务的速度非常快，而另外一个消费者 2 处理速度却很慢，这个时候我们还是采用轮训分发的化就会到这处理速度快的这个消费者很大一部分时间 处于空闲状态，而处理慢的那个消费者一直在干活，这种分配方式在这种情况下其实就不太好，但是 RabbitMQ 并不知道这种情况它依然很公平的进行分发。

为了避免这种情况，我们可以设置参数 channel.basicQos(1); 这是消费者代码设置的
不公平分发，来让消费速度快的消费者消费更多的消息。意思就是如果这个任务我还没有处理完或者我还没有应答你，你先别分配给我，我目前只能处理一个任务，然后 rabbitmq 就会把该任务分配给没有那么忙的那个空闲消费者。

 channel.basicQos(1); // 表示不公平分发，意思是只能处理一个任务，其余的任务给其他的消费者
 channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);//发送消息

案例

场景说明，现在有一个生产者，两个消费者，生产者发送多条消息到消费者中，消费者从mq中获取消息

  • 消费者1处理业务耗时1s
  • 消费者2处理业务耗时10s
  • 需求：采用不公平分发，能者多劳模式

生产者

package producer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import utils.RabbitMQUtil;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 不公平分发
 */
public class ProducerQueueThreeImbalance {
    // 队列名称
    public static final String  QUEUE_NAME = "imbalance:";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMQUtil.getChannel();
        /**
         * 生成一个队列
         * 1、队列名称
         * 2、队列里面的消息是否持久化  默认情况下 消息存储在内存中。
         * 3、该队列是否只供一个消费者进行消费，是否进行消息的共享，true--可以多个消费者消费。反之不可（默认）
         * 4、表示是否自动删除，最后一个消费者端开连接以后，该队列是否自动删除，true----自动删除，反之不删除
         * 5、其他参数；例如，延迟消息，死期消息等
         */
        boolean isLasting = true;
        channel.queueDeclare(QUEUE_NAME,isLasting,false,false,null);
        String message = "第%s次发送的消息";

        //发送消息
        /**
         * 发送一个消息
         * 1、发送到那个交换机-->null 为默认交换机
         * 2、路由的key值是哪个，本次是队列的名称
         * 3、其他参数信息
         * 4、发送消息的消息体
         */
        for (int i = 0 ; i < 10; i++){
            String strMessage = String.format(message, i);

            channel.basicPublish("",QUEUE_NAME, MessageProperties.MINIMAL_PERSISTENT_BASIC,strMessage.getBytes());
            System.out.println(strMessage);
        }

        System.out.println("消息全部发送成功");
    }
}

消费者1

package consumer;

import com.rabbitmq.client.*;
import utils.RabbitMQUtil;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 消费者
 */
public class ConsumerThreeImbalance {
    // 队列名称
    public static final String  QUEUE_NAME = "imbalance:";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.getChannel();
        // 声明 DeliverCallvack 接收消息 的回调
        /**
         * consumerTag–与消费者关联的消费者标签
         * message–传递的消息
         */
        DeliverCallback deliverCallback = (consumerTag, message)->{
            try {
                // 模拟业务处理耗时1s
                Thread.sleep(1000);
                //接收消息
                System.out.println(new String(message.getBody()));
                // 手动应答-->message.getEnvelope().getDeliveryTag() 取回消息,消息的标识
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };


        //CancelCallback  声明 取消消息的回调
        CancelCallback cancelCallback = consumerTag->{
            System.out.println("消息消费被中断了。。。。");
        };


        /**
         * 消费者接收消息
         * 1、消费哪个队列
         * 2、消费成功之后，是否要自动应答  true---自动应答
         * 3、DeliverCallvack： 消费者未成功消费的回调   --- 是一个接口，需要实现， 或者通过lambad表达式
         * 4、消费者取消消费的回调     --- 是一个接口，需要实现， 或者通过lambad表达式 - -- 正常接收是没有意义的
         */
        channel.basicQos(1); // 表示不公平分发，意思是只能处理一个任务，其余的任务给其他的消费者
        channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);

    }
}

消费者2

与消费者1相同，唯一不同点在于休眠时间为10s

// 模拟业务处理耗时1s
Thread.sleep(10000);

结果展示

生产者发送10条消息，全部由消费者1消费了
image