Java与RabbitMQ（四）Rabbirmq JAVA编程之工作队列 Work Queues

青山师

发布于 2023-05-04 20:09:43

1370

发布于 2023-05-04 20:09:43

文章被收录于专栏：IT当时语_青山师_JAVA技术栈IT当时语_青山师_JAVA技术栈

本系列教程主要针对使用Java语言进行Rabbitmq的相关编程。阅读前请确认已经安装过rabbit服务。关于如何安装rabbitmq，请参考如何使用rabbitmq.

工作队列（Work Queues）

(using the Java Client)

在 本系列教程的第一节中我们编写了两个程序（生产者Send、消费者Recv）分别从命名的队列中发送、获取消息。在本节中我们将会创建一个 工作队列(Work Queue)用来分发多个消费者处理消息的耗时任务。

工作队列(Work Queue)也称任务队列的主要思想是：尽量避免去做一个资源密集型任务（耗时的任务）并且等待它完成才能做其他的事情，而是分发之后再安排完成任务。我们将一个任务（task ）封装成一个消息，并将其发送到队列里。一个worker的后台进程将pop这些task并且最终执行完它们所表示的作业。在该程序中，当你运行许多工作时任务task是共享的。

这一概念在Web应用程序是很有用处的，因为不可能在一个很短的HTTP请求里处理一个非常复杂且耗时的任务。

准备工作

在本系列教程的第一节中我们发送了一个消息”Hello World!”。现在，我们将发送支持复杂任务的字符串作为消息内容。在本节中，我们没有一个真实世界的任务，如图像进行调整或PDF文件的渲染，假装我们很忙-采用Thread.sleep()。在这里我们将以字符串中的点符(.)的数量作为作业的复杂性。如发现一个点符(.)则将占用1秒钟的工作时间，线程将等待1秒钟。举个例子，如果作业描述是“Hello…”的形式将会占用3秒时间。

我们稍微修改一下第一节中的Send.java类，允许发送任意消息（官方教程是从命令行设置main方法参数，这里我们使用eclipse/idea设置mian方法的参数即可）。这个程序是安排任务到工作队列中，所以我们姑且叫它NewTask.java:

String message = getMessage(argv);

channel.basicPublish("", "hello", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");

对输入参数处理的两个方法：

private static String getMessage(String[] strings){
    if (strings.length < 1)
        return "Hello World!";
    return joinStrings(strings, " ");
}

private static String joinStrings(String[] strings, String delimiter) {
    int length = strings.length;
    if (length == 0) return "";
    StringBuilder words = new StringBuilder(strings[0]);
    for (int i = 1; i < length; i++) {
        words.append(delimiter).append(strings[i]);
    }
    return words.toString();
}

第一节中的Recv.java也需要做一些改变了：它在接收到每一个.符时需要休眠1秒钟。它要做的是传递消息和处理任务，所以我们可以叫它Worker.java:

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 {
      doWork(message);
    } finally {
      System.out.println(" [x] Done");
    }
  }
};
boolean autoAck = true; // acknowledgment is covered below
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);

我们的工作类通过以下方式模拟耗时处理:

private static void doWork(String task) throws InterruptedException {
    for (char ch: task.toCharArray()) {
        if (ch == '.') Thread.sleep(1000);
    }
}

你也可以在命令行中编译这两个文件(记得依赖的*.jar文件需要和将要编译的*.java文件放在同一个目录中):

$ javac -cp rabbitmq-core.jar NewTask.java Worker.java

轮询分发(Round-robin dispatching)

一个使用任务队列的优点就是容易并行化处理工作。如果我们积累了大量的工作，我们可以增加更多的工作者，通过这样的方式，很容易规模化。

首先，我们尝试同一时间让两个Worker实例运行。他们都会从队列中获得消息，但是怎样去实现呢，先让我想想。

你需要打开3个控制台，其中两个运行Worker。

shell1$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
Worker
 [*] Waiting for messages. To exit press CTRL+C

shell2$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
Worker
 [*] Waiting for messages. To exit press CTRL+C

但是在eclipse/idea中我们只需要两次执行Worker的main方法即可。

一旦你已经运行了消费者(Worker)之后就可以运行NewTask的main方法去发布消息了：

shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask First message.
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask Second message..
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask Third message...
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask Fourth message....
shell3$ java -cp .:commons-io-1.2.jar:commons-cli-1.1.jar:rabbitmq-client.jar
NewTask Fifth message.....

同样，在eclipse/idea中我们只需要多次执行NewTask的main方法即可,在执行时main方法时你可以传入不同的参数以示区别

现在我们看看第一个work接收到的输出信息：

 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'First message.'
 [x] Done
 [x] Received 'Third message...'
 [x] Done
 [x] Received 'Fifth message.....'
 [x] Done

第二个work接收到的输出信息：

 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'Second message..'
 [x] Done
 [x] Received 'Fourth message....'
 [x] Done

注意: 默认情况下，RabbitMQ会把在序列中的每个消息传递个下一个消费者，在本例中我们开启了两个消费者，则第一个消息会发送给第一个消费者，第二个消息发送给第二个消费者，第三个消息会发送给第一个消费者，第四个消息发送给第二个消费者…。平均每一个消费者将获得相同数量的消息。这种分配方式称为轮询消息。你尝试开启更多消费者验证这个机制。

消息确认答复（Message acknowledgment）

完成一个任务可能会花几秒。你可能会想知道，如果一个消费者开始处理一个会耗费大量时间的任务导致线程死去，结果只是部分完成，又该怎么处理呢？在我们现在的程序中，一旦RabbitMQ传送一个消息到消费者之后，RabbitMQ即刻会在内存中移除这个消息。在这个案例中，如果一个worker正在运行时被杀死了进程，我们将会失去这个消息。我们也会失去其他已经分发到这个消费者但是还未来得及处理的消息。

但是我们不想遗漏任何任务。如果一个worker死去，我们希望这个任务会再次发送到另一个worker去处理。

为了确保每一个消息都不会丢失，RabbitMQ支持消息确认答复(acknowledgments)机制。消费者在接收到消息、处理完毕后可以发送一个 ack(nowledgement)告知RabbitMQ，这个消息可以从队列中删除了。

如果一个消费者灭有发送ack就被销毁了（channel关闭 or connection关闭 or TCP连接丢失），RabbitMQ会知道消息没有完全处理完毕并且会重新分发。如果此时其他消费者是可用的，RabbitMQ会重新分发给可用的消费者。如此这般，就算消费者间或消亡不可用了，我们也可以保证不会有消息丢失。

这里没有任何消息超时。当消费者消亡时，RabbitMQ会重新分发该消息。这是OK的，即使某个消息处理需要话费很长、很长的时间。

消息确认答复默认是开启的。在前面的Worker例子中，通过 autoAck=true显示关闭了消息确认。现在在Worker中我们通过设置它为false，一旦消费者处理完任务，就会发送一个消息确认答复到RabbitMQ。

channel.basicQos(1); // accept only one unack-ed message at a time (see below)

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 {
      doWork(message);
    } finally {
      System.out.println(" [x] Done");
      channel.basicAck(envelope.getDeliveryTag(), false);
    }
  }
};
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);

使用上面这段代码，即使在开启消费者后杀死进程(运行Worker的main方法，还未处理完辄关闭控制台)，也能保证消息不会丢失。消费者如果消亡，还未收到确认答复的消息，RabbitMQ会再次分发出去。

遗漏消息确认答复 忘记使用basicAck是很常见的错误。尽管这是一个简单的错误，但后果是严重的。消息虽然将会被重新分发，但是RabbitMQ将会占用越来越多的内存如果那些消息一直得不到确认答复的话。为了调试监控这种错误，你可以使用rabbitmqctl 打印messages_unacknowledged 的状态(rabbitmqctl list_queues name messages_ready messages_unacknowledged)： C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.5\sbin>rabbitmqctl list_queues name messages_ready messages_unacknowledged Listing queues ... hello 0 0

消息持久化

前面我们学会了即使消费者消亡消息也不会丢失的处理方法。但是RabbitMQ服务关闭的话，我们的任务还是会丢失。

当RabbitMQ关闭服务或者崩溃，服务器将不会保存队列和消息，除非你告诉他让他记住这些队列和消息。要保证消息不会丢失需要做的两件事：保证队列、消息都是持久化(durable)的。

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

尽管上述代码是OK的，但是它不会起任何效用。因为我们已经定义了一个名为”hello”的未持久化的队列。RabbitMQ不允许通过不同的参数重新定义一个已经存在的队列，你坚持这样做的话会返回一个错误。我们可以采用一个变通方案 —— 我们可以声明一个另外名称的队列，将其设置为持久化(durable),例如 task_queue

boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);

注意:这段代码必须在生产者、消费者两端都使用。

此时此刻，我们已然可以保证 task_queue 队列不会丢失了，纵使重启RabbitMQ 。第二点：我们需要使我们的消息持久化 —— 通过设置MessageProperties (实现了BasicProperties )的一个属性值PERSISTENT_TEXT_PLAIN :

import com.rabbitmq.client.MessageProperties;

channel.basicPublish("", "task_queue",
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());

消息持久化的注意事项 标记消息持久化并不能完全保证消息就不会丢失了。虽然它告诉RabbitMQ将消息保存到磁盘中，但是在这区间仍然有一个非常短小的时间间隔。而且，RabbitMQ不会对每一个消息都做fsync(2)处理的 —— 可能仅仅只是缓存并不会写入到磁盘中。PERSISTENT_TEXT_PLAIN 的持久性不会很强，但是对于我们的简单的工作队列已经足够了。如果你需要一个强持久性的机制，可以访问我们的发布确认机制。

公平分发消息

你可能已经注意到，消息分发机制仍然有点不符合我们所想的。例如，一个工作场景中有两个消费者，所有的奇数消息是重量级的需要处理很长时间而所有的偶数消息是轻量级的可能瞬间即可处理完毕。但是RabbitMQ不知道这些，它还是会均匀的分发消息，这样第一个消费者一直处理着重量级奇数消息。

这是因为RabbitMQ只是在消息进入队列时做了分发处理。它不会去帮你看一个消费者返回的关于消息处理的ack(消息确认答复)数量。它仅仅是在盲目的分发 —— 就像第n个消息对应第n个消费者这样简单映射的分发处理。到目前为止，我们的RabbitMQ的头脑还真是简单啊。

为了对抗这种盲目分发机制，我们可以通过 basicQos 方法设置prefetchCount = 1来保证”公平”分发消息。 basicQos(1)旨在告诉RabbitMQ不要分发超过一个消息给消费者。换句话说，不要分发过多的消息给消费者，除非消费者已经处理完或者对前一个处理的消息做了确认应答了。而是，将消息分发给另外一个不太忙的消费者。

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

队列大小的一些注意事项 如果所有的消费者都很忙，队列将会填满。所以你需要随时关注队列的空间大小，有可能的话增加更多的消费者，或者采取其他行而有效的策略。

最终的NewTask.java代码我们使用NewTask2.java:

/**
 * 
 */
package org.byron4j.rabbitmq_core.workqueue;

import java.util.concurrent.TimeoutException;

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

/**
 * @author Byron.Y.Y
 * @date 2016年10月2日
 */
public class NewTask2 {
    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] argv) throws java.io.IOException, Exception {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

        String message = getMessage(argv);

        channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

        channel.close();
        connection.close();
    }

    private static String getMessage(String[] strings) {
        if (strings.length < 1)
            return "Hello World!";
        return joinStrings(strings, " ");
    }

    private static String joinStrings(String[] strings, String delimiter) {
        int length = strings.length;
        if (length == 0)
            return "";
        StringBuilder words = new StringBuilder(strings[0]);
        for (int i = 1; i < length; i++) {
            words.append(delimiter).append(strings[i]);
        }
        return words.toString();
    }
}

最终的Worker.java代码我们使用Worker2.java:

package org.byron4j.rabbitmq_core.workqueue;

import java.io.IOException;

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;

/**
 * 
 * @author Byron.Y.Y
 * @date 2016年10月2日
 */
public class Worker2 {
    private static final String TASK_QUEUE_NAME = "task_queue";

      public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        final Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME, true, 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 {
              doWork(message);
            } finally {
              System.out.println(" [x] Done");
              channel.basicAck(envelope.getDeliveryTag(), false);
            }
          }
        };
        boolean autoAck = false;
        channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);
      }

      private static void doWork(String task) {
        for (char ch : task.toCharArray()) {
          if (ch == '.') {
            try {
              Thread.sleep(1000);
            } catch (InterruptedException _ignored) {
              Thread.currentThread().interrupt();
            }
          }
        }
      }
}

通过使用消息确认答复机制、公平分发策略，你可以创建一个工作队列。持久化操作可以让任务在RabbitMQ重启后仍然可以工作。

更多关于Channel的方法和MessageProperties，你可以查看在线帮助文档。

现在我们可以开始第三节发布/订阅的学习了，将会学习怎样将相同的消息传递给很多消费者。

原文链接:http://www.rabbitmq.com/tutorials/tutorial-two-java.html

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2016-10-02，如有侵权请联系 cloudcommunity@tencent.com 删除

java