前言

在當今微服務和分布式系統盛行的背景下，事件驅動架構（Event-Driven Architecture，EDA）扮演著一個至關重要的角色，此架構的設計使得服務間可以通過事件進行同步或異步通信，替代了傳統的直接接口調用。基于事件的交互方式，促進了服務之間的松耦合，提高系統的可擴展性。

發布-訂閱模式是實現事件驅動架構的模式之一，它允許系統的不同組件或服務發布事件，而其他組件或服務可以訂閱這些事件并根據事件內容進行響應。相信大部分開發者都接觸過這一模式，常見的技術實現有消息隊列（MQ）和 Redis 發布/訂閱（PUB/SUB）功能等。

在 Go 語言中，我們可以利用其強大的 channel 和并發機制來實現發布-訂閱模式。本文將深入探討如何在 Go 中實現一個簡單的事件總線，這是發布-訂閱模式的具體實現。

準備好了嗎？準備一杯你最喜歡的咖啡或茶，隨著本文一探究竟吧。

事件總線

事件總線是發布-訂閱模式的具體實現，它作為發布者和訂閱者的中間件，管理著事件傳遞與分發，確保事件從發布者順利地傳達到訂閱者。

圖片

事件總線的優勢主要包括：

• 解耦：服務間不需要直接通信，而是通過時間進行交互，減少服務間的依賴。
• 異步處理：事件可以被異步處理，提高系統的響應性和性能。
• 可擴展性：新的訂閱者可以輕松訂閱事件，不需要修改現有的發布者代碼。
• 錯誤隔離：事件處理的失敗不會直接影響其他服務的正常運行。

事件總線的代碼實現

接下來將介紹如何在 Go 語言中實現一個簡單的事件總線，它包含以下關鍵功能：

• 發布：允許系統的各個服務發送事件。
• 訂閱：允許感興趣的服務訂閱接收特定類型的事件。
• 取消訂閱：允許各個服務將本身已訂閱的事件刪除。

項目源碼地址：https://github.com/chenmingyong0423/go-eventbus

事件數據結構定義

type Event struct {
    Payload any
}

Event 是一個封裝事件的結構體，其中 Payload 為事件的上下文信息，類型是 any。

事件總線定義

type (
    EventChan chan Event
)

type EventBus struct {
    mu    sync.RWMutex
    subscribers map[string][]EventChan
}

func NewEventBus() *EventBus {
    return &EventBus{
        subscribers: make(map[string][]EventChan),
    }
}

EventChan 是一個類型別名，定義為傳遞 Event 結構體的通道 chan Event。

EventBus 為事件總線的定義，它包含兩個屬性：

• mu：一個讀寫互斥鎖（sync.RWMutex），用于保證下面 subscribers 的并發讀寫安全。
• subscribers：一個映射，鍵為字符串類型，表示訂閱的主題；值為 EventChan 切片類型。該屬性用于存儲各個主體的所有訂閱者，每個訂閱者通過 EventChan 接收事件。

NewEventBus 函數用于創建一個新的 EventBus 事件總線。

事件總線的方法實現

事件總線實現了三個方法，分別為發布事件（Publish）和訂閱事件（Subscribe）以及取消訂閱事件（Unsubscribe）。

Publish 發布事件

func (eb *EventBus) Publish(topic string, event Event) {
	eb.mu.RLock()
	defer eb.mu.RUnlock()
	// 復制一個新的訂閱者列表，避免在發布事件時修改訂閱者列表
	subscribers := append([]EventChan{}, eb.subscribers[topic]...)
	gofunc() {
		for _, subscriber := range subscribers {
			subscriber <- event
		}
	}()
}

Publish 方法用于發布事件。該方法接收兩個參數：topic（主題）和 event （封裝事件的對象）。

在 Publish 方法的實現中，首先通過 mu 屬性獲取讀鎖，以確保接下來的 subscribers 寫操作是協程安全的。然后復制一份當前主題的訂閱者列表 subscribers。接下來開啟一個新 goroutine，在這個 goroutine 中遍歷復制的訂閱者列表，將事件通過通道發送給所有訂閱者。完成這些操作后，釋放讀鎖。

為什么會復制一個新的訂閱者列表？

答：復制訂閱者列表是為了在發送事件時保持數據的一致性和穩定性。由于向通道發送數據的操作是在一個新的 goroutine 中進行的，在發送數據時，讀鎖已經被釋放，原來的訂閱者列表可能會由于添加或刪除訂閱者而發生變化。如果直接使用原來的訂閱者列表，可能會發生預料之外的錯誤（如向一個已經關閉的通道發送數據會產生 panic）。

Subscribe 訂閱事件

func (eb *EventBus) Subscribe(topic string) EventChan {
	eb.mu.Lock()
	defer eb.mu.Unlock()
	ch := make(EventChan)
	eb.subscribers[topic] = append(eb.subscribers[topic], ch)
	return ch
}

Subscribe 方法用于訂閱特定主題的事件。該方法有接收一個 topic 參數，表示希望訂閱的主題。通過此方法，可以獲得一個 EventChan 通道，用于接收該主題的事件。

在 Subscribe 方法的實現中，首先通過 mu 屬性獲取寫鎖，以保證接下來的 subscribers 讀寫操作是協程安全的；接著創建一個新的 EventChan 通道 ch，將其添加到相應主題的訂閱者切片中。完成這些操作后，釋放寫鎖。

Unsubscribe 取消訂閱事件

func (eb *EventBus) Unsubscribe(topic string, ch EventChan) {
	eb.mu.Lock()
	defer eb.mu.Unlock()
	if subscribers, ok := eb.subscribers[topic]; ok {
		for i, subscriber := range subscribers {
			if ch == subscriber {
				eb.subscribers[topic] = append(subscribers[:i], subscribers[i+1:]...)
				close(ch)
				// 清空通道
				forrange ch {
				}
				return
			}
		}
	}
}

Unsubscribe 方法用于取消訂閱事件。該方法接收兩個參數：topic（已訂閱的主題）和 ch（被頒發的通道）。

在 Unsubscribe 方法里，首先通過 mu 屬性獲取寫鎖，以保證接下來的 subscribers 讀寫操作是協程安全的；然后檢查 topic 主題是否存在對應的訂閱者。如果存在，遍歷該主題的訂閱者切片，找到與 ch 相匹配的通道，將其從訂閱者切片里移除并關閉該通道。然后清空通道。完成這些操作后，釋放寫鎖。

使用示例

// https://github.com/chenmingyong0423/blog/blob/master/tutorial-code/go/eventbus/main.go
package main

import (
	"fmt"
	"time"

	"github.com/chenmingyong0423/go-eventbus"
)

func main() {
	eventBus := eventbus.NewEventBus()

	// 訂閱 post 主題事件
	subscribe := eventBus.Subscribe("post")

	gofunc() {
		for event := range subscribe {
			fmt.Println(event.Payload)

		}
	}()

	eventBus.Publish("post", eventbus.Event{Payload: map[string]any{
		"postId": 1,
		"title":  "Go 事件驅動編程：實現一個簡單的事件總線",
		"author": "陳明勇",
	}})
	// 不存在訂閱者的 topic
	eventBus.Publish("pay", eventbus.Event{Payload: "pay"})

	time.Sleep(time.Second * 2)
	// 取消訂閱 post 主題事件
	eventBus.Unsubscribe("post", subscribe)
}

擴展建議

本文實現的事件總線較為簡單，如果要增強時間總線的靈活性，可靠性和易用性等方面，我們可以考慮擴展它，以下是一些建議：

• 事件持久化：實現時間的持久化存儲功能，確保系統崩潰后可以恢復未處理的事件。
• 通配符和模式匹配訂閱：允許使用通配符或正則表達式來訂閱一組相關主題，而不是單個具體的主題。
• 負載均衡和消息分發策略：在多個訂閱者之間分配事件，實現負載均衡。
• 插件支持：支持通過插件來擴展功能，如日志記錄、消息過濾、轉換等。

小結

本文深入探討了在 Go 語言中實現簡單事件總線的過程。通過利用 Go 語言的強大特性，如 channel 和并發機制，我們可以輕松地實現發布-訂閱模式。

文章從事件總線的優勢開始，介紹了其解耦、異步處理、可擴展性和錯誤隔離等特點。然后詳細解釋了如何定義事件數據結構和事件總線結構，并實現了發布、訂閱和取消訂閱事件的方法。最后，提出了一些可能的擴展方向，如事件持久化、通配符訂閱、負載均衡和插件支持，以增強事件總線的靈活性和功能性。

通過閱讀本文，你可以學會在 Go 語言中實現一個簡單但功能強大的事件總線，并根據可能的需求進行擴展。

★項目源碼地址：https://github.com/chenmingyong0423/go-eventbus