rabbitMQ学习笔记(3):Work Queues

在上一篇文章中我们解决了最简单的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();
                    }
            }
        }
    }
}
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,456评论 5 477
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,370评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,337评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,583评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,596评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,572评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,936评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,595评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,850评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,601评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,685评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,371评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,951评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,934评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,167评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 43,636评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,411评论 2 342

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 134,585评论 18 139
  • 什么叫消息队列 消息(Message)是指在应用间传送的数据。消息可以非常简单,比如只包含文本字符串,也可以更复杂...
    lijun_m阅读 1,335评论 0 1
  • 来源 RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控...
    jiangmo阅读 10,343评论 2 34
  • 上篇讲过简单的hello消息,这篇我们将实现一个可以在多个Consumer上发送持久化消息的work queue。...
    初级赛亚人阅读 1,842评论 0 3
  • 1. 历史 RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue )的...
    高广超阅读 6,092评论 3 51