RabbitMQ入門-高效的Work模式 - bigdataZJ－IT工程師數位筆記本

文章出處

扛不住的Hello World模式

上篇《RabbitMQ入門-從HelloWorld開始》介紹了RabbitMQ中最基本的Hello World模型。正如其名，Hello World模型組成簡單，也很好理解，我們也看到了一條消息時如何從一個生產者最終流向隊列并最終被消費者消費的過程。

但是，過于簡單、單調的模型設計也存在一些缺陷。假使現在隊列Queue中擠壓了很多的消息沒有被消費，Hello World模型中只有一個消費者，在消費消息時會顯得力不從心。如果遇上網絡狀況異常等情況，則消費速率就更加不同樂觀，從而影響了消息的處理效率，影響網站應用的性能。
很直觀的思路，我們能想到的是，一個人不行，那就多來幾個人，這時候就有了我們的Work模型。

多管齊下的Work模式

該模型具有以下特征

一個消息生產者P，一個消息存儲隊列Q，多個消息消費者C
Work模型能夠較好的解決資源密集型場景的問題，不需要像Hello World那樣孤注一擲的等唯一的消費者消費完
多個消費者，多管齊下，更加高效的并行處理消息

實例

如何構造一個資源密集型的場景
相較于Hello World，Work模式主要是在資源密集型的場景更能發揮威力，那么沒有工作環境或者很難遇到這樣的情況，我們怎么辦？
其實，這個場景的本質是為了體現一個消費者處理要很長時間的時候，這個模式是如何發揮作用的。那么，我們可以讓每個消費者處理的時間長點不就行了，要讓Consumer處理的時間長很簡單，只要調用Thread.sleep()即可。
發送端

對于發送端相對Hello World類型來說，沒有什么不同。這是我們隊這里發送的消息采用指定的格式比如“hello......”，在后面的發送端接收消息后，當遇到"."則停頓1秒或者2秒，所以程序如下

package com.ximalaya.openapi.rabbitmq.work;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
/**
 * Created by jackie on 17/8/4.
 */
public class NewTask {

    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.3.161");
        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("UTF-8"));
        channel.basicPublish("", TASK_QUEUE_NAME,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                message.getBytes("UTF-8"));
        channel.basicPublish("", TASK_QUEUE_NAME,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                message.getBytes("UTF-8"));
        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();
    }
}

注意：這里getMessage方法，如果在運行的配置參數中添加了輸入參數，則使用輸入參數，如果沒有填寫，則使用默認值"Hello World"。
填寫輸入參數的方法是在如下圖位置寫上輸入參數

我們執行發送端代碼，向隊列"task_queue"中塞入4條消息

從這個動態圖片可以發現，通過發送端一次性發送了4條消息。

接收端

package com.ximalaya.openapi.rabbitmq.work;

import com.rabbitmq.client.*;

import java.io.IOException;
/**
 * Created by jackie on 17/8/4.
 */
public class Worker {

    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.3.161");
        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);
                }
            }
        };
        channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
    }

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

注意：這里的doWork方法，該方法當遇到"."是就會睡眠2秒鐘，所以像"hello..."這樣的消息就會睡眠6秒。
下面分兩種情況來看接收端的處理信息的情況
一個消費者
如果此時只運行一個接收端的代碼，說明只啟動了一個Consumer，我們看看消息的消費過程