rabbitMq 交換機介紹 系列五 Header [golang 版本]

4種不一樣的交換機類型

直連交換機：Direct exchange
扇形交換機：Fanout exchange
主題交換機：Topic exchange
首部交換機：Headers exchange

交換機具體含義參考 https://www.jianshu.com/p/469...

Header 交換機

如下代碼參數具體含義能夠參考 https://segmentfault.com/a/11...

Demo

生產者示例 producer.go

package main

import (
    "fmt"
    "github.com/streadway/amqp"
    "time"
)

//因：快速實現邏輯，故：不處理錯誤邏輯
func main() {
    conn, _ := amqp.Dial("amqp://user:password@host:ip/vhost")
    ch, _ := conn.Channel()
    body := "Hello World! " + time.Now().Format("2006-01-02 15:04:05")
    fmt.Println(body)
    var exchange_name string = "j_exch_head"
    var routing_key string = "jkey"
    var etype string = amqp.ExchangeHeaders
    //聲明交換器
    ch.ExchangeDeclare(exchange_name, etype, true, false, false, false, nil)

    ch.Publish(
        exchange_name, // exchange 這裏爲空則不選擇 exchange
        routing_key,   // routing key
        false,         // mandatory
        false,         // immediate
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(body),
            Headers:     amqp.Table{"x-match": "any", "mail": "470047253@qq.com", "author": "Jhonny"}, // 頭部信息 any:匹配一個便可 all:所有匹配
            //Expiration:  "3000", // 設置過時時間
        })

    // defer 關鍵字
    defer conn.Close() // 壓棧 後進先出
    defer ch.Close()   // 壓棧 後進先出

}

$ go run producer.go
Hello World! 2021-03-18 16:08:13

消費者示例 consumer.go

package main

import (
    "github.com/streadway/amqp"
    "log"
)

func main() {
    conn, _ := amqp.Dial("amqp://user:password@host:ip/vhost")
    ch, _ := conn.Channel()
    var exchange_name string = "j_exch_head"
    var routing_key string = "jkey"
    var queue_name string = "j_queue"
    var etype string = amqp.ExchangeHeaders // 頭部交換機

    ch.QueueDeclare(queue_name, true, false, true, false, nil)
    //聲明交換器
    ch.ExchangeDeclare(exchange_name, etype, true, false, false, false, nil)


    ch.QueueBind(
        queue_name,    // queue name
        routing_key,   // routing key: Headers 頭部交換機跟routing_key 不要緊
        exchange_name, // exchange
        false,
        amqp.Table{"mail": "470047253@qq.com"}, //  頭部信息 any:匹配一個便可 all:所有匹配
    )

    //監聽
    msgs, _ := ch.Consume(
        queue_name, // queue name
        "",         // consumer
        true,       // auto-ack
        false,      // exclusive
        false,      // no-local
        false,      // no-wait
        nil,        // args
    )

    forever := make(chan bool)

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

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

}

$ go run consumer.go
2021/03/18 15:52:30  [*] Waiting for logs. To exit press CTRL+C
2021/03/18 15:52:34  [x] Hello World! 2021-03-18 15:52:34
2021/03/18 15:53:14  [x] Hello World! 2021-03-18 15:53:14
2021/03/18 16:08:13  [x] Hello World! 2021-03-18 16:08:13