工作队列的主要任务:避免立刻执行资源密集型任务,然后必须等待其完成。相反地,我们进行任务调度,我们把任务封装为消息发送给队列。工作进行在后台运行并不断的从队列中取出任务然后执行。当你运行了多个工作进程时,任务队列中的任务将会被工作进程共享执行。
这样的概念在web应用中极其有用,当在很短的HTTP请求间需要执行复杂的任务。
准备工作
我们使用Thread.sleep来模拟耗时的任务。现在,我们对上面“Hello World”的例子进行改造。
发送端
public class NewTask { private final static String QUEUE_NAME = "Somnus"; public static void main(String[] args) throws IOException, TimeoutException { // 创建连接 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); // 发送消息 for (int i = 0; i < 10; i++) { String message = "Somnus-" + i; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] 发送消息是:'" + message + "'"); } // 关闭连接 channel.close(); connection.close(); } }
接收端
import java.io.IOException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; public class Worker { private final static String QUEUE_NAME = "Somnus"; public static void main(String[] args) throws IOException, TimeoutException { // 创建连接 ConnectionFactory factory = new ConnectionFactory(); // 设置MabbitMQ, 主机ip或者主机名 factory.setHost("localhost"); // 创建一个连接 Connection connection = factory.newConnection(); // 创建一个通道 Channel channel = connection.createChannel(); // 指定一个队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(" [*] 等待消息进入.请按 CTRL+C 结束"); // 创建队列消费者 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] 接收消息是:'" + message + "'"); try { doWork(message); } catch (InterruptedException e) { e.printStackTrace(); } } }; // acknowledgment is covered below boolean autoAck = true; channel.basicConsume(QUEUE_NAME, autoAck, consumer); } private static void doWork(String task) throws InterruptedException { String[] taskArr = task.split("-"); TimeUnit.SECONDS.sleep(Long.valueOf(taskArr[1])); } }
轮询调度(Round-robin dispatching)
任务队列的优点是能够很容易的进行并行工作。如果我们积压了很多工作,我们仅仅通过增加更多的workers就可以解决问题,使系统的伸缩性更加容易。
默认情况下,RabbitMQ将会发送的每一条消息按照顺序给下一个消费者。平均每一个消费者将获得相同数量的消息。这种分配方式称为轮询调度。值得注意的是,这种分配过程是一次性分配,并非一个一个分配。
首先,我们开启两个Worker。然后,启动NewTask进行消息发送。打印结果如下
Worker1
[*] 等待消息进入.请按 CTRL+C 结束 [x] 接收消息是:'Somnus-0' [x] 接收消息是:'Somnus-2' [x] 接收消息是:'Somnus-4' [x] 接收消息是:'Somnus-6' [x] 接收消息是:'Somnus-8'
Worker2
[*] 等待消息进入.请按 CTRL+C 结束 [x] 接收消息是:'Somnus-1' [x] 接收消息是:'Somnus-3' [x] 接收消息是:'Somnus-5' [x] 接收消息是:'Somnus-7' [x] 接收消息是:'Somnus-9'
再做一个实验,我们开启3个worker,然后,启动NewTask进行消息发送。打印结果如下
Worker1
[*] 等待消息进入.请按 CTRL+C 结束 [x] 接收消息是:'Somnus-0' [x] 接收消息是:'Somnus-3' [x] 接收消息是:'Somnus-6' [x] 接收消息是:'Somnus-9'
Worker2
[*] 等待消息进入.请按 CTRL+C 结束 [x] 接收消息是:'Somnus-1' [x] 接收消息是:'Somnus-4' [x] 接收消息是:'Somnus-7'
Worker3
[*] 等待消息进入.请按 CTRL+C 结束 [x] 接收消息是:'Somnus-2' [x] 接收消息是:'Somnus-5' [x] 接收消息是:'Somnus-8'
我们发现,通过增加更多的workers就可以进行并行工作,并且接受的消息平均分配。
消息应答(Message acknowledgment)
执行一个任务需要花费几秒钟。在执行任务时发生中断。上面例子,RabbItMQ一旦给消费者发送消息,会马上从内存中移除这个信息。在这种情况下,如果杀死正在执行任务的某个Worker,我们会丢失它正在处理的信息。我们也会丢失已经转发给这个工作者且它还未执行的消息。
我们不希望丢失任何消息。当某个Worker被杀死时,我们希望将任务传递给另一个Worker。
为了保证消息永远不会丢失,RabbitMQ支持消息应答。消费者发送应答给RabbitMQ,告诉它信息已经被接收和处理,然后RabbitMQ可以自由的进行信息删除。如果消费者被杀死而没有发送应答,RabbitMQ会认为该信息没有被完全的处理,然后将会重新转发给别的消费者。通过这种方式,你可以确认信息不会被丢失,即使消者被杀死。
这种机制并没有超时时间这么一说,RabbitMQ只有在消费者连接断开是重新转发此信息。如果消费者处理一个信息需要耗费特别特别长的时间是允许的。
消息应答默认是打开的。上面的代码中我们通过显示的设置autoAsk=true关闭了这种机制。下面我们修改代码Worker.java。主要修改两个地方。
// 打开消息应答 boolean ack = false ; channel.basicConsume(QUEUE_NAME, ack, consumer);
此时,可以做一个实验,我们开启3个worker,然后,启动NewTask进行消息发送。立即关闭一个worker,观察输出结果,是否重新转发关闭的worker的消息。
消息持久化(Message durability)
我们已经学会了如何确保即使消费者死了,消息也不会丢失。但是如果此时RabbitMQ服务被停止,我们的消息仍然会丢失。当RabbitMQ退出或者异常退出,将会丢失所有的队列和信息,除非你告诉它不要丢失。为了保证消息在服务停止的时候,不丢失,我们要考虑保证消息的持久化。
首先,我们需要确认RabbitMQ永远不会丢失我们的队列。为此,我们需要声明它为持久化的。
boolean durable = true; channel.queueDeclare("task_queue", durable, false, false, null);
其次,我们需要标识我们的信息为持久化的。通过设置MessageProperties值为PERSISTENT_TEXT_PLAIN。
channel.basicPublish("", "task_queue",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
RabbitMQ不允许使用不同的参数重新定义一个队列,所以已经存在的队列,我们无法修改其属性。
此时,读者,可以做一个实验,我们启动NewTask进行消息发送,然后RabbitMQ关闭服务。再重新启动服务abbitMQ,开启1个worker,观察输出结果,是否worker的接受到之前NewTask发送消息。
公平转发(Fair dispatch)
你可能已经注意到,目前的消息调度并非是我们想要的。例如,这样一种情况下,一个消费者将一直繁忙,另一个消费者却很快执行完任务后等待,没有任何工作。造成这样的原因,是因为RabbitMQ仅仅是当消息到达队列后进行转发消息,并不在乎有多少任务。
为了解决这样的问题,我们可以使用basicQos方法,传递参数为prefetchCount = 1。这样告诉RabbitMQ不要在同一时间给一个消费者超过一条消息。换句话说,只有在消费者空闲的时候会发送下一条信息。
int prefetchCount = 1; channel.basicQos(prefetchCount) ;
相关连接:
RabbitMq3.6官方文档