RabbitMQ指南之二：工作队列（Work Queues）

  　　在上一章的指南中，我们写了一个命名队列：生产者往该命名队列发送消息、消费从从该命名队列中消费消息。在本章中，我们将创建一个工作队列，用于在多个工作者之间分配耗时的任务。工作队列(即任务队列)的主要思想是避免立即执行那些需要等他们执行完成的资源密集型任务。相反，我们将任务安排在稍后完成。我们将任务封装为消息并将其发送到队列，后台运行的工作进程将取出任务并执行完成。如果你启动了多个工作者，这些任务将在多个工作者之间分享。

　　这个概念也即我们说的异步，在项目中，有时候一个简单的web请求，后台要做一系统的操作，这时候，如果后台执行完成之后再给前台返回消息将会导致浏览器页面等待从而出现假死状态。因此，通常的做法是，在这个http请求到后台，后台获取到正确的参数等信息后立即给前台返回一个成功标志，然后后台异步地进行后续的操作。

 

1、准备

　　本章中，我们将发送字符串消息来模拟复杂的任务。这里因为没有一个真实的复杂任务，因此用thread.sleep()方法来模拟复杂耗时的任务。我们用字符串中的含点（“."）的数量来表示任务的复杂程度，一个点表示一秒钟的耗时，例如：一个发送”hello ...“字符串的任务将会耗时3秒钟。

　　我们可以直接将上一章中的send.java代码拿过来修改，允许从命令行发送消息。本程序将会把任务调试到工作队列，因此，我们将类名改为newtask.java：

string message = string.join(" ", argv);

channel.basicpublish("", "hello", null, message.getbytes());
system.out.println(" [x] sent '" + message + "'");

 　　此时完整的newtask.java代码为：

 1 public class newtask {
 2 
 3     private final static string queue_name = "hello";
 4 
 5     public static void main(string[] argv) throws ioexception, timeoutexception {
 6 
 7         connectionfactory connectionfactory = new connectionfactory();
 8         connectionfactory.sethost("host");
 9 
10         try(connection connection = connectionfactory.newconnection();
11             channel channel = connection.createchannel()) {
12 
13             channel.queuedeclare(queue_name,false,false,false,null);
14 
15             string message = string.join(" ", argv);
16             
17             channel.basicpublish("",queue_name,null,message.getbytes("utf-8"));
18             system.out.println(" [x] sent '" + message + "'");
19         }
20     }
21 }

　　之前的recv.java也要做一些修改：模拟字符串消息中的每个点耗时1秒钟，它将处理传送过来的消息并执行任务，因此，我们修改为work.java：

 1 delivercallback delivercallback = (consumertag, delivery) -> {
 2   string message = new string(delivery.getbody(), "utf-8");
 3 
 4   system.out.println(" [x] received '" + message + "'");
 5   try {
 6     dowork(message);
 7   } finally {
 8     system.out.println(" [x] done");
 9   }
10 };
11 boolean autoack = true; // acknowledgment is covered below
12 channel.basicconsume(task_queue_name, autoack, delivercallback, consumertag -> { });

　　我们模拟执行过程中耗时的伪任务：

1 private static void dowork(string task) throws interruptedexception {
2     for (char ch: task.tochararray()) {
3         if (ch == '.') thread.sleep(1000);
4     }
5 }

　　此时完整的work.java为：

 1 public class worker {
 2     private final static string task_queue_name = "hello";
 3 
 4     public static void main(string[] args) throws exception {
 5 
 6         connectionfactory connectionfactory = new connectionfactory();
 7         connectionfactory.sethost("host");
 8 
 9         connection connection = connectionfactory.newconnection();
10         channel channel = connection.createchannel();
11         channel.queuedeclare(task_queue_name,false,false,false,null);
12 
13         delivercallback delivercallback = (consumertag, delivery) -> {
14             string message = new string(delivery.getbody(), "utf-8");
15 
16             system.out.println(" [x] received '" + message + "'");
17             try {
18                 dowork(message);
19             } catch (interruptedexception e) {
20                 e.printstacktrace();
21             } finally {
22                 system.out.println(" [x] done");
23             }
24         };
25 
26         boolean autoack = true; // acknowledgment is covered below
27         channel.basicconsume(task_queue_name, autoack, delivercallback, consumertag -> { });
28     }
29 
30     private static void dowork(string task) throws interruptedexception {
31         for (char ch: task.tochararray()) {
32             if (ch == '.') thread.sleep(1000);
33         }
34     }
35 }

2、循环调度

　　使用工作队列的优点之一是能够轻松地进行并行化操作。假设我们在做一个后台日志收集系统，我们可以很容易地增加更多的worker从而提高系统性能。

　　首先，我们同时启动两个worker，同样地，我这里也放到idea中启动：

　　

　　

　　接下来，我们先后启动5个task，并分别通过main()参数传入五个字符串消息：

1 first message.
2 second message..
3 third message...
4 fourth message....
5 fifth message.....

　　

 

 　　执行五个发送任务之后，来看一下两个worker都接收到了什么样的消息：

　　        

　　默认情况下，rabbitmq将按顺序将每个消息发送给下一个使用者。平均每个消费者将得到相同数量的消息。这种消息的调度方式称之为循环调度，你可以开启更多的worker来进行测试。

3、消息回执

　　因为消费者执行一个任务会有时间耗时，假设一个消费者在执行一个任务执行一半的时候挂掉了将会怎样？消息会不会因此丢失？在我们目前的代码里，一旦rabbitmq将一条消息转发给了一个消费者后，将会立即将消息删除（注意worker.java里的autoack），因此，在我们上面例子里，如kill掉一个正在处理数据的worker，那么该数据将会丢失。不仅如此，所有那些指派给该worker的还未处理的消息也会丢失。

　　但在实际工作的，我们并不希望一个worker挂掉之后就会丢失数据，我们希望的是：如果该worker挂掉了，所有转发给该worker的消息将会重新转发给其他worker进行处理（包括处理了一半的消息）。为了确保一条消息永不丢失，rabbitmq支持消息回执。消费者在接收到一条消息，并且成功处理完成之后会给rabbitmq回发一条确认ack确认消息，rabbitmq此时才会删除该条消息。

　　如果一个worker正在处理一条消息时挂掉了（信道关闭、连接关闭、tcp连接丢失），它将没有机会发送ack回执，rabbitmq就认为该消息没有消费成功，于是便会将该消息重新放到队列中，如果此时有其他消费者还是在线状态，rabbitmq会立即将该条消息再转发给其他在线的消费者。这种机制可以保证任何消息都不会丢失。

　　默认情况下，需要手动进行消息确认，在前面的例子里，我们通过autoack=true显示地关闭了手动消息确认，因此，rabbitmq将采用自动消息确认的机制。现在，我们修改我们的程序，采用手动发送回执的方式，当我们完成对消息的处理后，再手动发送回执确认：

 1 channel.basicqos(1); // accept only one unack-ed message at a time (see below)
 2 
 3 delivercallback delivercallback = (consumertag, delivery) -> {
 4   string message = new string(delivery.getbody(), "utf-8");
 5 
 6   system.out.println(" [x] received '" + message + "'");
 7   try {
 8     dowork(message);
 9   } finally {
10     system.out.println(" [x] done");
11     channel.basicack(delivery.getenvelope().getdeliverytag(), false);
12   }
13 };
14 boolean autoack = false;
15 channel.basicconsume(task_queue_name, autoack, delivercallback, consumertag -> { });

　　ack发送信道必须和接收消息的信道(channel)是同一个，如果尝试通过一个不同的信道发送ack回执，将会抛出channel等级协议异常（官网说会抛出异常，但是我在实际测试中并没有抛异常，只是该条消息得不到回执，从而也无法删除）。

　　一个常见的错误是忘了手动回执，虽然只是一个简单的错误，但是带来的后果却是严重的，它将导致已经消费掉的消费不会被删除，并且当消费该消息的消费者在退出之后，rabbitmq会将该条消息重新进行转发，内存将被慢慢耗尽。我们可以通过正面的命令来检查这种错误：

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

    

　　该命令有三列内容，第一列是在监听的队列名称，第二列是ready状态的消息数量，第三列是unacked的消息数量。

4、消息的持久化

　　在3中我们讲解了如何保证当消费者挂掉之后消息不被丢失，但是，如果rabbitmq服务或者部署rabbitmq的服务器挂掉了之后，消息仍然会丢失。当rabbitmq崩溃之后，它将会忘记所有的队列和消息，除非，有什么机制让rabbitmq将队列信息和消息保存下来。

　　要确保消息和队列不会丢失，我们必须要确保两件事情。

　　首先，我们要确保rabbitmq永远不丢失队列，要做到这点，我们在定义的时候就需要告诉rabbitmq它是需要持久化的，通过指定durable参数实现：

boolean durable = true;
channel.queuedeclare("hello", durable, false, false, null);

　　虽然这个命令本身是正确的，但是在我们目前它不能工作。因为我们前面已经定义了一个非持久化的hello队列，rabbitmq不允许重新定义一个已经存在的队列（用不同的参数），否则会抛出异常：

exception in thread "main" java.io.ioexception
    at com.rabbitmq.client.impl.amqchannel.wrap(amqchannel.java:126)
    at com.rabbitmq.client.impl.amqchannel.wrap(amqchannel.java:122)
    at com.rabbitmq.client.impl.amqchannel.exnwrappingrpc(amqchannel.java:144)
    at com.rabbitmq.client.impl.channeln.queuedeclare(channeln.java:962)
    at com.rabbitmq.client.impl.recovery.autorecoveringchannel.queuedeclare(autorecoveringchannel.java:333)
    at myblog.myblog.java8.methodreference.rabbitmq.workqueue.newtask.main(newtask.java:23)
    suppressed: com.rabbitmq.client.alreadyclosedexception: channel is already closed due to channel error; protocol method: #method<channel.close>(reply-code=406, 
reply-text=precondition_failed - parameters for queue 'hello' in vhost '/' not equivalent, class-id=50, method-id=10)
        at com.rabbitmq.client.impl.amqchannel.processshutdownsignal(amqchannel.java:396)
        at com.rabbitmq.client.impl.channeln.startprocessshutdownsignal(channeln.java:292)
        at com.rabbitmq.client.impl.channeln.close(channeln.java:607)
        at com.rabbitmq.client.impl.channeln.close(channeln.java:541)
        at com.rabbitmq.client.impl.channeln.close(channeln.java:534)
        at com.rabbitmq.client.impl.recovery.autorecoveringchannel.close(autorecoveringchannel.java:68)
        at myblog.myblog.java8.methodreference.rabbitmq.workqueue.newtask.main(newtask.java:29)

　　要么重启rabbitmq让该临时队列消失，要么在控制台将该队列删除，或者重新创建一个新的队列：

1 boolean durable = true;
2 channel.queuedeclare("task_queue", durable, false, false, null);

　　生产者和消费者要做同步修改。

　　上面这一步，我们保证了队列（task_quee）的持久化，此时，即便rabbitmq崩溃了也不会丢失该队列，当rabbitmq重启后将自动重新加载该队列。

　　其次，我们需要确保我们的消息也被持久化，要做到这一点，在生产者发布消息的时候需要指定消息的属性为：persistent_text_plain。

1 import com.rabbitmq.client.messageproperties;
2 
3 channel.basicpublish("", "task_queue",
4             messageproperties.persistent_text_plain,
5             message.getbytes());

　　注意，即便设置了消息的持久化属性也不能保证消息会被100%地写入到磁盘中，因为rabbitmq在接收到消息和写入到磁盘不是同步的，有可能消息只是被写入到缓存中而还没来和及写入磁盘的时候，rabbitmq崩溃了，此时也会丢失消息。但无论如何，比前面简单的消息队列已经强大了很多。

5、公平调度

　　您可能已经注意到，任务调度仍然不能完全按照我们希望的方式工作。举个例子，在只有两个worker的环境中，奇数的消息比较重，偶数的消息比较轻时，一个worker将会一直处于忙碌状态，而另一个worker将会一直处于空闲状态，但rabbitmq并不知道这种情况，它会依然均衡地向两个worker传递消息。

　　发生这种情况是因为，当一个消息进入队列之后，rabbitmq只是盲目地将该第n个消息转发给第n个消费者，它并不关注每个消费者发了多少个回执。

　　为了解决这个问题，我们可以通过调用basicqos方法，给它传入1。这将告诉rabbitmq不要同时给一个队列转发多于1条的消息，换句话说，在一个消费者没有完成并回执前一条消息时，不要再给它转发其他消息。

1 int prefetchcount = 1;
2 channel.basicqos(prefetchcount);

6、完整的代码

　　一、newtask.java

 1 import com.rabbitmq.client.channel;
 2 import com.rabbitmq.client.connection;
 3 import com.rabbitmq.client.connectionfactory;
 4 import com.rabbitmq.client.messageproperties;
 5 
 6 public class newtask {
 7 
 8   private static final string task_queue_name = "task_queue";
 9 
10   public static void main(string[] argv) throws exception {
11     connectionfactory factory = new connectionfactory();
12     factory.sethost("localhost");
13     try (connection connection = factory.newconnection();
14          channel channel = connection.createchannel()) {
15         channel.queuedeclare(task_queue_name, true, false, false, null);
16 
17         string message = string.join(" ", argv);
18 
19         channel.basicpublish("", task_queue_name,
20                 messageproperties.persistent_text_plain,
21                 message.getbytes("utf-8"));
22         system.out.println(" [x] sent '" + message + "'");
23     }
24   }
25 
26 }

　　二、worker.java

　　

 1 import com.rabbitmq.client.channel;
 2 import com.rabbitmq.client.connection;
 3 import com.rabbitmq.client.connectionfactory;
 4 import com.rabbitmq.client.delivercallback;
 5 
 6 public class worker {
 7 
 8   private static final string task_queue_name = "task_queue";
 9 
10   public static void main(string[] argv) throws exception {
11     connectionfactory factory = new connectionfactory();
12     factory.sethost("localhost");
13     final connection connection = factory.newconnection();
14     final channel channel = connection.createchannel();
15 
16     channel.queuedeclare(task_queue_name, true, false, false, null);
17     system.out.println(" [*] waiting for messages. to exit press ctrl+c");
18 
19     channel.basicqos(1);
20 
21     delivercallback delivercallback = (consumertag, delivery) -> {
22         string message = new string(delivery.getbody(), "utf-8");
23 
24         system.out.println(" [x] received '" + message + "'");
25         try {
26             dowork(message);
27         } finally {
28             system.out.println(" [x] done");
29             channel.basicack(delivery.getenvelope().getdeliverytag(), false);
30         }
31     };
32     channel.basicconsume(task_queue_name, false, delivercallback, consumertag -> { });
33   }
34 
35   private static void dowork(string task) {
36     for (char ch : task.tochararray()) {
37         if (ch == '.') {
38             try {
39                 thread.sleep(1000);
40             } catch (interruptedexception _ignored) {
41                 thread.currentthread().interrupt();
42             }
43         }
44     }
45   }
46 }

　　至此，工作队列模式讲解完了，下一章，将讲解发布-订阅模式。