RabbitMQ之Work Queues

本文翻译自RabbitMQ官网。

一、工作队列介绍

      上一部分通过简单的代码介绍了通过队列发送和接收消息，这一部分创建一个工作队列来发送和接收耗时的消息。工作队列的主要设计思想是将资源密集型的任务先放进队列中，避免立刻做资源密集型的任务，也就避免了一直等待该任务处理完成造成的堵塞。在短暂的HTTP请求中不可能一直等待复杂的任务处理完成后返回，而是先将消息放入消息队列就可以返回了，消息的处理待消费者从消息队列中取出后再处理，达到了异步的效果。

      在本节中模拟一个复杂的任务，如下所示。

 

 private static void doWork(String msg){
        char[] array = msg.toCharArray();
        for (char ch : array){
            if (ch == '.') {
                try {
                    Thread.sleep(1000);  
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
        }
    }

 当消息中有字符‘.’时，sleep 1ms。

将消费队列改为：task_queue3。

结果如下：

 

producer:jing... quan
before send message:jing... quan
after send message:jing... quan
******time is :******2ms
start consume....
consumer:jing... quan

 二、相关属性设置

       1、消息确认（message acknowledgment）：为了防止在消费端挂掉之后消息丢失

       在消费端接收到消息之后，开始处理复杂的任务，如果当某一个消费者在执行任务的过程中突然挂掉了（原因可能是channel被关闭了、连接被关闭了或者是TCP的连接被断掉了）怎么办？消息会不会丢失呢？一旦RabbitMQ发送消息给消费者后，该消息就立刻被从内存中删除掉了，这种情况下，如果某一个消费者挂掉了，那么它正在处理的所有消息就会丢失掉。但是在实际情况中，我们是不希望丢失任何消息的，所以需要使用到RabbitMQ的消息确认属性。消息确认即当消费者已经对某一个特定的消息处理完毕之后它会发给RabbitMQ一个ACK（nowledgment），告知RabbitMQ可以将这段消息删除掉了。如果消费端是在处理消息的过程中挂掉的，那么肯定不会发送给RabbitMQ一个ACK，这个时候RabbitMQ会将消息重新发送给其他未挂掉的消费者。

       在默认情况下，ACK是打开的，可以通过basicConsume来进行设置：

channel.basicConsume(queue, false, this);

 以上情况会将ACK关掉，但是为了防止丢失，还是将其打开：

channel.basicConsume(queue, true, this);

 

      2、 消息的持久性：为了防止在服务器挂掉之后消息丢失

      当RabbitMQ挂掉需要重启时，在默认情况下，它会将其之前创建的channel和message全部删除，这个时候就会造成数据丢失。为了防止这种情况发生，需要做两个方面的设置：a,消息队列的持久性设置；b,消息的持久性设置。

      首先是消息队列的持久性设置：

durable = true;
channel.queueDeclare(queue, durable, exclusive, autoDelete, arguments);

      以上代码在生产者端和消费者端需同时设置。

     然后是消息的持久性设置：

channel.basicPublish("", queue, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());

    在发布消息的时候，将消息的属性设置为MessageProperties.PERSISTENT_TEXT_PLAIN

    以上设置就保证了消息的持久性，当Rabbit服务器因为某些原因重启后，还是可以消费到尚未被消费的消息。

 

       3、合理分发

      有这样的情景：两个消费者，一个耗费资源比较小的消息和一个耗费资源比较大的消息，在默认情况下，RabbitMQ会平均将消息分发给两个消费者，这样就容易造成一个消费者一直很忙而另一个消费者比较空闲的情况，这是因为当消息被放入消息队列后RabbitMQ只负责分发，而不关注消费的处理情况，如果进行以下设置：

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

 RabbitMQ就会在某一个消费者处理完成前一个消息之后才会给其分发下一个消息，如果该消费者还没有处理完毕，它就会将待处理消息分发给其他消费者。