CLOVER🍀

That was when it all began.

RabbitMQで、キューにConsumerを複数つけて動かす

RabbitMQを、チュートリアルに沿って勉強していこうという流れで。前回はインストールとHello Worldでしたが、今回はこちら。

RabbitMQ - RabbitMQ tutorial - Work Queues

チュートリアルの最初(RabbitMQ - RabbitMQ tutorial - "Hello World!")では、キューに対してメッセージを投げる役(Producer)と受ける役(Consumer)がひとりずつでしたが、今回はメッセージを受け取る役が2つとなります。

これで、複数のWorker(というかConsumer)がキューに入れられたタスクを受け取って、それぞれ作業を行う…ということができるというわけですね。まあ、チュートリアル自体はとても簡単なものですが。

準備

RabbitMQのサーバーと、アクセス可能できるユーザーは作成済みとします。

今回はJavaクライアントからアクセスするので、Maven依存関係には以下のようにクライアントライブラリを追加します。

        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>3.6.2</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.assertj</groupId>
            <artifactId>assertj-core</artifactId>
            <version>3.4.1</version>
            <scope>test</scope>
        </dependency>

JUnitとAssertJはテスト用です。

テストコードの雛形作成

では、最初にテストコードの雛形を書いてみます。
src/test/java/org/littlewings/rabbitmq/workqueues/WorkQueuesTest.java

package org.littlewings.rabbitmq.workqueues;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.MessageProperties;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import static org.assertj.core.api.Assertions.assertThat;

public class WorkQueuesTest {
    // ここに、テストを書く!
}

利用するimport文は、だいたいこのあたりですね、と。

そして、メッセージを2通投げて、2つのスレッドでそれぞれ動かすConsumerでメッセージを受け取るコードを書いてみます。

    @Test
    public void workQueues() throws InterruptedException, ExecutionException, IOException, TimeoutException {
        String firstMessage = "Hello World!!";
        String secondMessage = "Hello RabbitMQ!!";

        Set<String> receivedMessages = new HashSet<>(Arrays.asList(firstMessage, secondMessage));

        // setup workers
        ExecutorService es = Executors.newFixedThreadPool(2);
        Future<String> f1 = es.submit(this::receive);
        Future<String> f2 = es.submit(this::receive);

        // messages send
        send(firstMessage);
        send(secondMessage);

        receivedMessages.remove(f1.get());
        receivedMessages.remove(f2.get());

        assertThat(receivedMessages).isEmpty();

        es.shutdown();
        es.awaitTermination(10L, TimeUnit.SECONDS);
    }

受け取り側は、受信したメッセージを返すようにして、合わせて2通のメッセージが個々のConsumerで受け取れたかどうかを確認します。

1通目は「Hello World!!」で、2通目は「Hello RabbitMQ!!」です。

受信側(Consumer)

今回は、先に受信側からコードを載せていきます。

    public String receive() throws IOException, InterruptedException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("kazuhira");
        factory.setPassword("password");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

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

        boolean durable = true;
        channel.queueDeclare("my-queue", durable, false, false, null);

        AtomicInteger counter = new AtomicInteger();
        AtomicReference<String> receivedMessage = new AtomicReference<>();

        boolean autoAck = false;
        channel.basicConsume("my-queue", autoAck, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    assertThat(envelope.getRoutingKey()).isEqualTo("my-queue");

                    receivedMessage.set(new String(body, StandardCharsets.UTF_8));
                    counter.incrementAndGet();

                    try {
                        TimeUnit.SECONDS.sleep(3L);
                    } catch (InterruptedException e) {
                    }
                } finally {
                    boolean multiple = false;
                    channel.basicAck(envelope.getDeliveryTag(), multiple);
                }
            }
        });

        TimeUnit.SECONDS.sleep(5L);

        assertThat(counter.get()).isEqualTo(1);

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

        return receivedMessage.get();
    }

このメソッドで今回は2つの受信側、Consumerを使うわけですが、デフォルトではメッセージの受信はラウンド・ロビンで行われるようです。

前回のHello World時のコードと変更があるのは、まずはChannel#basicConsumeの第2引数(autoAck)をfalseをして

        boolean autoAck = false;
        channel.basicConsume("my-queue", autoAck, new DefaultConsumer(channel) {

Consumer#handleDelivery中で、メッセージを処理し終わったらChannel#basicAckを呼び出していること。

                    boolean multiple = false;
                    channel.basicAck(envelope.getDeliveryTag(), multiple);

これで、Consumerがメッセージを処理し終わったので、RabbitMQ側にメッセージを削除してよいことを表すAckを送信するようです。
ConsumerからAckが送られてこない場合は、再度別のConsumerにRabbitMQ側からメッセージを再送するのだとか。ただ、再送のタイミングはConsumerが落ちた時のようで、それには時間がかかるかも…と。

メッセージがいずれかのConsumerで処理され、失われないようにする仕組みってことですね。

前回はautoAckがtrueだったので、メッセージがConsumerに渡ったらAckをConsumerから明示的に送らずともRabbitMQから削除されてしまう、ということなのでしょう。

続いて、Channel#queueDeclareで第2引数(durable)をtrueに。

        boolean durable = true;
        channel.queueDeclare("my-queue", durable, false, false, null);

こうすることで、RabbitMQサーバーが停止しても、メッセージが失われないようにすることができるようです。なお、このコードはConsumer側でもProducer側でも設定する必要があります。

そして、最後にChannel#basicQosの設定。

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

デフォルトでは、RabbitMQはConsumerに均等にメッセージを送るようですが、Consumer側の負荷には偏りがある可能性があります。ここで、処理中のConsumerにメッセージを送信しないように、代わりに次のConsumerにメッセージを送信するようにするために入れる設定が、こちらのbasicQosのようです。これで、あるメッセージを複数のConsumerに同時に伝えることがなくなる、と。

送信側(Producer)

送信側(Producer)のコードは、こんな感じで作成。

    public void send(String message) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("kazuhira");
        factory.setPassword("password");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        boolean durable = true;
        channel.queueDeclare("my-queue", durable, false, false, null);

        channel.basicPublish("", "my-queue", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));

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

基本的にはHello Worldの時と変わりませんが、Channel#queueDeclareの第2引数(durable)をtrueにし、Channel#basicPublishでメッセージを送る時に第3引数にMessageProperties.PERSISTENT_TEXT_PLAINを指定しています。


Channel#queueDeclareのdurableについてはConsumer側と同じですが、メッセージが失われないようにどのようにRabbitMQが保存(Persistence)するのかを指定するようです。

        channel.basicPublish("", "my-queue", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));

今回は、テキストですと。他には、Basicとかあるようです。

Producer側の変更点は、このくらいです。

まとめ

今回は、複数のConsumerを使ったプログラムを書いてみましたが、合わせてAckやメッセージの永続化(durable)についても知ることができました。

まだキューに慣れないので、感覚的に掴みづらい部分もけっこうあるのですが、気長にチュートリアルをこなして頑張ってみようと思います。