RabbitMQ任务分发机制

灰太狼 2021-12-22 07:05 255阅读 0赞

### Round-robin（轮询分发） ###

使用任务队列的优点之一就是可以轻易的并行工作。如果我们积压了好多工作，我们可以通过增加工作者（消费者）来解决这一问题，使得系统的伸缩性更加容易。

修改一下NewTask，使用for循环模拟多次发送消息的过程：

for (int i = 0; i < 5; i++) {
    	// 发送的消息
    	String message = "Hello World"+Strings.repeat(".", i);
    	// 往队列中发出一条消息
    	channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    	System.out.println(" [x] Sent '" + message + "'");
    }

我们先启动1个生产者实例，2个工作者实例，看一下如何执行：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d3ZDA1MDE_size_16_color_FFFFFF_t_70][]

从上述的结果中，我们可以得知，在**默认情况下，RabbitMQ不会顾虑消息者处理消息的能力**，即使其中有的消费者闲置有的消费者高负荷。RabbitMQ会逐个发送消息到在序列中的下一个消费者(而不考虑每个任务的时长等等，且是提前一次性分配，并非一个一个分配)。平均每个消费者获得相同数量的消息，这种方式分发消息机制称为**Round-Robin（轮询）**。

### Fair dispatch（公平分发） ###

您可能已经注意到，任务分发仍然没有完全按照我们想要的那样。比如：现在有2个消费者，所有的奇数的消息都是繁忙的，而偶数则是轻松的。按照轮询的方式，奇数的任务交给了第一个消费者，所以一直在忙个不停。偶数的任务交给另一个消费者，则立即完成任务，然后闲得不行。而RabbitMQ则是不了解这些的。这是因为当消息进入队列，RabbitMQ就会分派消息。它不看消费者为应答的数目，只是盲目的将第n条消息发给第n个消费者。

**公平分发，则是根据消费者的处理能力来进行分发处理的。**这里主要是通过设置prefetchCount 参数来实现的。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理规定的数量级个数的Message。换句话说，在接收到该Consumer的ack前，它不会将新的Message分发给它。 比如prefetchCount=1，则在同一时间下，每个Consumer在同一个时间点最多处理1个Message，同时在收到Consumer的ack前，它不会将新的Message分发给它。

![20190629121224635.png][]

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

int prefetchCount = 1;
    channel.basicQos(prefetchCount);

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

整体代码如下：生产者NewTask.java

public class NewTask {
    	private final static String QUEUE_NAME = "hello";
     
    	public static void main(String[] args) throws IOException {
    		/**
    		 * 创建连接连接到MabbitMQ
    		 */
    		ConnectionFactory factory = new ConnectionFactory();
    		// 设置MabbitMQ所在主机ip或者主机名
    		factory.setHost("127.0.0.1");
    		// 创建一个连接
    		Connection connection = factory.newConnection();
    		// 创建一个频道
    		Channel channel = connection.createChannel();
    		// 指定一个队列
    		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    		int prefetchCount = 1;
    		//限制发给同一个消费者不得超过1条消息
    		channel.basicQos(prefetchCount);
    		for (int i = 0; i < 5; i++) {
    			// 发送的消息
    			String message = "Hello World"+Strings.repeat(".",5-i)+(5-i);
    			// 往队列中发出一条消息
    			channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    			System.out.println(" [x] Sent '" + message + "'");
    		}
    		// 关闭频道和连接
    		channel.close();
    		connection.close();
    	}
    }

消费者Worker.java

public class Worker {
    	private final static String QUEUE_NAME = "hello";
     
    	public static void main(String[] argv) throws IOException, InterruptedException {
    		ConnectionFactory factory = new ConnectionFactory();
    		factory.setHost("127.0.0.1");
    		// 打开连接和创建频道，与发送端一样
    		Connection connection = factory.newConnection();
    		final Channel channel = connection.createChannel();
     
    		// 声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
    		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    		channel.basicQos(1);//保证一次只分发一个
    		// 创建队列消费者
    		final Consumer consumer = new DefaultConsumer(channel) {
    			  @Override
    			  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    			    String message = new String(body, "UTF-8");
     
    			    System.out.println(" [x] Received '" + message + "'");
    			    try {
    			    	for (char ch: message.toCharArray()) {
    				        if (ch == '.') {
    				        	Thread.sleep(1000);
    				        }
    				    }
    				} catch (InterruptedException e) {
    				} finally {
    			      System.out.println(" [x] Done! at " +new Date().toLocaleString());
    			      channel.basicAck(envelope.getDeliveryTag(), false);  
    			    }
    			  }
    			};
    			channel.basicConsume(QUEUE_NAME, false, consumer);
    	}
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d3ZDA1MDE_size_16_color_FFFFFF_t_70 1][]

文章来源：[https://blog.csdn.net/a724888/article/details/80735185][https_blog.csdn.net_a724888_article_details_80735185]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d3ZDA1MDE_size_16_color_FFFFFF_t_70]: /images/20211220/4fad499bfe994e49a4f6c9bcc9122702.png
[20190629121224635.png]: /images/20211220/4c93329a49b64340aab6ddc0966dac71.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d3ZDA1MDE_size_16_color_FFFFFF_t_70 1]: /images/20211220/321197fd40524978b7bca310b7b80a94.png
[https_blog.csdn.net_a724888_article_details_80735185]: https://blog.csdn.net/a724888/article/details/80735185