ApsaraMQ for RabbitMQ:Spring 統合

接続とチャネルのベストプラクティス

ApsaraMQ for RabbitMQ は、バックエンドで分散型のマルチノードアーキテクチャーを使用します。サーバー側での単一ノードのパフォーマンスボトルネックを回避するために、クライアント接続を設定する際には、以下のベストプラクティスに従ってください。

• CONNECTION モードの使用：クライアントがサーバーへの複数の TCP 接続を確立するように設定します。CONNECTION モードでは、単一のクライアントがバックエンドの複数のサービスノードに接続を分散させ、クラスター内の複数のノードのリソースを利用します。クライアントあたりの接続数 (CONNECTION) は、クライアントのトラフィック量に基づいて調整し、数十個程度に設定することを推奨します。

• 頻繁な接続の確立と切断の回避：頻繁に接続を作成したり切断したりしないでください。接続の確立と切断を頻繁に繰り返すことは、メッセージの送受信と消費の効率を低下させます。操作ごとに接続を作成するのではなく、永続的な接続を維持してください。

• 接続ごとのチャネル数の制限：単一の接続で作成できるチャネルの数を 50 以下に設定します。単一の接続にチャネルが多すぎると、TCP トラフィックが急増する可能性があります。時折発生するパケットの並べ替えと相まって、Linux Netfilter Conntrack メカニズムがパケットを INVALID として分類し、クライアントが RST を送信して接続をリセットする原因となる場合があります。

次のパラメーターを使用して、接続モードと CONNECTION/CHANNEL 数を設定できます。

• spring.rabbitmq.cache.connection.mode - CONNECTION モードにするには、connection に設定します。

• spring.rabbitmq.cache.connection.size - 接続数。推奨: 数十。

• spring.rabbitmq.cache.channel.size - 接続あたりのチャネル数。推奨： 50 以下。

メッセージの生成

RabbitMQService で、依存性注入を介して RabbitTemplate を取得し、その send メソッドを呼び出してメッセージを送信します。

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Service
public class RabbitMQService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String exchange, String routingKey, String content) {
        // MessageId を設定します。
        String msgId = UUID.randomUUID().toString();
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setMessageId(msgId);
        // メッセージを作成します。
        Message message = new Message(content.getBytes(StandardCharsets.UTF_8), messageProperties);
        /*
         * send() メソッドを呼び出してメッセージを送信します。
         * exchange：エクスチェンジの名前。
         * routingKey：ルーティングキー。
         * message：メッセージの内容。
         * correlationData はパブリッシャー確認に使用されます。
         */
        rabbitTemplate.send(exchange, routingKey, message, null);
    }
}

メッセージの消費

@RabbitListener アノテーションを使用して、メッセージを消費します：

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Component
public class MessageListener {

     /**
     * メッセージを受信します。
     * @param message メッセージ。
     * @param channel チャネル。
     * @throws IOException
     * queues を作成したキューの名前に置き換えます。
     */
    @RabbitListener(queues = "myQueue")
    public void receiveFromMyQueue(Message message, Channel channel) throws IOException {
        // ここにメッセージ消費のビジネスロジックを記述します。
        ...
        // Ack の有効期間 (消費タイムアウト) 内に Ack を返す必要があります。そうしないと、確認が無効になり、メッセージが再配信されます。
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

以下に、RabbitListener の一般的なオプション設定を示します。

重要事項

クライアントがメッセージの消費に RabbitTemplate の receiveAndConvert メソッドを使用しているかどうかを確認します。

このメソッドは、キュー名やタイムアウト期間などのパラメーターを受け入れます。タイムアウトが指定されている場合、内部処理ロジックには、メッセージを消費する前にまず BasicCancel コマンドを送信してコンシューマーサブスクリプションをキャンセルし、次に Ack を送信してメッセージを確認するという問題があります。

しかし、ApsaraMQ for RabbitMQ が BasicCancel リクエストを受信すると、対応するコンシューマーを直ちにキャンセルし、未確認のメッセージをすべて再キューイングします。このため、クライアントから送信された後続の Ack は事実上無効となり、サーバーはこれを有効な確認応答として扱わずにこれらのメッセージを再配信することで、クライアント側で重複消費が発生します。

さらに、キューのメッセージスループットが高い場合、キャンセルと再キューイングの操作が頻繁に行われると、メッセージの蓄積につながり、消費の安定性にさらに影響を及ぼす可能性があります。