rabbitMQ学习笔记(3):Work Queues

作者: lsfire | 来源:发表于2016-10-14 19:39 被阅读0次

在上一篇文章中我们解决了最简单的helloworld 消息传递，这一篇中我们来探讨rabbitMQ中的任务分发

rabbitMQ任务分发机制的核心出发点就是避免立刻进行“资源密集”或者说time-consuming的任务，因为这样就必须同步等待耗时任务的完成。取而代之的是schedule这些任务再稍后完成，在本篇的demo中我们将task封装成一条message将其发送到队列中。一个后台运行的worker进程会从队列中获取message并执行任务。
任务分发机制在web应用中非常有用，因为通常我们不会在一次http请求响应过程中处理复杂的耗时任务。当有Consumer需要大量的运算时，RabbitMQ Server需要一定的分发机制来balance每个Consumer的load。
rabbitMQ的任务分发机制模型如下图所示：

准备

在上一篇文章中的实例中，我们发送一个“hello world”的消息，在这篇文章中，我们发送一个字符串代表复杂的任务，用thread.sleep()函数模拟可能的操作，比如图片的resize，pdf的内容渲染或者提取。
复用上文中的code，为了便于区别，我们还是命名为new_task.java

String[] messages = {"a","b","c","d"};
        String message = getMessage(messages);
        //the concept of channel in rabbitMQ,the first parameter defines the name of exchange,
        //"" means the default exchange
        channel.basicPublish("", TASK_QUEUE_NAME,
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");

getMessage方法，非常简单：

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();
      }

原来的receiver代码也需要略作改动，同样为了便于理解，我们将其重新命名为worker.java,并且根据message中的“.”进行任务处理的模拟。

final Consumer consumer  = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                    throws IOException {
                // TODO Auto-generated method stub
                //super.handleDelivery(consumerTag, envelope, properties, body);
                String message = new String(body,"UTF-8");
                
                try {
                    doWork(message);

doWork方法：

private static void doWork(String task){
        for(char c : task.toCharArray()){
            System.out.print(c + "\t");
            if(c == '.'){
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        Thread.currentThread().interrupt();
                    }
                
            }
        }
    }

至此我们已经完成了工作的大半。

round-robin dispatching 循环分发

RabbitMQ的分发机制非常适合扩展，而且它是专门为并发程序设计的。如果现在load加重，那么只需要创建更多的Consumer来进行任务处理即可。首先我们来运行两个worker实例，这里通过命令行的方式完成：

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

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

然后producer将要发布新任务：

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

我们来观察一下 worker收到的消息

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
 [x] Received 'Second message..'
 [x] Received 'Fourth message....'

默认情况下，rabbitMQ会按顺序的将message依次分发给下一个consumer，这种分发方式就叫做round-robin。

Message Acknowledgement 消息确认

运行一个任务可能需要好几秒甚至更久，那么有个问题值得探究，如果一个consumer开始了一段长任务，但是在任务处理到一半时consumer进程异常退出会发生什么。不幸的是，如果我们采用no-ack的方式，这个消息就消失了。也就是说，也就是说，每次Consumer接到数据后，而不管是否处理完成，RabbitMQ Server会立即把这个Message标记为完成，然后从queue中删除了。
如果一个Consumer异常退出了，它处理的数据能够被另外的Consumer处理，这样数据在这种情况下就不会丢失了（注意是这种情况下）。为了保证数据不被丢失，RabbitMQ支持消息确认机制，即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到，那么我们不能采用no-ack。而应该是在处理完数据后发送ack。
在处理数据后发送的ack，就是告诉RabbitMQ数据已经被接收，处理完成，RabbitMQ可以去安全的删除它了。如果Consumer退出了但是没有发送ack，那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说，RabbitMQ给了Consumer足够长的时间来做数据处理。
message ack 默认情况下是开启的，在上一节中我们通过autoAck=true来显式的关闭了acknowledgement，现在我们修改handleDelivery回调函数，来发送确认信息。

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);
    }
  }
};

Message Durability 消息持久化

上文中我们学习了在consumer异常退出或者中断的情况下如何通过消息确认来保证消息的不丢失，但是在rabbitMQ server异常退出或者中断情况下就无能为力了，这种情况持久化消息可以帮忙。消息持久化需要做两件事情就是声明queue和message都是durable的：

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

上述语句执行不会有什么错误，但是确得不到我们想要的结果，原因就是RabbitMQ Server已经维护了一个叫hello的queue，那么上述执行不会有任何的作用，也就是hello的任何属性都不会被影响。这一点在上篇文章也讨论过。那么workaround也很简单，声明一个另外的名字的queue，比如名字定位task_queue：

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

再次强调，Producer和Consumer都应该去创建这个queue，尽管只有一个地方的创建是真正起作用的。接下来，需要持久化Message，即在Publish的时候指定一个properties，方式如下：

import com.rabbitmq.client.MessageProperties;

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

Fair Dispatch 公平分发

你可能也注意到了，分发机制不是那么优雅。默认状态下，RabbitMQ将第n个Message分发给第n个Consumer。当然n是取余后的。它不管Consumer是否还有unacked Message，只是按照这个默认机制进行分发。那么如果有个Consumer工作比较重，那么就会导致有的Consumer基本没事可做，有的Consumer却是毫无休息的机会。那么，RabbitMQ是如何处理这种问题呢？

过 basic.qos 方法设置prefetch_count=1 。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。设置方法如下：

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

整合后的整个代码如下：

new task.java

package cn.edu.nju.liushao.worker;

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

public class NewTask {

      private static final String TASK_QUEUE_NAME = "task_queue";
      private static final String MQ_ADDRESS = "localhost";
      public static void main(String[] argv) throws Exception {
          /*
           * init factory,connection and channel 
           */
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(MQ_ADDRESS);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //declare a queue
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

        String[] messages = {"a","b","c","d"};
        String message = getMessage(messages);
        //the concept of channel in rabbitMQ,the first parameter defines the name of exchange,
        //"" means the default exchange
        channel.basicPublish("", TASK_QUEUE_NAME,
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes("UTF-8"));
        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

package cn.edu.nju.liushao.worker;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

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;
import com.rabbitmq.client.AMQP.BasicProperties;

public class Worker {
    private static final String TASK_QUEUE_NAME = "task_queue";
    private static final String MQ_ADDRESS = "localhost";
    
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(MQ_ADDRESS);
        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, BasicProperties properties, byte[] body)
                    throws IOException {
                // TODO Auto-generated method stub
                //super.handleDelivery(consumerTag, envelope, properties, body);
                String message = new String(body,"UTF-8");
                
                try {
                    doWork(message);
                } finally {
                    System.out.println("[x] done");
                    // send back acknowledgement
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
                
                
            }
        };
        
        channel.basicConsume(TASK_QUEUE_NAME, false,consumer);
    }
    
    private static void doWork(String task){
        for(char c : task.toCharArray()){
            System.out.print(c + "\t");
            if(c == '.'){
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        Thread.currentThread().interrupt();
                    }
            }
        }
    }
}

rabbitMQ学习笔记(3):Work Queues

准备

round-robin dispatching 循环分发

Message Acknowledgement 消息确认

Message Durability 消息持久化

Fair Dispatch 公平分发

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读