Golang で RabbitMQ を使用した複數(shù)のメッセージ モードの比較と選択

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

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

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

機(jī)能:

1. メッセージは 1 人のコンシューマによってのみ使用できます。
2. 同じキューをリッスンしている複數(shù)のコンシューマーがいる場(chǎng)合、メッセージは各コンシューマーに均等に配信されます。
3. 処理速度が速い消費(fèi)者は、より多くのメッセージを消費(fèi)します。

該當(dāng)するシナリオ:

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

サンプル コード:

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. ワーク キュー (ワーク キュー)

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

機(jī)能:

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

該當(dāng)するシナリオ:

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. パブリッシュ/サブスクライブ モード (パブリッシュ/サブスクライブ)

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

特徴:

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

該當(dāng)するシナリオ:

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()

    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 メッセージは、一致するトピックのサブスクライバーに送信されます。

機(jī)能:

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

該當(dāng)するシナリオ:

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

サンプル コード:

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 サンプル コードを示します。開発者は、ニーズに基づいて適切なメッセージ モードを選択して分散システムを構(gòu)築できます。

以上がGolang の RabbitMQ を使用した複數(shù)のメッセージ モードの比較と選択の詳細(xì)內(nèi)容です。詳細(xì)については、PHP 中國(guó)語(yǔ) Web サイトの他の関連記事を參照してください。