RabbitMQ消息分发模式----"Work Queues"模式

不念不忘少年蓝@ 2022-08-20 02:19 185阅读 0赞

**2、Workqueues工作队列模式：**

![Image 1][]

上篇博客我们完成了一个简单的对声明的队列进行发送和接受消息程序。下面我们将创建一个工作队列，来向多个工作者（consumer）分发耗时任务。

今天我们来看工作队列（又名：任务队列）。主要是为了避免立即做一个资源密集型的却又必须等待完成的任务。相反的，我们进行任务调度：将任务封装为消息并发给队列。在后台运行的工作者（consumer）将其取出，然后最终执行。当你运行多个工作者（consumer），队列中的任务被工作进行共享执行。

任务队列的消息分发机制分两种：轮询分发（Round-robin）和公平分发（Fair dispatch）。

**准备**  
       使用Thread.Sleep()方法来模拟耗时。

2.1轮询分发

使用任务队列的优点之一就是可以轻易的并行工作。如果我们积压了好多工作，我们可以通过增加工作者（消费者）来解决这一问题，使得系统的伸缩性更加容易。在默认情况下，RabbitMQ将逐个发送消息到在序列中的下一个消费者(而不考虑每个任务的时长等等，且是提前一次性分配，并非一个一个分配)。平均每个消费者获得相同数量的消息。这种方式分发消息机制称为Round-Robin（轮询）。

消息生产端代码：

package cn.rabbitmq.work;
    
    import cn.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /**
     * 工作队列模式生产者
     * 
     * @author Administrator
     *
     */
    public class WorkSend {
        private final static String QUEUE_NAME = "hellowork";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    <span style="white-space:pre">	</span>// 同一时刻服务器只会发一条消息给消费者
            // channel.basicQos(1);
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            for (int i = 0; i < 100; i++) {
                // 消息内容
                String message = "" + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
    
                Thread.sleep(i * 10);
            }
    
            channel.close();
            connection.close();
        }
    }

消息消费端代码（复制两份都启动）：

package cn.rabbitmq.work;
    
    import cn.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    /**
     * 工作队列模式消费者2
     * 
     * @author Administrator
     *
     */
    public class WorkRecv2 {
        private final static String QUEUE_NAME = "hellowork";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成状态
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                // 休眠1秒
                Thread.sleep(1000);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }  
             
    }

结果就是不管谁忙或清闲，都不会给谁多一个任务或少一个任务，任务总是你一个我一个的分配。  
2.1公平分发

虽然上面的分配法方式也还行，但是有个问题就是：比如：现在有2个消费者，所有的奇数的消息都是繁忙的，而偶数则是轻松的。按照轮询的方式，奇数的任务交给了第一个消费者，所以一直在忙个不停。偶数的任务交给另一个消费者，则立即完成任务，然后闲得不行。而RabbitMQ则是不了解这些的。这是因为当消息进入队列，RabbitMQ就会分派消息。它不看消费者为应答的数目，只是盲目的将消息发给轮询指定的消费者。

为了解决这个问题，我们使用basicQos( prefetchCount = 1)方法，来限制RabbitMQ只发不超过1条的消息给同一个消费者。当消息处理完毕后，有了反馈，才会进行第二次发送。

![Center][]

将上面消息生产端代码中注释掉下面这行代码放开注释，再次执行即可。

// channel.basicQos(1);

注意：如果所有的工作者都处于繁忙状态，你的队列有可能被填充满。你可能会观察队列的使用情况，然后增加工作者，或者使用别的什么策略。  
           还有一点需要注意，使用公平分发，必须关闭自动应答，改为手动应答。这些内容会在后面讲述。

[http://blog.csdn.net/xiaoxian8023/article/details/48681987][http_blog.csdn.net_xiaoxian8023_article_details_48681987]

[Image 1]: 
[Center]: /images/20220731/36575c5d178844688cf90a35ec85e9e4.png
[http_blog.csdn.net_xiaoxian8023_article_details_48681987]: http://blog.csdn.net/xiaoxian8023/article/details/48681987