RabbitMQ （两）工作队列-阿里云开发者社区

转载请注明出处：http://blog.csdn.net/lmj623565791/article/details/37620057

本系列教程主要来自于官网新手教程的翻译，然后自己进行了部分的改动与实验，内容仅供參考。

上一篇博客中我们写了通过一个命名的队列发送和接收消息，假设你还不了解请点击：RabbitMQ 入门 Helloworld。这篇中我们将会创建一个工作队列用来在工作者（consumer）间分发耗时任务。

工作队列的主要任务是：避免立马运行资源密集型任务，然后必须等待其完毕。相反地，我们进行任务调度：我们把任务封装为消息发送给队列。工作进行在后台运行并不断的从队列中取出任务然后运行。当你运行了多个工作进程时，任务队列中的任务将会被工作进程共享运行。
这种概念在web应用中极事实上用，当在非常短的HTTP请求间须要运行复杂的任务。

1、准备

我们使用Thread.sleep来模拟耗时的任务。

我们在发送到队列的消息的末尾加入一定数量的点。每一个点代表在工作线程中须要耗时1秒。比如hello…将会须要等待3秒。

发送端：

NewTask.java

package com.zhy.rabbit._02_workqueue;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class NewTask
{
	//队列名称
	private final static String QUEUE_NAME = "workqueue";

	public static void main(String[] args) throws IOException
	{
		//创建连接和频道
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		//声明队列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		//发送10条消息，依次在消息后面附加1-10个点
		for (int i = 0; i < 10; i++)
		{
			String dots = "";
			for (int j = 0; j <= i; j++)
			{
				dots += ".";
			}
			String message = "helloworld" + dots+dots.length();
			channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
			System.out.println(" [x] Sent '" + message + "'");
		}
		//关闭频道和资源
		channel.close();
		connection.close();

	}


}

接收端：

Work.java

package com.zhy.rabbit._02_workqueue;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Work
{
	//队列名称
	private final static String QUEUE_NAME = "workqueue";

	public static void main(String[] argv) throws java.io.IOException,
			java.lang.InterruptedException
	{
		//区分不同工作进程的输出
		int hashCode = Work.class.hashCode();
		//创建连接和频道
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		//声明队列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		System.out.println(hashCode
				+ " [*] Waiting for messages. To exit press CTRL+C");
	
		QueueingConsumer consumer = new QueueingConsumer(channel);
		// 指定消费队列
		channel.basicConsume(QUEUE_NAME, true, consumer);
		while (true)
		{
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			String message = new String(delivery.getBody());

			System.out.println(hashCode + " [x] Received '" + message + "'");
			doWork(message);
			System.out.println(hashCode + " [x] Done");

		}

	}

	/**
	 * 每一个点耗时1s
	 * @param task
	 * @throws InterruptedException
	 */
	private static void doWork(String task) throws InterruptedException
	{
		for (char ch : task.toCharArray())
		{
			if (ch == '.')
				Thread.sleep(1000);
		}
	}
}

Round-robin 转发
使用任务队列的优点是可以非常easy的并行工作。假设我们积压了非常多工作。我们只通过添加很多其它的工作者就行解决这个问题。使系统的伸缩性更加easy。
以下我们先执行3个工作者（Work.java）实例，然后执行NewTask.java，3个工作者实例都会得到信息。可是怎样分配呢？让我们来看输出结果：[x] Sent 'helloworld.1'
[x] Sent 'helloworld..2'
[x] Sent 'helloworld...3'
[x] Sent 'helloworld....4'
[x] Sent 'helloworld.....5'
[x] Sent 'helloworld......6'
[x] Sent 'helloworld.......7'
[x] Sent 'helloworld........8'
[x] Sent 'helloworld.........9'
[x] Sent 'helloworld..........10'

工作者1：
605645 [*] Waiting for messages. To exit press CTRL+C
605645 [x] Received 'helloworld.1'
605645 [x] Done
605645 [x] Received 'helloworld....4'
605645 [x] Done
605645 [x] Received 'helloworld.......7'
605645 [x] Done
605645 [x] Received 'helloworld..........10'
605645 [x] Done

工作者2：
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld..2'
18019860 [x] Done
18019860 [x] Received 'helloworld.....5'
18019860 [x] Done
18019860 [x] Received 'helloworld........8'
18019860 [x] Done

工作者3：
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld...3'
18019860 [x] Done
18019860 [x] Received 'helloworld......6'
18019860 [x] Done
18019860 [x] Received 'helloworld.........9'
18019860 [x] Done
能够看到，默认的，RabbitMQ会一个一个的发送信息给下一个消费者(consumer)，而不考虑每一个任务的时长等等，且是一次性分配，并不是一个一个分配。平均的每一个消费者将会获得相等数量的消息。这样分发消息的方式叫做round-robin。

2、消息应答（message acknowledgments）

运行一个任务须要花费几秒钟。你可能会操心当一个工作者在运行任务时发生中断。我们上面的代码。一旦RabbItMQ交付了一个信息给消费者，会立即从内存中移除这个信息。在这样的情况下，假设杀死正在运行任务的某个工作者，我们会丢失它正在处理的信息。

我们也会丢失已经转发给这个工作者且它还未运行的消息。
上面的样例，我们首先开启两个任务。然后运行发送任务的代码（NewTask.java）。然后马上关闭第二个任务。结果为：
工作者2：

31054905 [*] Waiting for messages. To exit press CTRL+C
31054905 [x] Received 'helloworld..2'
31054905 [x] Done
31054905 [x] Received 'helloworld....4'

工作者1：
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld.1'
18019860 [x] Done
18019860 [x] Received 'helloworld...3'
18019860 [x] Done
18019860 [x] Received 'helloworld.....5'
18019860 [x] Done
18019860 [x] Received 'helloworld.......7'
18019860 [x] Done
18019860 [x] Received 'helloworld.........9'
18019860 [x] Done
能够看到，第二个工作者至少丢失了6，8，10号任务。且4号任务未完毕。

可是，我们不希望丢失不论什么任务（信息）。

当某个工作者（接收者）被杀死时，我们希望将任务传递给还有一个工作者。
为了保证消息永远不会丢失，RabbitMQ支持消息应答（message acknowledgments）。消费者发送应答给RabbitMQ，告诉它信息已经被接收和处理，然后RabbitMQ能够自由的进行信息删除。
假设消费者被杀死而没有发送应答，RabbitMQ会觉得该信息没有被全然的处理。然后将会又一次转发给别的消费者。

通过这样的方式，你能够确认信息不会被丢失，即使消者偶尔被杀死。

这样的机制并没有超时时间这么一说。RabbitMQ仅仅有在消费者连接断开是又一次转发此信息。假设消费者处理一个信息须要耗费特别特别长的时间是同意的。

消息应答默认是打开的。上面的代码中我们通过显示的设置autoAsk=true关闭了这样的机制。

以下我们改动代码（Work.java）：

boolean ack = false ; //打开应答机制
channel.basicConsume(QUEUE_NAME, ack, consumer);
//另外须要在每次处理完毕一个消息后。手动发送一次应答。
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

完整改动后的Work.java

package com.zhy.rabbit._02_workqueue.ack;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Work
{
	//队列名称
	private final static String QUEUE_NAME = "workqueue";

	public static void main(String[] argv) throws java.io.IOException,
			java.lang.InterruptedException
	{
		//区分不同工作进程的输出
		int hashCode = Work.class.hashCode();
		//创建连接和频道
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		//声明队列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
		System.out.println(hashCode
				+ " [*] Waiting for messages. To exit press CTRL+C");
		QueueingConsumer consumer = new QueueingConsumer(channel);
		// 指定消费队列
		boolean ack = false ; //打开应答机制
		channel.basicConsume(QUEUE_NAME, ack, consumer);
		while (true)
		{
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			String message = new String(delivery.getBody());

			System.out.println(hashCode + " [x] Received '" + message + "'");
			doWork(message);
			System.out.println(hashCode + " [x] Done");
			//发送应答
			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

		}

	}
}

測试：
我们把消息数量改为5。然后先打开两个消费者（Work.java）。然后发送任务（NewTask.java），马上关闭一个消费者，观察输出：
[x] Sent 'helloworld.1'
[x] Sent 'helloworld..2'
[x] Sent 'helloworld...3'
[x] Sent 'helloworld....4'
[x] Sent 'helloworld.....5'

工作者2
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld..2'
18019860 [x] Done
18019860 [x] Received 'helloworld....4'

工作者1
31054905 [*] Waiting for messages. To exit press CTRL+C
31054905 [x] Received 'helloworld.1'
31054905 [x] Done
31054905 [x] Received 'helloworld...3'
31054905 [x] Done
31054905 [x] Received 'helloworld.....5'
31054905 [x] Done
31054905 [x] Received 'helloworld....4'
31054905 [x] Done

能够看到工作者2没有完毕的任务4，又一次转发给工作者1进行完毕了。

3、消息持久化（Message durability）

我们已经学习了即使消费者被杀死。消息也不会被丢失。

可是假设此时RabbitMQ服务被停止，我们的消息仍然会丢失。

当RabbitMQ退出或者异常退出，将会丢失全部的队列和信息，除非你告诉它不要丢失。我们须要做两件事来确保信息不会被丢失：我们须要给全部的队列和消息设置持久化的标志。
第一，我们须要确认RabbitMQ永远不会丢失我们的队列。为了这样，我们须要声明它为持久化的。

boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);
注：RabbitMQ不同意使用不同的參数又一次定义一个队列。所以已经存在的队列，我们无法改动其属性。
第二。我们须要标识我们的信息为持久化的。通过设置MessageProperties（implements BasicProperties）值为PERSISTENT_TEXT_PLAIN。
channel.basicPublish("", "task_queue",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
如今你能够执行一个发送消息的程序。然后关闭服务，再又一次启动服务，执行消费者程序做下实验。

4、公平转发（Fair dispatch）

也许会发现。眼下的消息转发机制（Round-robin）并不是是我们想要的。比如，这样一种情况。对于两个消费者，有一系列的任务，奇数任务特别耗时，而偶数任务却非常轻松。这样造成一个消费者一直繁忙，还有一个消费者却非常快运行完任务后等待。

造成这种原因是由于RabbitMQ不过当消息到达队列进行转发消息。并不在乎有多少任务消费者并未传递一个应答给RabbitMQ。只盲目转发全部的奇数给一个消费者，偶数给还有一个消费者。
为了解决这种问题，我们能够使用basicQos方法，传递參数为prefetchCount = 1。这样告诉RabbitMQ不要在同一时间给一个消费者超过一条消息。

换句话说。仅仅有在消费者空暇的时候会发送下一条信息。

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

注：假设全部的工作者都处于繁忙状态，你的队列有可能被填充满。

你可能会观察队列的使用情况，然后添加工作者，或者使用别的什么策略。
測试：改变发送消息的代码。将消息末尾点数改为6-2个，然后首先开启两个工作者，接着发送消息：

[x] Sent 'helloworld......6'
[x] Sent 'helloworld.....5'
[x] Sent 'helloworld....4'
[x] Sent 'helloworld...3'
[x] Sent 'helloworld..2'

工作者1：
18019860 [*] Waiting for messages. To exit press CTRL+C
18019860 [x] Received 'helloworld......6'
18019860 [x] Done
18019860 [x] Received 'helloworld...3'
18019860 [x] Done

工作者2：
31054905 [*] Waiting for messages. To exit press CTRL+C
31054905 [x] Received 'helloworld.....5'
31054905 [x] Done
31054905 [x] Received 'helloworld....4'
31054905 [x] Done
31054905 [x] Received 'helloworld..2'
31054905 [x] Done

能够看出此时并没有依照之前的Round-robin机制进行转发消息。而是当消费者不忙时进行转发。

且这样的模式下支持动态添加消费者，由于消息并没有发送出去，动态添加了消费者立即投入工作。而默认的转发机制会造成，即使动态添加了消费者。此时的消息已经分配完毕。无法立即添加工作，即使有非常多未完毕的任务。

5、完整的代码

NewTask.java

package com.zhy.rabbit._02_workqueue.ackandpersistence;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

public class NewTask
{
	// 队列名称
	private final static String QUEUE_NAME = "workqueue_persistence";

	public static void main(String[] args) throws IOException
	{
		// 创建连接和频道
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		// 声明队列
		boolean durable = true;// 1、设置队列持久化
		channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
		// 发送10条消息，依次在消息后面附加1-10个点
		for (int i = 5; i > 0; i--)
		{
			String dots = "";
			for (int j = 0; j <= i; j++)
			{
				dots += ".";
			}
			String message = "helloworld" + dots + dots.length();
			// MessageProperties 2、设置消息持久化
			channel.basicPublish("", QUEUE_NAME,
					MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
			System.out.println(" [x] Sent '" + message + "'");
		}
		// 关闭频道和资源
		channel.close();
		connection.close();

	}

}

Work.java

package com.zhy.rabbit._02_workqueue.ackandpersistence;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Work
{
	// 队列名称
	private final static String QUEUE_NAME = "workqueue_persistence";

	public static void main(String[] argv) throws java.io.IOException,
			java.lang.InterruptedException
	{
		// 区分不同工作进程的输出
		int hashCode = Work.class.hashCode();
		// 创建连接和频道
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		// 声明队列
		boolean durable = true;
		channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
		System.out.println(hashCode
				+ " [*] Waiting for messages. To exit press CTRL+C");
		//设置最大服务转发消息数量
		int prefetchCount = 1;
		channel.basicQos(prefetchCount);
		QueueingConsumer consumer = new QueueingConsumer(channel);
		// 指定消费队列
		boolean ack = false; // 打开应答机制
		channel.basicConsume(QUEUE_NAME, ack, consumer);
		while (true)
		{
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			String message = new String(delivery.getBody());

			System.out.println(hashCode + " [x] Received '" + message + "'");
			doWork(message);
			System.out.println(hashCode + " [x] Done");
			//channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

		}

	}

	/**
	 * 每一个点耗时1s
	 * 
	 * @param task
	 * @throws InterruptedException
	 */
	private static void doWork(String task) throws InterruptedException
	{
		for (char ch : task.toCharArray())
		{
			if (ch == '.')
				Thread.sleep(1000);
		}
	}
}

本文转自mfrbuaa博客园博客，原文链接：http://www.cnblogs.com/mfrbuaa/p/4628595.html，如需转载请自行联系原作者

RabbitMQ （两）工作队列

1、准备

2、消息应答（message acknowledgments）

3、消息持久化（Message durability）

4、公平转发（Fair dispatch）

5、完整的代码

热门文章

最新文章

相关课程

相关电子书

相关实验场景

RabbitMQ （两）工作队列

1、 准备

2、 消息应答（message acknowledgments）

3、 消息持久化（Message durability）

4、公平转发（Fair dispatch）

5、完整的代码

热门文章

最新文章

相关课程

相关电子书

相关实验场景

1、准备

2、消息应答（message acknowledgments）

3、消息持久化（Message durability）