主頁 >  其他 > RabbitMQ（Golang版本）

RabbitMQ（Golang版本）

2021-06-14 06:56:28 其他

一、RabbitMQ 簡介

訊息佇列是一種應用程式對應用程式的通信方法，應用程式通過讀寫出入佇列的訊息（針對應用程式的資料）來通信，而無需專用連接來鏈接它們，訊息傳遞指的是程式之間通過在訊息中發送資料進行通信，而不是通過直接呼叫彼此來通信，直接呼叫通常是/用于諸如遠程程序呼叫的技術，排隊指的是應用程式通過 佇列來通信，佇列的使用除去了接收和發送應用程式同時執行的要求，

二、使用場景

1.解耦

場景說明：用戶下單后，訂單系統需要通知庫存系統，傳統的做法是，訂單系統呼叫庫存系統的介面，

在這里插入圖片描述

傳統模式的缺點：

  • 如果庫存系統出現故障，訂單系統當收到訂單時也會出現問題
  • 訂單與庫存系統之間高耦合

采用訊息佇列后

在這里插入圖片描述

  • 訂單系統：用戶下單后，訂單系統完成持久化處理，將訊息寫入訊息佇列，回傳用戶訂單下單成功
  • 庫存系統：訂閱下單的訊息，采用拉/推的方式，獲取下單資訊，庫存系統根據下單資訊，進行庫存操作
  • 假如：在下單時庫存系統不能正常使用，也不影響正常下單，因為下單后，訂單系統寫入訊息佇列就不再關心其他的后續操作了，實作訂單系統與庫存系統的應用解耦
  • 為了保證庫存肯定有，可以將佇列大小設定成庫存數量，或者采用其他方式解決，

2.異步

場景說明：用戶注冊后，需要發注冊郵件和注冊短信，傳統的做法有兩種 1.串行的方式；2.并行方式

（1）串行方式：將注冊資訊寫入資料庫成功后，發送注冊郵件，再發送注冊短信，以上三個任務全部完成后，回傳給客戶端

在這里插入圖片描述

（2）并行方式：將注冊資訊寫入資料庫成功后，發送注冊郵件的同時，發送注冊短信，以上三個任務完成后，回傳給客戶端，與串行的差別是，并行的方式可以提高處理的時間

在這里插入圖片描述

（3）引入訊息佇列，將不是必須的業務邏輯，異步處理，

在這里插入圖片描述

3.流量削峰

一般應用于秒殺活動中，因為秒殺活動一般會因為流量過大，導致應用掛掉,為了解決這個問題，一般在應用前端加入訊息佇列，

作用: 1.可以控制活動人數，超過此一定閥值的訂單直接丟棄 2.可以緩解短時間的高流量壓垮應用

4.缺點

  • 系統的可用性降低

    例如系統只需要呼叫abc三個介面，現在引入了mq之后雖然之前呼叫都沒有出錯，系統可以正常運行，但是mq出現問題的話，整個系統也同樣會受影響

  • 系統的復雜性提高

    需要考慮訊息是否丟失，還需要考慮訊息傳遞的順序

  • 一致性問題

    系統發送完訊息回傳成功，但是abc中若有系統寫入失敗，就會產生資料不一致的問題

三、docker安裝rabbitmq

1.先將rabbitmq和erlang的軟體包上傳至阿里云服務器，然后解壓安裝

rpm -Uvh esl-erlang_23.0-1_centos_7_amd64.rpm
yum install erlang
yum install -y socat
rpm -Uvh rabbitmq-server-3.8.14-1.el7.noarch.rpm
yum install rabbit-server -y

2.啟動rabbitmq-server

systemctl start rabbitmq-server
systemctl status rabbitmq-server # 查看狀態，

3.添加超級用戶(web管理界面)

rabbitmqctl add_user admin admin        # 添加用戶  用戶名 密碼
 rabbitmqctl set_user_tags admin administrator        # 修改用戶權限 用戶名 權限
 rabbitmqctl change_password admin admin   # 修改密碼

4.運行rabbitmq

docker run -di --name myrabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 8030:15672 -p 8031:5672 -p 8032:25672 -p 8033:61613 -p 8034:1883 rabbitmq:management

四、作業模式

（一）.simple模式（即最簡單的收發模式）

在這里插入圖片描述

1.訊息產生訊息，將訊息放入佇列

2.訊息的消費者(consumer) 監聽 訊息佇列,如果佇列中有訊息,就消費掉,訊息被拿走后,自動從佇列中洗掉(隱患 訊息可能沒有被消費者正確處理,已經從佇列中消失了,造成訊息的丟失，這里可以設定成手動的ack,但如果設定成手動ack，處理完后要及時發送ack訊息給佇列，否則會造成記憶體溢位)，

//send.go

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() {
  //鏈接rabbitmq server
  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(
    "hello world", // 佇列名
    false,   // 持久化
    false,   // 是否自動洗掉
    false,   // 排他性
    false,   // no-wait
    nil,     // 附屬引數
  )
  FailOnError(err, "Failed to declare a queue")
  body := "Hello World!"
  err = ch.Publish(    // 發送訊息（生產者）
    "",     // exchange
    q.Name, // routing key
    false,  // mandatory
    false,  // immediate
    amqp.Publishing{
      ContentType: "text/plain",
      Body:        []byte(body),
    })
  FailOnError(err, "Failed to publish a message")
  log.Printf(" [x] Sent %s", body)
}


//receive.go
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() {
  //連接rabbitmq server
  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(
    "hello world", // name
    false,   // durable
    false,   // delete when unused
    false,   // exclusive
    false,   // no-wait
    nil,     // arguments
  )
  failOnError(err, "Failed to declare a queue")

  msgs, err := ch.Consume(  // 注冊一個消費者（接收訊息）
    q.Name, // queue
    "",     // consumer
    true,   // auto-ack
    false,  // exclusive
    false,  // no-local
    false,  // no-wait
    nil,    // args
  )
  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
}

（二）.work作業模式(資源的競爭)

在這里插入圖片描述

1.訊息產生者將訊息放入佇列消費者可以有多個,消費者1,消費者2同時監聽同一個佇列,訊息被消費，C1 C2共同爭搶當前的訊息佇列內容,誰先拿到誰負責消費訊息(隱患：高并發情況下,默認會產生某一個訊息被多個消費者共同使用,可以設定一個開關(syncronize) 保證一條訊息只能被一個消費者使用)，

2.訊息確認

完成一個任務需要消耗時間，如果一個消費者開始了一個任務并在完成期間關倍訓者死亡，我們就會丟失他剛剛處理的訊息，同時還將丟失所有已經派發給特定worker但未處理的資訊，我們可以使用訊息確認來避免這樣的事情發生，消費者發回一個 ack(nowledgement) 來告訴 RabbitMQ 一個特定的訊息已經被接收、處理并且 RabbitMQ 可以自由地洗掉它，如果消費者在沒有發送 ack 的情況下死亡（其通道關閉、連接關倍訓 TCP 連接丟失），RabbitMQ 將理解訊息未完全處理并將重新排隊，如果有其他消費者同時在線，它會迅速將其重新交付給另一個消費者，這樣您就可以確保不會丟失任何訊息，即使作業人員偶爾會死亡，

3.訊息持久性

RabbitMQ服務器停止，如果沒有持久化，任務就會丟失，我們只需要在宣告佇列時將durable引數設定為true，同時將amqp.Publishing中加入DeliveryMode: amqp.Persistent,即可實作持久化，這樣即使我們rabbit重啟，我們的佇列也不會丟失

//new_task.go

package main

import (
  "log"
  "os"
  "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 queues", // name
    true,          // durable
    false,         // delete when unused
    false,         // exclusive
    false,         // no-wait
    nil,           // arguments
  )
  failOnError(err, "Failed to declare a queue")

  body := bodyFrom(os.Args)

  err = ch.Publish(
    "",     // exchange  這里使用默認交換機,訊息是通過交換機傳給佇列
    q.Name, // routing key
    false,  // mandatory
    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)

}

//獲取運行時的引數
//例如：go run main.go 1 3 -X ?    輸出如下：
//引數0: /tmp/go-build116558042/command-line-arguments/_obj/exe/main
//引數1: 1
//引數2: 3
//引數3: -X
//引數4: ?
func bodyFrom(args []string) string {
  var s string
  if (len(args) < 2) || os.Args[1] == "" { //args長度小于2 或者引數為空的時候
    s = "hello"
  } else {
    s = strings.Join(args[1:], " ") //Join 連接其第一個引數的元素以創建單個字串，分隔符字串 sep 放置在結果字串中的元素之間
  }
  return s
}


//work.go
package main

import (
  "bytes"
  "log"
  "time"

  "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 queues", // name
    true,         // durable
    false,        // delete when unused
    false,        // exclusive
    false,        // no-wait
    nil,          // arguments
  )
  FailOnError(err, "Failed to declare a queue")

  err = ch.Qos(
    1,     // prefetch count 服務器將在收到確認之前將那么多訊息傳遞給消費者，
    0,     // prefetch size  服務器將嘗試在收到消費者的確認之前至少將那么多位元組的交付保持重繪到網路
    false, // 當 global 為 true 時，這些 Qos 設定適用于同一連接上所有通道上的所有現有和未來消費者，當為 false 時，Channel.Qos 設定將應用于此頻道上的所有現有和未來消費者
  )
  FailOnError(err, "Failed to set QoS")

  msgs, err := ch.Consume(
    q.Name, // queue
    "",     // consumer
    false,  // auto-ack
    false,  // exclusive
    false,  // no-local
    false,  // no-wait
    nil,    // args
  )
  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)
      // func Count(s, sep [] byte ) int  計算s中sep的非重疊實體數，如果sep是空切片，則Count回傳1+s中UTF-8編碼的代碼點數
      dotCount := bytes.Count(d.Body, []byte("."))  // 回傳 . 的個數
      t := time.Duration(dotCount)  // 表示為 int64 納秒計數
      time.Sleep(t * time.Second)  //將當前 goroutine 暫停至少持續時間d
      log.Printf("Done")
      d.Ack(false)     // 必須在成功處理此交付后呼叫 Delivery.Ack,如果autoAck為true則不需要. 引數 true表示回復當前信道所有未回復的ack，用于批量確認，false表示回復當前條目
    }
  }()

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

}

啟動兩個終端用于執行work.go，一個終端用來執行new_task，相當于newtask是生產者，work是兩個消費者，兩個消費者共同消費訊息，（輪詢訪問，每個消費者消費的次數相同，公平分發，處理速度快的消費者消費的多，能者多勞）

（三）.publish/subscribe發布訂閱(共享資源)

在這里插入圖片描述

1、每個消費者監聽自己的佇列；

2、生產者將訊息發給broker，由交換機將訊息轉發到系結此交換機的每個佇列，每個系結交換機的佇列都將接收到訊息，

//emit_log.go
package main

import (
  "log"
  "os"
  "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()

  err = ch.ExchangeDeclare(
    "logs",   // name  交換機名字
    "fanout", // type  交換機模式 廣播到每個佇列
    true,     // durable  持久化
    false,    // auto-deleted
    false,    // internal
    false,    // no-wait
    nil,      // arguments
  )
  failOnError(err, "Failed to declare an exchange")

  body := bodyFrom(os.Args)
  err = ch.Publish(
    "logs", // exchange  生產者將訊息發往logs交換機
    "",     // routing key
    false,  // mandatory
    false,  // immediate
    amqp.Publishing{
      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"
  } else {
    s = strings.Join(args[1:], " ")
  }
  return s
}

//receive_log.go
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",   // name
    "fanout", // type
    true,     // durable
    false,    // auto-deleted
    false,    // internal
    false,    // no-wait
    nil,      // arguments
  )
  FailOnError(err, "Failed to declare an exchange")

  q, err := ch.QueueDeclare(
    "",    // name
    false, // durable
    false, // delete when unused
    true,  // exclusive
    false, // no-wait
    nil,   // arguments
  )
  FailOnError(err, "Failed to declare a queue")

  err = ch.QueueBind(  //宣告的佇列要和交換機系結
    q.Name, // queue name
    "",     // routing key
    "logs", // exchange
    false,
    nil)
  FailOnError(err, "Failed to bind a queue")

  msgs, err := ch.Consume(
    q.Name, // queue
    "",     // consumer
    true,   // auto-ack
    false,  // exclusive
    false,  // no-local
    false,  // no-wait
    nil,    // args
  )
  FailOnError(err, "Failed to register a consumer")

  forever := make(chan bool)

  go func() {
    for d := range msgs {
      log.Printf(" [x] %s", d.Body)
    }
  }()

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


//測驗
//啟動兩個消費者,打開兩個終端
go run receive_log.go 
//啟動一個生產者
 go run emit_log.go // 看到兩個消費者的終端輸出hello
 go run emit_log.go 1  // 兩個消費者的終端輸出1

（四）.routing路由模式

在這里插入圖片描述

1.訊息生產者將訊息發送給交換機按照路由判斷,路由是字串(info) 當前產生的訊息攜帶路由字符(物件的方法),交換機根據路由的key,只能匹配上路由key對應的訊息佇列,對應的消費者才能消費訊息;

2.根據業務功能定義路由字串

3.從系統的代碼邏輯中獲取對應的功能字串,將訊息任務扔到對應的佇列中，

4.業務場景:error 通知;EXCEPTION;錯誤通知的功能;傳統意義的錯誤通知;客戶通知;利用key路由,可以將程式中的錯誤封裝成訊息傳入到訊息佇列中,開發者可以自定義消費者,實時接收錯誤;

//emit_log_direct.go
package main

import (
  "log"
  "os"
  "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()

  err = ch.ExchangeDeclare(
    "logs_direct", // name
    "direct",      // direct交換機將會對binding key和routing key進行精確匹配，從而確定訊息該分發到哪個佇列
    true,          // durable
    false,         // auto-deleted
    false,         // internal
    false,         // no-wait
    nil,           // arguments
  )
  failOnError(err, "Failed to declare an exchange")

  body := bodyFrom(os.Args)
  err = ch.Publish(
    "logs_direct",         // exchange
    severityFrom(os.Args), // routing key
    false, // mandatory
    false, // immediate
    amqp.Publishing{
      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) < 3) || os.Args[2] == "" {
    s = "hello"
  } else {
    s = strings.Join(args[2:], " ")
  }
  return s
}

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


//receive_logs_direct.go
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(
    "logs_direct", // name
    "direct",      // type
    true,          // durable
    false,         // auto-deleted
    false,         // internal
    false,         // no-wait
    nil,           // arguments
  )
  FailOnError(err, "Failed to declare an exchange")

  q, err := ch.QueueDeclare(
    "",    // name
    false, // durable
    false, // delete when unused
    true,  // exclusive
    false, // no-wait
    nil,   // arguments
  )
  FailOnError(err, "Failed to declare a queue")

  // 命令輸入的時候如果引數小于2 證明沒有傳引數info，warning，error
  if len(os.Args) < 2 {
    log.Printf("Usage: %s [info] [warning] [error]", os.Args[0])
    os.Exit(0)
  }

  //給每一個引數創建一個系結
  for _, s := range os.Args[1:] {
    log.Printf("Binding queue %s to exchange %s with routing key %s", q.Name, "logs_direct", s)
    err = ch.QueueBind(
      q.Name,        // 佇列名稱
      s,             // 將引數作為 routing key
      "logs_direct", // exchange
      false,
      nil)
    FailOnError(err, "Failed to bind a queue")
  }

  msgs, err := ch.Consume(
    q.Name, // queue
    "",     // consumer
    true,   // auto ack
    false,  // exclusive
    false,  // no local
    false,  // no wait
    nil,    // args
  )
  FailOnError(err, "Failed to register a consumer")

  forever := make(chan bool)

  go func() {
    for d := range msgs {
      log.Printf(" [x] %s", d.Body)
    }
  }()

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

測驗：

go run receive_logs_direct.go info warning error

go run emit_log_direct.go error haha

可以在receive終端中看到輸出haha，route key為error

（五）.topic 主題模式(路由模式的一種)

在這里插入圖片描述

1.星號井號代表通配符

2.星號代表多個單詞,井號代表一個單詞

3.路由功能添加模糊匹配

4.訊息產生者產生消息,把訊息交給交換機

5.交換機根據key的規則模糊匹配到對應的佇列,由佇列的監聽消費者接收訊息消費

（在我的理解看來就是routing查詢的一種模糊匹配，就類似sql的模糊查詢方式）

//emit_log_topic.go
package main

import (
  "log"
  "os"
  "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()

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

  body := bodyFrom(os.Args)
  err = ch.Publish(
    "logs_topic",          // exchange
    severityFrom(os.Args), // routing key
    false, // mandatory
    false, // immediate
    amqp.Publishing{
      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) < 3) || os.Args[2] == "" {
    s = "hello"
  } else {
    s = strings.Join(args[2:], " ")
  }
  return s
}

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

//receive_logs_topic.go
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(
    "logs_topic", // name
    "topic",      // type
    true,         // durable
    false,        // auto-deleted
    false,        // internal
    false,        // no-wait
    nil,          // arguments
  )
  FailOnError(err, "Failed to declare an exchange")

  q, err := ch.QueueDeclare(
    "",    // name
    false, // durable
    false, // delete when unused
    true,  // exclusive
    false, // no-wait
    nil,   // arguments
  )
  FailOnError(err, "Failed to declare a queue")

  if len(os.Args) < 2 {
    log.Printf("Usage: %s [binding_key]...", os.Args[0])
    os.Exit(0)
  }

  for _, s := range os.Args[1:] {
    log.Printf("Binding queue %s to exchange %s with routing key %s", q.Name, "logs_topic", s)
    err = ch.QueueBind(
      q.Name,       // queue name
      s,            // routing key
      "logs_topic", // exchange
      false,
      nil)
    FailOnError(err, "Failed to bind a queue")
  }

  msgs, err := ch.Consume(
    q.Name, // queue
    "",     // consumer
    true,   // auto ack
    false,  // exclusive
    false,  // no local
    false,  // no wait
    nil,    // args
  )
  FailOnError(err, "Failed to register a consumer")

  forever := make(chan bool)

  go func() {
    for d := range msgs {
      log.Printf(" [x] %s", d.Body)
    }
  }()

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

測驗：

打開兩個終端，一個是從kern中接收日志，一個是從critical中接收日志

go run receive_logs_topic.go "kern.*"
go run receive_logs_topic.go "*.critical"

打開一個終端，使用kern.critical作為route key

go run emit_log_topic.go "kern.critical" lalalalalala

可以在上面兩個終端中輸出日志資訊lalalalalala

五、RabbitMQ高級特性-TTL佇列/訊息TTL-死信佇列

1、TTL

  • time to live 訊息存活時間

  • RabbitMQ支持訊息的過期時間, 在訊息發送時可以進行指定

  • 如果訊息在存活時間內未被消費，則會被清除

  • RabbitMQ支持兩種ttl設定

    • 單獨訊息進行配置ttl
    • 整個佇列進行配置ttl（居多）
  • //整個佇列進行配置ttl
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() {
  //鏈接rabbitmq server
  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()

  //使用x-message-ttl引數設定當前佇列中所有訊息的過期時間
  args := make(map[string]interface{})
  args["x-message-ttl"] = 5000


  //宣告一個佇列，用于訊息發送
  q, err := ch.QueueDeclare(
    "direct_ttl", // 佇列名
    false,   // 持久化
    false,   // 是否自動洗掉
    false,   // 排他性
    false,   // no-wait
    args,     // 設定ttl為5s
  )
  FailOnError(err, "Failed to declare a queue")
  body := "Hello World!"
  err = ch.Publish(    // 發送訊息（生產者）
    "",     // exchange
    q.Name, // routing key
    false,  // mandatory
    false,  // immediate
    amqp.Publishing{
      ContentType: "text/plain",
      Body:        []byte(body),
    })
  FailOnError(err, "Failed to publish a message")
  log.Printf(" [x] Sent %s", body)
}

  • 打開web管理界面可以看到features上顯示TTL，發送一條訊息，ready中會顯示1，在5s后自動清除，（放入死信佇列中，而對單條訊息設定ttl洗掉了就消失了），如圖：

  • [外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-W2jECIh4-1623400450231)(/Users/apple/Desktop/截圖/截屏2021-06-08 下午4.18.51.png)]

2、什么是rabbitmq的死信佇列/死信交換機

  • 沒有被及時消費的訊息存放的佇列

  • Dead Letter Exchange（死信交換機，縮寫： DLX）當訊息成為死信后，會被重新發送到另?個交換機，這個交換機就是DLX死信交換機，

  • 導致死信佇列的原因有：

    • 訊息被拒絕
    • 訊息過期
    • 佇列達到最大長度
    • 結果：訊息成為死信后，如果該佇列系結了死信交換機，則訊息會被死信交換機重新路由到死信佇列

在這里插入圖片描述
在這里插入圖片描述

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/287123.html

標籤：其他

上一篇：Vim入門教程

下一篇：Docker和K8S的比較（初探）

標籤雲
其他(157675) Python(38076) JavaScript(25376) Java(17977) C(15215) 區塊鏈(8255) C＃(7972) AI(7469) 爪哇(7425) MySQL(7132) html(6777) 基礎類(6313) sql(6102) 熊猫(6058) PHP(5869) 数组(5741) R(5409) Linux(5327) 反应(5209) 腳本語言(PerlPython)(5129) 非技術區(4971) Android(4554) 数据框(4311) css(4259) 节点.js(4032) C語言(3288) json(3245) 列表(3129) 扑(3119) C++語言(3117) 安卓(2998) 打字稿(2995) VBA(2789) Java相關(2746) 疑難問題(2699) 细绳(2522) 單片機工控(2479) iOS(2429) ASP.NET(2402) MongoDB(2323) 麻木的(2285) 正则表达式(2254) 字典(2211) 循环(2198) 迅速(2185) 擅长(2169) 镖(2155) 功能(1967) .NET技术(1958) Web開發(1951) python-3.x(1918) HtmlCss(1915) 弹簧靴(1913) C++(1909) xml(1889) PostgreSQL(1872) .NETCore(1853) 谷歌表格(1846) Unity3D(1843) for循环(1842)

熱門瀏覽
  • 網閘典型架構簡述

    網閘架構一般分為兩種：三主機的三系統架構網閘和雙主機的2+1架構網閘。 三主機架構分別為內端機、外端機和仲裁機。三機無論從軟體和硬體上均各自獨立。首先從硬體上來看，三機都用各自獨立的主板、記憶體及存盤設備。從軟體上來看，三機有各自獨立的作業系統。這樣能達到完全的三機獨立。對于“2+1”系統，“2”分為 ......

    uj5u.com 2020-09-10 02:00:44 more
  • 如何從xshell上傳檔案到centos linux虛擬機里

    如何從xshell上傳檔案到centos linux虛擬機里及：虛擬機CentOs下執行 yum -y install lrzsz命令，出現錯誤：鏡像無法找到軟體包 前言 一、安裝lrzsz步驟 二、上傳檔案 三、遇到的問題及解決方案 總結 前言 提示：其實很簡單，往虛擬機上安裝一個上傳檔案的工具 ......

    uj5u.com 2020-09-10 02:00:47 more
  • 一、SQLMAP入門

    一、SQLMAP入門 1、判斷是否存在注入 sqlmap.py -u 網址/id=1 id=1不可缺少。當注入點后面的引數大于兩個時。需要加雙引號， sqlmap.py -u "網址/id=1&uid=1" 2、判斷文本中的請求是否存在注入 從文本中加載http請求，SQLMAP可以從一個文本檔案中 ......

    uj5u.com 2020-09-10 02:00:50 more
  • Metasploit 簡單使用教程

    metasploit 簡單使用教程 浩先生， 2020-08-28 16:18:25 分類專欄： kail 網路安全 linux 文章標簽： linux資訊安全 編輯 著作權 metasploit 使用教程 前言 一、Metasploit是什么？ 二、準備作業 三、具體步驟 前言 Msfconsole ......

    uj5u.com 2020-09-10 02:00:53 more
  • 游戲逆向之驅動層與用戶層通訊

    驅動層代碼： #pragma once #include <ntifs.h> #define add_code CTL_CODE(FILE_DEVICE_UNKNOWN,0x800,METHOD_BUFFERED,FILE_ANY_ACCESS) /* 更多游戲逆向視頻www.yxfzedu.com ......

    uj5u.com 2020-09-10 02:00:56 more
  • 北斗電力時鐘（北斗授時服務器）讓網路資料更精準

    北斗電力時鐘（北斗授時服務器）讓網路資料更精準 北斗電力時鐘（北斗授時服務器）讓網路資料更精準 京準電子科技官微——ahjzsz 近幾年，資訊技術的得了快速發展，互聯網在逐漸普及，其在人們生活和生產中都得到了廣泛應用，并且取得了不錯的應用效果。計算機網路資訊在電力系統中的應用，一方面使電力系統的運行 ......

    uj5u.com 2020-09-10 02:01:03 more
  • 【CTF】CTFHub 技能樹 彩蛋 writeup

    ?碎碎念 CTFHub：https://www.ctfhub.com/ 筆者入門CTF時時剛開始刷的是bugku的舊平臺，后來才有了CTFHub。 感覺不論是網頁UI設計，還是題目質量，賽事跟蹤，工具軟體都做得很不錯。 而且因為獨到的金幣制度的確讓人有一種想去刷題賺金幣的感覺。 個人還是非常喜歡這個 ......

    uj5u.com 2020-09-10 02:04:05 more
  • 02windows基礎操作

    我學到了一下幾點 Windows系統目錄結構與滲透的作用 常見Windows的服務詳解 Windows埠詳解 常用的Windows注冊表詳解 hacker DOS命令詳解（net user / type /md /rd/ dir /cd /net use copy、批處理 等） 利用dos命令制作 ......

    uj5u.com 2020-09-10 02:04:18 more
  • 03.Linux基礎操作

    我學到了以下幾點 01Linux系統介紹02系統安裝，密碼啊破解03Linux常用命令04LAMP 01LINUX windows： win03 8 12 16 19 配置不繁瑣 Linux：redhat,centos(紅帽社區版)，Ubuntu server,suse unix:金融機構，證券，銀 ......

    uj5u.com 2020-09-10 02:04:30 more
  • 05HTML

    01HTML介紹 02頭部標簽講解03基礎標簽講解04表單標簽講解 HTML前段語言 js1.了解代碼2.根據代碼 懂得挖掘漏洞 （POST注入/XSS漏洞上傳）3.黑帽seo 白帽seo 客戶網站被黑帽植入劫持代碼如何處理4.熟悉html表單 <html><head><title>TDK標題，描述 ......

    uj5u.com 2020-09-10 02:04:36 more
最新发布
  • 2023年最新微信小程式抓包教程

    01 開門見山 隔一個月發一篇文章，不過分。 首先回顧一下《微信系結手機號資料庫被脫庫事件》，我也是第一時間得知了這個訊息，然后跟蹤了整件事情的經過。下面是這起事件的相關截圖以及近日流出的一萬條資料樣本： 個人認為這件事也沒什么，還不如關注一下之前45億快遞資料查詢渠道疑似在近日復活的訊息。 訊息是 ......

    uj5u.com 2023-04-20 08:48:24 more
  • web3 產品介紹：metamask 錢包 使用最多的瀏覽器插件錢包

    Metamask錢包是一種基于區塊鏈技術的數字貨幣錢包，它允許用戶在安全、便捷的環境下管理自己的加密資產。Metamask錢包是以太坊生態系統中最流行的錢包之一，它具有易于使用、安全性高和功能強大等優點。 本文將詳細介紹Metamask錢包的功能和使用方法。 一、 Metamask錢包的功能 數字資 ......

    uj5u.com 2023-04-20 08:47:46 more
  • vulnhub_Earth

    前言 靶機地址->>>vulnhub_Earth 攻擊機ip：192.168.20.121 靶機ip：192.168.20.122 參考文章 https://www.cnblogs.com/Jing-X/archive/2022/04/03/16097695.html https://www.cnb ......

    uj5u.com 2023-04-20 07:46:20 more
  • 從4k到42k，軟體測驗工程師的漲薪史，給我看哭了

    清明節一過，盲猜大家已經無心上班，在數著日子準備過五一，但一想到銀行卡里的余額……瞬間心情就不美麗了。最近，2023年高校畢業生就業調查顯示，本科畢業月平均起薪為5825元。調查一出，便有很多同學表示自己又被平均了。看著這一資料，不免讓人想到前不久中國青年報的一項調查：近六成大學生認為畢業10年內會 ......

    uj5u.com 2023-04-20 07:44:00 more
  • 最新版本 Stable Diffusion 開源 AI 繪畫工具之中文自動提詞篇

    🎈 標簽生成器 由于輸入正向提示詞 prompt 和反向提示詞 negative prompt 都是使用英文，所以對學習母語的我們非常不友好 使用網址：https://tinygeeker.github.io/p/ai-prompt-generator 這個網址是為了讓大家在使用 AI 繪畫的時候 ......

    uj5u.com 2023-04-20 07:43:36 more
  • 漫談前端自動化測驗演進之路及測驗工具分析

    隨著前端技術的不斷發展和應用程式的日益復雜，前端自動化測驗也在不斷演進。隨著 Web 應用程式變得越來越復雜，自動化測驗的需求也越來越高。如今，自動化測驗已經成為 Web 應用程式開發程序中不可或缺的一部分，它們可以幫助開發人員更快地發現和修復錯誤，提高應用程式的性能和可靠性。 ......

    uj5u.com 2023-04-20 07:43:16 more
  • CANN開發實踐：4個DVPP記憶體問題的典型案例解讀

    摘要：由于DVPP媒體資料處理功能對存放輸入、輸出資料的記憶體有更高的要求（例如，記憶體首地址128位元組對齊），因此需呼叫專用的記憶體申請介面，那么本期就分享幾個關于DVPP記憶體問題的典型案例，并給出原因分析及解決方法。 本文分享自華為云社區《FAQ_DVPP記憶體問題案例》，作者：昇騰CANN。 DVPP ......

    uj5u.com 2023-04-20 07:43:03 more
  • msf學習

    msf學習 以kali自帶的msf為例 一、msf核心模塊與功能 msf模塊都放在/usr/share/metasploit-framework/modules目錄下 1、auxiliary 輔助模塊，輔助滲透（埠掃描、登錄密碼爆破、漏洞驗證等） 2、encoders 編碼器模塊，主要包含各種編碼 ......

    uj5u.com 2023-04-20 07:42:59 more
  • Halcon軟體安裝與界面簡介

    1. 下載Halcon17版本到到本地 2. 雙擊安裝包后 3. 步驟如下 1.2 Halcon軟體安裝 界面分為四大塊 1. Halcon的五個助手 1) 影像采集助手：與相機連接，設定相機引數，采集影像 2) 標定助手：九點標定或是其它的標定，生成標定檔案及內參外參，可以將像素單位轉換為長度單位 ......

    uj5u.com 2023-04-20 07:42:17 more
  • 在MacOS下使用Unity3D開發游戲

    第一次發博客，先發一下我的游戲開發環境吧。 去年2月份買了一臺MacBookPro2021 M1pro(以下簡稱mbp)，這一年來一直在用mbp開發游戲。我大致分享一下我的開發工具以及使用體驗。 1、Unity 官網鏈接： https://unity.cn/releases 我一般使用的Apple ......

    uj5u.com 2023-04-20 07:40:19 more

Copyright © 2010-2020 有解無憂