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 message = new Message(content.getBytes(StandardCharsets.UTF_8), messageProperties);
        /*
         * send() メソッドを呼び出してメッセージを送信します。
         * exchange: エクスチェンジの名前。
         * routingKey: ルーティングキー。
         * message: メッセージの内容。
         * correlationData は Publisher Confirm に使用されます。
         */
        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 が実質的に無効であることを意味します。サーバーはそれを有効な確認応答として扱わず、代わりにこれらのメッセージを再配信し、クライアント側で重複消費を引き起こします。

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