Golang の RabbitMQ を使用した複数のメッセージ モードの比較と選択
- WBOYオリジナル
- 2023-09-28 12:10:441314ブラウズ
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 サイトの他の関連記事を参照してください。