go-micro 集成 RabbitMQ 實戰和原理

在 go-micro 中異步消息的收發是通過 Broker 這個組件來完成的，底層實現有 RabbitMQ、Kafka、Redis 等等很多種方式，這篇文章主要介紹 go-micro 使用 RabbitMQ 收發數據的方法和原理。

Broker 的核心功能

Broker 的核心功能是 Publish 和 Subscribe，也就是發佈和訂閱。它們的定義是：

Publish(topic string, m *Message, opts ...PublishOption) error
Subscribe(topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)

發佈

發佈第一個參數是 topic（主題），用於標識某類消息。

發佈的數據是通過 Message 承載的，其包括消息頭和消息體，定義如下：

type Message struct {
    Header map[string]string
    Body   []byte
}

消息頭是 map，也就是一組 KV（鍵值對）。

消息體是字節數組，在發送和接收時需要開發者進行編碼和解碼的處理。

訂閱

訂閱的第一個參數也是 topic（主題），用於過濾出要接收的消息。

訂閱的數據是通過 Handler 處理的，Handler 是一個函數，其定義如下：

type Handler func(Event) error

其中的參數 Event 是一個接口，需要具體的 Broker 來實現，其定義如下：

type Event interface {
    Topic() string
    Message() *Message
    Ack() error
    Error() error
}

開發者訂閱數據時，需要實現 Handler 這個函數，接收 Event 的實例，提取數據進行處理，根據不同的 Broker，可能還需要調用 Ack()，處理出現錯誤時，返回 error。

go-micro 集成 RabbitMQ 實戰

大概瞭解了 Broker 的定義之後，再來看下如何使用 go-micro 收發 RabbitMQ 消息。

啓動一個 RabbitMQ

如果你已經有一個 RabbitMQ 服務器，請跳過這個步驟。

這裏介紹一個使用 docker 快速啓動 RabbitMQ 的方法，當然前提是你得安裝了 docker。

執行如下命令啓動一個 rabbitmq 的 docker 容器：

docker run --name rabbitmq1 -p 5672:5672 -p 15672:15672 -d rabbitmq

然後進入容器進行一些設置：

docker exec -it rabbitmq1 /bin/bash

啓動管理工具、禁用指標採集（會導致某些 API500 錯誤）：

rabbitmq-plugins enable rabbitmq_management

cd /etc/rabbitmq/conf.d/
echo management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf

最後重啓容器：

docker restart rabbitmq1

最後瀏覽器中輸入 http://127.0.0.0:15672 即可訪問，默認用戶名和密碼都是 guest 。

編寫收發函數

爲了方便演示，先來定義發佈消息和接收消息的函數。其中發佈函數使用了 go-micro 提供的 Event 類型，還有其它類型也可以提供 Publish 的功能，這裏發送的數據格式是 Json 字符串。接收消息的函數名稱可以隨意取，但是參數和返回值必須符合規範，也就是下邊代碼中的樣子，這個函數也可以是綁定到某個類型的。

// 定義一個發佈消息的函數：每隔1秒發佈一條消息
func loopPublish(event micro.Event) {
    for {
        time.Sleep(time.Duration(1) * time.Second)

        curUnix := strconv.FormatInt(time.Now().Unix(), 10)
        msg := "{\"Id\":" + curUnix + ",\"Name\":\"張三\"}"
        event.Publish(context.TODO(), msg)
    }
}

// 定義一個接收消息的函數：將收到的消息打印出來
func handle(ctx context.Context, msg interface{}) (err error) {
    defer func() {
        if r := recover(); r != nil {
            err = errors.New(fmt.Sprint(r))
            log.Println(err)
        }
    }()

    b, err := json.Marshal(msg)
    if err != nil {
        log.Println(err)
        return
    }

    log.Println(string(b))
    return
}

編寫主體代碼

這裏先給出代碼，裏面提供了一些註釋，後邊還會有詳細介紹。

func main() {
    // RabbitMQ的連接參數
    rabbitmqUrl := "amqp://guest:guest@127.0.0.1:5672/"
    exchangeName := "amq.topic"
    subcribeTopic := "test"
    queueName := "rabbitmqdemo_test"

    // 默認是application/protobuf，這裏演示用的是Json，所以要改下
    server.DefaultContentType = "application/json"

    // 創建 RabbitMQ Broker
    b := rabbitmq.NewBroker(
        broker.Addrs(rabbitmqUrl),           // RabbitMQ訪問地址，含VHost
        rabbitmq.ExchangeName(exchangeName), // 交換機的名稱
        rabbitmq.DurableExchange(),          // 消息在Exchange中時會進行持久化處理
        rabbitmq.PrefetchCount(1),           // 同時消費的最大消息數量
    )

    // 創建Service，內部會初始化一些東西，必須在NewSubscribeOptions前邊
    service := micro.NewService(
        micro.Broker(b),
    )
    service.Init()

    // 初始化訂閱上下文：這裏不是必需的，訂閱會有默認值
    subOpts := broker.NewSubscribeOptions(
        rabbitmq.DurableQueue(),   // 隊列持久化，消費者斷開連接後，消息仍然保存到隊列中
        rabbitmq.RequeueOnError(), // 消息處理函數返回error時，消息再次入隊列
        rabbitmq.AckOnSuccess(),   // 消息處理函數沒有error返回時，go-micro發送Ack給RabbitMQ
    )

    // 註冊訂閱
    micro.RegisterSubscriber(
        subcribeTopic,    // 訂閱的Topic
        service.Server(), // 註冊到的rpcServer
        handle,           // 消息處理函數
        server.SubscriberContext(subOpts.Context), // 訂閱上下文，也可以使用默認的
        server.SubscriberQueue(queueName),         // 隊列名稱
    )

    // 發佈事件消息
    event := micro.NewEvent(subcribeTopic, service.Client())
    go loopPublish(event)

    log.Println("Service is running ...")
    if err := service.Run(); err != nil {
        log.Println(err)
    }
}

主要邏輯是：

1、先創建一個 RabbitMQ Broker，它實現了標準的 Broker 接口。其中主要的參數是 RabbitMQ 的訪問地址和 RabbitMQ 交換機，PrefetchCount 是訂閱者（或稱爲消費者）使用的。

2、然後通過 NewService 創建 go-micro 服務，並將 broker 設置進去。這裏邊會初始化很多東西，最核心的是創建一個 rpcServer，並將 rpcServer 和這個 broker 綁定起來。

3、然後是通過 RegisterSubscriber 註冊訂閱，這個註冊有兩個層面的功能：一是如果 RabbitMQ 上還不存在這個隊列時創建隊列，並訂閱指定 topic 的消息；二是定義 go-micro 程序從這個 RabbitMQ 隊列接收數據的處理方式。

這裏詳細看下訂閱的參數：

func RegisterSubscriber(topic string, s server.Server, h interface{}, opts ...server.SubscriberOption) error

4、然後這裏爲了演示，通過 NewEvent 創建了一個 Event，通過它每隔一秒發送 1 條消息。

5、最後通過 service.Run() 把這個程序啓動起來。

辛苦寫了半天，看一下這個程序的運行效果：

注意一般發佈者和訂閱者是在不同的程序中，這裏只是爲了方便演示，才把他們放在一個程序中。所以如果只是發佈消息，就不需要訂閱的代碼，如果只是訂閱，也不需要發佈消息的代碼，大家使用的時候根據需要自己裁剪吧。

go-micro 集成 RabbitMQ 的處理流程

這個部分來看一下消息在 go-micro 和 RabbitMQ 中是怎麼流轉的，我畫了一個示意圖：

這個圖有點複雜，這裏詳細講解下。

首先分成三塊：RabbitMQ、消息發佈部分、消息接收部分，這裏用不同的顏色進行了區分。

這個處理過程還可以劃分爲業務部分、核心模塊部分和插件部分。

從上邊的圖中可以看到消息都需要經過這個 RabbitMQ 插件進行處理，實際上可以只使用這個插件，就能實現消息的發送和接收。這個演示代碼我已經提交到了 Github，有興趣的同學可以在文末獲取 Github 倉庫的地址。

從上邊這些劃分中，我們可以理解到設計者的整體設計思路，把握關鍵節點，用好用對，出現問題時可以快速定位。

填的幾個坑

不能接收其它框架發佈的消息

這個是因爲 route.ProcessMessage 查找訂閱時使用了 go-micro 專用的一個頭信息：

// get the subscribers by topic
    subs, ok := router.subscribers[msg.Topic()]

這個 msg.Topic 返回的是如下實例中的 topic 字段：

    rpcMsg := &rpcMessage{
        topic:       msg.Header["Micro-Topic"],
        contentType: ct,
        payload:     &raw.Frame{Data: msg.Body},
        codec:       cf,
        header:      msg.Header,
        body:        msg.Body,
    }

其它框架不會有這麼一個頭信息，除非專門適配 go-micro。

因爲使用 RabbitMQ 的場景下，整個開發都是圍繞 RabbitMQ 做的，而且 go-micro 的處理邏輯沒有考慮 RabbitMQ 訂閱可以使用通配符的情況，發佈消息的 Topic、接收消息的 Topic 與 Micro-Topic 的值匹配時都是按照是否相等的原則處理的，因此可以用 RabbitMQ 消息自帶的 topic 來設置這個消息頭。rabbitmq.rbroker.Subscribe 中接收到消息後，就可以進行這個設置：

// Messages sent from other frameworks to rabbitmq do not have this header.
        // The 'RoutingKey' in the message can be used as this header.
        // Then the message can be transfered to the subscriber which bind this topic.
        msgTopic := header["Micro-Topic"]
        if msgTopic == "" {
            header["Micro-Topic"] = msg.RoutingKey
        }

這樣 go-micro 開發的消費者程序就能接收其它框架發佈的消息了，其它框架無需適配。

RabbitMQ 重啓後訂閱者和發佈者無限阻塞

go-micro 的 RabbitMQ 插件底層使用另一個庫：github.com/streadway/amqp

對於發佈者，RabbitMQ 斷開連接時 amqp 庫會通過 Go Channel 同步通知 go-micro，然後 go-micro 可以發起重新連接。問題出現在這個同步通知上，go-micro 的 RabbitMQ 插件設置了接收連接和通道的關閉通知，但是隻處理了一個通知就去重新連接了，這就導致有一個 Go Channel 一直阻塞，而這個阻塞會導致某個鎖不能釋放，這個鎖又是 Publish 時候需要的，因此導致發佈者無限阻塞。解決辦法就是外層增加一個循環，等所有的通知都收到了，再去做重新連接。

對於訂閱者，RabbitMQ 斷開連接時，它會一直阻塞在某個 Go Channel 上，直到它返回一個值，這個值代表連接已經重新建立，訂閱者可以重建消費通道。問題也是出現在這個阻塞的 Go Channel 上，因爲這個 Go Channel 在每次收到 amqp 的關閉通知時會重新賦值，而訂閱者等待的 Go Channel 可能是之前的舊值，永遠也不會返回，訂閱者也就無限阻塞了。解決辦法呢，就是在 select 時增加一個 time.After，讓等待的 Go Channel 有機會更新到新值。

代碼就不貼了，有興趣的可以到 Github 中去看：https://github.com/go-micro/plugins/commit/9f64710807221f3cc649ba4fe05f75b07c66c00c

關於這兩個問題的修改已經合併到官方倉庫中，大家去 get 最新的代碼就可以了。

這兩個坑填了，基本上就能滿足我的需要了。當然可能還有其它的坑，比如 go-micro 的 RabbitMQ 插件好像沒有發佈者確認的功能，這個要實現，還得好好想想怎麼改。

好了，以上就是本文的主要內容。

老規矩，代碼已經上傳到 Github，歡迎訪問：https://github.com/bosima/go-demo/tree/main/go-micro-broker-rabbitmq

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/oO5K9x10P8HWmL8ADW7-KA

猜你喜歡