Go でデッド RabbitMQ 接続を適切に処理する方法

Go でのデッド RabbitMQ 接続の検出

このシナリオでは、streadway/amqp ライブラリを使用して Go で記述されたコンシューマ スクリプトは、次の場合に終了に失敗します。 RabbitMQ サーバーが停止しています。さらに、サーバーが再起動されると、コンシューマはメッセージの受信を停止します。

解決策: 接続通知の使用

amqp.Connection タイプには、トランスポートまたはプロトコルのエラーを通知するためのチャネル。このメソッドを使用すると、接続ドロップ イベントを処理し、それらを適切に処理できます。

次のコード スニペットは、接続通知を使用して切断された接続を検出し、必要なアクションを実行する方法を示しています。

import (
    "log"
    "time"

    "github.com/streadway/amqp"
)

func main() {
    for {  // Reconnection loop
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") // Setup
        if err != nil {
            log.Fatal(err)
        }
        notify := conn.NotifyClose(make(chan *amqp.Error)) // Error channel

        ch, err := conn.Channel()
        if err != nil {
            log.Fatal(err)
        }
        msgs, err := ch.Consume(
            "test_task_queue", // Queue
            "",     // Consumer
            false,  // Auto-ack
            false,  // Exclusive
            false,  // No-local
            false,  // No-wait
            nil,    // Args
        )
        if err != nil {
            log.Fatal(err)
        }

        for {  // Receive loop
            select {  // Check connection
            case err = <-notify:
                // Work with error
                log.Println(err)
                break // Reconnect
            case d := <-msgs:
                // Work with message
                log.Printf("Received a message: %s", d.Body)
                d.Ack(false)
                dotCount := bytes.Count(d.Body, []byte("."))
                t := time.Duration(dotCount)
                time.Sleep(t * time.Second)
                log.Printf("Done")
            }
        }
    }
}

この変更されたスクリプトでは、通知チャネルを使用して接続の終了を継続的にチェックするループ内にメインのコンシューマ ロジックをラップしました。エラーを受信すると、エラーが記録され、プロセスが再接続します。

接続通知を利用することで、切断された RabbitMQ 接続を検出し、再接続、エラーの記録、スクリプトの終了などの適切なアクションを実行できます。

以上がGo でデッド RabbitMQ 接続を適切に処理する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。