1. 程式人生 > 實用技巧 >RabbitMQ Go客戶端教程5——topic

RabbitMQ Go客戶端教程5——topic

本文翻譯自RabbitMQ官網的Go語言客戶端系列教程,本文首發於我的個人部落格:liwenzhou.com,教程共分為六篇,本文是第五篇——topic。

這些教程涵蓋了使用RabbitMQ建立訊息傳遞應用程式的基礎知識。 你需要安裝RabbitMQ伺服器才能完成這些教程,請參閱安裝指南或使用Docker映象。 這些教程的程式碼是開源的,官方網站也是如此。

先決條件

本教程假設RabbitMQ已安裝並執行在本機上的標準埠(5672)。如果你使用不同的主機、埠或憑據,則需要調整連線設定。

topic交換器(主題交換器)

傳送到topic交換器的訊息不能具有隨意的routing_key——它必須是單詞列表,以點分隔。這些詞可以是任何東西,但通常它們指定與訊息相關的某些功能。一些有效的routing_key

示例:“stock.usd.nyse”,“nyse.vmw”,“quick.orange.rabbit”。routing_key中可以包含任意多個單詞,最多255個位元組。

繫結鍵也必須採用相同的形式。topic交換器背後的邏輯類似於direct交換器——用特定路由鍵傳送的訊息將傳遞到所有匹配繫結鍵繫結的佇列。但是,繫結鍵有兩個重要的特殊情況:

  • *(星號)可以代替一個單詞。
  • (井號)可以替代零個或多個單詞。

通過下面這個示例可以很容易看明白這一點:

img

在這個例子中,我們將傳送一些都是描述動物的資訊。將使用包含三個詞(兩個點)的路由金鑰傳送訊息。路由鍵中的第一個單詞將描述速度,第二個是顏色,第三個是種類:“<speed>.<colour>.<species>

”。

我們建立了三個繫結關係:Q1與繫結鍵“*.orange.*”繫結,Q2與“*.*.rabbit”和“lazy.#”繫結。

這些繫結可以總結為:

  • Q1對所有橙色動物都感興趣。
  • Q2想接收有關兔子(rabbit)的一切訊息,以及有關懶惰(lazy)動物的一切訊息。

路由鍵設定為“quick.orange.rabbit”的訊息將傳遞到兩個佇列。訊息“lazy.orange.elephant”也將傳送給他們兩個。另一方面,“quick.orange.fox”將僅進入第一個佇列,而“lazy.brown.fox”將僅進入第二個佇列。即使“lazy.pink.rabbit”與兩個繫結匹配(匹配Q2的兩個繫結),也只會傳遞到第二個佇列一次。 “quick.brown.fox

”與任何繫結都不匹配,因此將被丟棄。

如果我們打破約定併發送一個或四個單詞的訊息,例如“orange”或“quick.orange.male.rabbit”,會發生什麼?好吧,這些訊息將不匹配任何繫結,並且將會丟失。

另外,“lazy.orange.male.rabbit”即使有四個單詞,也將匹配最後一個繫結,並將其傳送到第二個佇列。

topic交換器

topic交換器功能強大,可以像其他交換器一樣執行。

當佇列用“#”(井號)繫結鍵繫結時,它將接收所有訊息,而與路由鍵無關,就像在fanout交換器中一樣。

當在繫結中不使用特殊字元“*”(星號)和“#”(井號)時,topic交換器的行為就像direct交換器一樣。

完整示例

我們將在日誌記錄系統中使用topic交換器。我們將從一個可行的假設開始,即日誌的路由鍵將包含兩個詞:“<facility>.<severity>”。

該程式碼與上一教程中的程式碼幾乎相同。

emit_log_topic.go的程式碼:

packagemain

import(
"log"
"os"
"strings"

"github.com/streadway/amqp"
)

funcfailOnError(errerror,msgstring){
iferr!=nil{
log.Fatalf("%s:%s",msg,err)
}
}

funcmain(){
conn,err:=amqp.Dial("amqp://guest:guest@localhost:5672/")
failOnError(err,"FailedtoconnecttoRabbitMQ")
deferconn.Close()

ch,err:=conn.Channel()
failOnError(err,"Failedtoopenachannel")
deferch.Close()

err=ch.ExchangeDeclare(
"logs_topic",//name
"topic",//type
true,//durable
false,//auto-deleted
false,//internal
false,//no-wait
nil,//arguments
)
failOnError(err,"Failedtodeclareanexchange")

body:=bodyFrom(os.Args)
err=ch.Publish(
"logs_topic",//exchange
severityFrom(os.Args),//routingkey
false,//mandatory
false,//immediate
amqp.Publishing{
ContentType:"text/plain",
Body:[]byte(body),
})
failOnError(err,"Failedtopublishamessage")

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

funcbodyFrom(args[]string)string{
varsstring
if(len(args)<3)||os.Args[2]==""{
s="hello"
}else{
s=strings.Join(args[2:],"")
}
returns
}

funcseverityFrom(args[]string)string{
varsstring
if(len(args)<2)||os.Args[1]==""{
s="anonymous.info"
}else{
s=os.Args[1]
}
returns
}

receive_logs_topic.go的程式碼:

packagemain

import(
"log"
"os"

"github.com/streadway/amqp"
)

funcfailOnError(errerror,msgstring){
iferr!=nil{
log.Fatalf("%s:%s",msg,err)
}
}

funcmain(){
conn,err:=amqp.Dial("amqp://guest:guest@localhost:5672/")
failOnError(err,"FailedtoconnecttoRabbitMQ")
deferconn.Close()

ch,err:=conn.Channel()
failOnError(err,"Failedtoopenachannel")
deferch.Close()

err=ch.ExchangeDeclare(
"logs_topic",//name
"topic",//type
true,//durable
false,//auto-deleted
false,//internal
false,//no-wait
nil,//arguments
)
failOnError(err,"Failedtodeclareanexchange")

q,err:=ch.QueueDeclare(
"",//name
false,//durable
false,//deletewhenunused
true,//exclusive
false,//no-wait
nil,//arguments
)
failOnError(err,"Failedtodeclareaqueue")

iflen(os.Args)<2{
log.Printf("Usage:%s[binding_key]...",os.Args[0])
os.Exit(0)
}
//繫結topic
for_,s:=rangeos.Args[1:]{
log.Printf("Bindingqueue%stoexchange%swithroutingkey%s",
q.Name,"logs_topic",s)
err=ch.QueueBind(
q.Name,//queuename
s,//routingkey
"logs_topic",//exchange
false,
nil)
failOnError(err,"Failedtobindaqueue")
}

msgs,err:=ch.Consume(
q.Name,//queue
"",//consumer
true,//autoack
false,//exclusive
false,//nolocal
false,//nowait
nil,//args
)
failOnError(err,"Failedtoregisteraconsumer")

forever:=make(chanbool)

gofunc(){
ford:=rangemsgs{
log.Printf("[x]%s",d.Body)
}
}()

log.Printf("[*]Waitingforlogs.ToexitpressCTRL+C")
<-forever
}

想要接收所有的日誌:

gorunreceive_logs_topic.go"#"

要從“kern”接收所有日誌:

go run receive_logs_topic.go "kern.*"

或者,如果你只想接收“critical”日誌:

gorunreceive_logs_topic.go"*.critical"

你可以建立多個繫結:

gorunreceive_logs_topic.go"kern.*""*.critical"

併發出帶有路由鍵“kern.critical”的日誌:

gorunemit_log_topic.go"kern.critical""Acriticalkernelerror"

你可以自己嘗試玩一下這個程式。請注意,程式碼沒有對路由鍵或繫結鍵進行任何假設,你可能希望使用兩個以上的路由鍵引數。

(關於emit_log_topic.goreceive_logs_topic.go的完整原始碼)

接下來,我們將在教程6中瞭解如何將往返訊息用作遠端過程呼叫。