Golang の RabbitMQ を使用した複数のメッセージモードの比較と選択-Golang-php.cn

Golang の RabbitMQ を使用した複数のメッセージモードの比較と選択

WBOY

リリース： 2023-09-28 12:10:44

オリジナル

1356 人が閲覧しました

Golang の RabbitMQ を使用した複数のメッセージモードの比較と選択

Golang で RabbitMQ を使用した複数のメッセージモードの比較と選択

はじめに:
分散システムでは、メッセージキューは、メッセージ送信者とメッセージを分離するための一般的な通信メカニズムです。受信機と非同期通信を可能にします。現在最も人気のあるメッセージキューの 1 つである RabbitMQ は、開発者が選択できるさまざまなメッセージモードを提供します。この記事では、RabbitMQ の 4 つの古典的なメッセージモード、つまりシンプルキュー、ワークキュー、パブリッシュ/サブスクライブモード、およびトピックモードを比較し、それらの特性と適用可能なシナリオを分析し、Golang のサンプルコードを示します。

1. シンプルキュー (シンプルキュー)

シンプルキューは、コンシューマにメッセージを送信する RabbitMQ の最も基本的なメッセージモードです。メッセージはキューに送信され、コンシューマーによって順番に読み取られます。

機能:

メッセージは 1 人のコンシューマによってのみ使用できます。
同じキューをリッスンしている複数のコンシューマーがいる場合、メッセージは各コンシューマーに均等に配信されます。
処理速度が速い消費者は、より多くのメッセージを消費します。

該当するシナリオ:

ログ収集、タスク分散など、タスクまたはメッセージを複数のワークユニットに分散する必要があるアプリケーションシナリオ。

サンプルコード:

package main

import (
    "log"

    "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "simple_queue",
        false,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare a queue")

    msgs, err := ch.Consume(
        q.Name,
        "",
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to register a consumer")

    forever := make(chan bool)

    go func() {
        for d := range msgs {
            log.Printf("Received a message: %s", d.Body)
        }
    }()

    log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
    <-forever
}

ログイン後にコピー

2. ワークキュー (ワークキュー)

ワークキューモードは、複数のコンシューマを介したメッセージの負荷分散メカニズムです。キュー内のメッセージを処理します。ワークキューモードを使用すると、メッセージはキューに送信され、コンシューマによって順番に取得および処理されます。

機能:

メッセージは 1 人のコンシューマによってのみ処理できます。
各コンシューマによって処理されるタスクは比較的均等です。つまり、処理速度が速いコンシューマはより多くのメッセージを処理します。

該当するシナリオ:

画像処理、ビデオトランスコーディングなどのバックグラウンドタスク処理。

サンプルコード:

package main

import (
    "log"
    "os"
    "strconv"
    "strings"

    "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "work_queue",
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare a queue")

    body := bodyFrom(os.Args)
    err = ch.Publish(
        "",
        q.Name,
        false,
        false,
        amqp.Publishing{
            DeliveryMode: amqp.Persistent,
            ContentType:  "text/plain",
            Body:         []byte(body),
        })
    failOnError(err, "Failed to publish a message")

    log.Printf(" [x] Sent %s", body)
}

func bodyFrom(args []string) string {
    var s string
    if (len(args) < 2) || os.Args[1] == "" {
        s = "Hello, World!"
    } else {
        s = strings.Join(args[1:], " ")
    }
    return strconv.Itoa(os.Getpid()) + ":" + s
}

ログイン後にコピー

3. パブリッシュ/サブスクライブモード (パブリッシュ/サブスクライブ)

パブリッシュ/サブスクライブモードでは、メッセージはすべてのサブスクライバーにブロードキャストされます。。すべての購読者は同じメッセージを受信します。

特徴:

各メッセージはすべての購読者にブロードキャストされます。
サブスクライバが異なれば、メッセージ処理ロジックも異なる場合があります。

該当するシナリオ:

ログブロードキャスト、通知ブロードキャストなどのブロードキャストメッセージ。

サンプルコード:

package main

import (
    "log"

    "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    err = ch.ExchangeDeclare(
        "logs",
        "fanout",
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare an exchange")

    q, err := ch.QueueDeclare(
        "",
        false,
        false,
        true,
        false,
        nil,
    )
    failOnError(err, "Failed to declare a queue")

    err = ch.QueueBind(
        q.Name,
        "",
        "logs",
        false,
        nil,
    )
    failOnError(err, "Failed to bind a queue")

    msgs, err := ch.Consume(
        q.Name,
        "",
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to register a consumer")

    forever := make(chan bool)

    go func() {
        for d := range msgs {
            log.Printf("Received a message: %s", d.Body)
        }
    }()

    log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
    <-forever
}

ログイン後にコピー

4. トピックモード (トピック)

トピックモードは、トピックのワイルドカードルールを使用する、より複雑なメッセージモードです。 to メッセージは、一致するトピックのサブスクライバーに送信されます。

機能:

メッセージは、トピックの一致ルールに従ってルーティングされます。
ワイルドカード形式でのトピックの一致をサポートします。
さまざまな購読者が、興味のあるトピックに応じて購読できます。

該当するシナリオ:

トピックに基づいたメッセージのフィルタリングとルーティングが必要なシナリオ。

サンプルコード:

package main

import (
    "log"
    "os"

    "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    err = ch.ExchangeDeclare(
        "direct_logs",
        "direct",
        true,
        false,
        false,
        false,
        nil,
    )
    failOnError(err, "Failed to declare an exchange")

    severity := severityFrom(os.Args)
    body := bodyFrom(os.Args)

    err = ch.Publish(
        "direct_logs",
        severity,
        false,
        false,
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(body),
        },
    )
    failOnError(err, "Failed to publish a message")

    log.Printf(" [x] Sent %s", body)
}

func severityFrom(args []string) string {
    var severity string
    if len(args) < 3 || os.Args[2] == "" {
        severity = "info"
    } else {
        severity = os.Args[2]
    }
    return severity
}

func bodyFrom(args []string) string {
    var s string
    if len(args) < 4 || os.Args[3] == "" {
        s = "Hello, World!"
    } else {
        s = strings.Join(args[3:], " ")
    }
    return s
}

ログイン後にコピー

概要:
高性能メッセージキューシステムとして、RabbitMQ にはさまざまなシナリオのニーズを満たす豊富なメッセージモードがあります。実際のビジネスニーズに応じて、対応するメッセージモードを選択できます。この記事では、シンプルキュー、ワークキュー、パブリッシュ/サブスクライブモード、トピックモードの 4 つの典型的なメッセージモードを比較し、対応する Golang サンプルコードを示します。開発者は、ニーズに基づいて適切なメッセージモードを選択して分散システムを構築できます。

以上がGolang の RabbitMQ を使用した複数のメッセージモードの比較と選択の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。