引言

我們先來簡(jiǎn)單學(xué)習(xí)一下用 Go 實(shí)現(xiàn)觀察者設(shè)計(jì)模式，給怎么實(shí)現(xiàn)事件驅(qū)動(dòng)編程、事件源這些模式做個(gè)鋪墊。主要也是我也老沒看設(shè)計(jì)模式了，一起再?gòu)?fù)習(xí)一下。以前看的設(shè)計(jì)模式教程都是 Java 的，這次用 Go 實(shí)現(xiàn)一番。

觀察者模式

咱們先來看一下觀察者模式的概念，我盡量加一些自己的理解，讓它變成咱們都能理解的大俗話：

概念

觀察者模式 (Observer Pattern)，定義對(duì)象間的一種一對(duì)多依賴關(guān)系，使得每當(dāng)一個(gè)對(duì)象狀態(tài)發(fā)生改變時(shí)，其相關(guān)依賴對(duì)象皆得到通知，依賴對(duì)象在收到通知后，可自行調(diào)用自身的處理程序，實(shí)現(xiàn)想要干的事情，比如更新自己的狀態(tài)。

發(fā)布者對(duì)觀察者唯一了解的是它實(shí)現(xiàn)了某個(gè)接口（觀察者接口）。這種松散耦合的設(shè)計(jì)最大限度地減少了對(duì)象之間的相互依賴，因此使我們能夠構(gòu)建靈活的系統(tǒng)來處理主體的變化。

我的理解

上面這段話看完，相信幾乎對(duì)于理解觀察者模式能起到的作用微乎其微，類似于現(xiàn)實(shí)職場(chǎng)里加班對(duì)項(xiàng)目進(jìn)度起到的作用一樣，加班的時(shí)候誰(shuí)還沒打過幾把王者榮耀，嘿。下面我用自己的理解再給你們嘮一下。

觀察者模式也經(jīng)常被叫做發(fā)布 - 訂閱（Publish/Subscribe）模式、上面說的定義對(duì)象間的一種一對(duì)多依賴關(guān)系，一 - 指的是發(fā)布變更的主體對(duì)象，多 - 指的是訂閱變更通知的訂閱者對(duì)象。

發(fā)布的狀態(tài)變更信息會(huì)被包裝到一個(gè)對(duì)象里，這個(gè)對(duì)象被稱為事件，事件一般用英語(yǔ)過去式的語(yǔ)態(tài)來命名，比如用戶注冊(cè)時(shí)，用戶模塊在用戶創(chuàng)建好后發(fā)布一個(gè)事件 UserCreated 或者 UserWasCreated 都行，這樣從名字上就能看出，這是一個(gè)已經(jīng)發(fā)生過的事件。

事件發(fā)布給訂閱者的過程，其實(shí)就是遍歷一下已經(jīng)注冊(cè)的事件訂閱者，逐個(gè)去調(diào)用訂閱者實(shí)現(xiàn)的觀察者接口方法，比如叫 handleEvent 之類的方法，這個(gè)方法的參數(shù)一般就是當(dāng)前的事件對(duì)象。

至于很多人會(huì)好奇的，事件的處理是不是異步的？主要看我們的需求是什么，一般情況下是同步的，即發(fā)布事件后，觸發(fā)事件的方法會(huì)阻塞等到全部訂閱者返回后再繼續(xù)，當(dāng)然也可以讓訂閱者的處理異步執(zhí)行，完全看我們的需求。

大部分場(chǎng)景下其實(shí)是同步執(zhí)行的，單體架構(gòu)會(huì)在一個(gè)數(shù)據(jù)庫(kù)事務(wù)里持久化因?yàn)橹黧w狀態(tài)變更，而需要更改的所有實(shí)體類。

微服務(wù)架構(gòu)下常見的做法是有一個(gè)事件存儲(chǔ)，訂閱者接到事件通知后，會(huì)把事件先存到事件存儲(chǔ)里，這兩步也需要在一個(gè)事務(wù)里完成才能保證最終一致性，后面會(huì)再有其他線程把事件從事件存儲(chǔ)里搞到消息設(shè)施里，發(fā)給其他服務(wù)，從而在微服務(wù)架構(gòu)下實(shí)現(xiàn)各個(gè)位于不同服務(wù)的實(shí)體間的最終一致性。

所以觀察者模式，從程序效率上看，大多數(shù)情況下沒啥提升，更多的是達(dá)到一種程序結(jié)構(gòu)上的解耦，讓代碼不至于那么難維護(hù)。

Go 實(shí)現(xiàn)觀察者模式

說了這么多，我們?cè)倏聪掠?Go 怎么實(shí)現(xiàn)最簡(jiǎn)單的觀察者模式：

package main
import "fmt"
// Subject 接口，它相當(dāng)于是發(fā)布者的定義
type Subject interface {
	Subscribe(observer Observer)
	Notify(msg string)
}
// Observer 觀察者接口
type Observer interface {
	Update(msg string)
}
// Subject 實(shí)現(xiàn)
type SubjectImpl struct {
	observers []Observer
}
// Subscribe 添加觀察者（訂閱者）
func (sub *SubjectImpl) Subscribe(observer Observer) {
	sub.observers = append(sub.observers, observer)
}
// Notify 發(fā)布通知
func (sub *SubjectImpl) Notify(msg string) {
	for _, o := range sub.observers {
		o.Update(msg)
	}
}
// Observer1 Observer1
type Observer1 struct{}
// Update 實(shí)現(xiàn)觀察者接口
func (Observer1) Update(msg string) {
	fmt.Printf("Observer1: %s\n", msg)
}
// Observer2 Observer2
type Observer2 struct{}
// Update 實(shí)現(xiàn)觀察者接口
func (Observer2) Update(msg string) {
	fmt.Printf("Observer2: %s\n", msg)
}
func main(){
	sub := &SubjectImpl{}
	sub.Subscribe(&Observer1{})
	sub.Subscribe(&Observer2{})
	sub.Notify("Hello")
}

這就是 Go 實(shí)現(xiàn)觀察者模式的代碼，實(shí)際應(yīng)用的時(shí)候，一般會(huì)定義個(gè)事件總線 EventBus 或者事件分發(fā)器 Event Dispatcher，來管理事件和訂閱者間的關(guān)系和分發(fā)事件，它們兩個(gè)就是名不一樣，角色定位一樣。

下面看一下用 Go 怎么實(shí)現(xiàn)事件總線。

Go 實(shí)現(xiàn)事件總線

下面我們實(shí)現(xiàn)一個(gè)支持以下功能的事件總線

• 異步不阻塞
• 支持任意參數(shù)值

這個(gè)代碼不是我自己寫的，出處見代碼注釋首行。

代碼

// 代碼來自https://lailin.xyz/post/observer.html
package eventbus
import (
	"fmt"
	"reflect"
	"sync"
)
// Bus Bus
type Bus interface {
	Subscribe(topic string, handler interface{}) error
	Publish(topic string, args ...interface{})
}
// AsyncEventBus 異步事件總線
type AsyncEventBus struct {
	handlers map[string][]reflect.Value
	lock     sync.Mutex
}
// NewAsyncEventBus new
func NewAsyncEventBus() *AsyncEventBus {
	return &AsyncEventBus{
		handlers: map[string][]reflect.Value{},
		lock:     sync.Mutex{},
	}
}
// Subscribe 訂閱
func (bus *AsyncEventBus) Subscribe(topic string, f interface{}) error {
	bus.lock.Lock()
	defer bus.lock.Unlock()
	v := reflect.ValueOf(f)
	if v.Type().Kind() != reflect.Func {
		return fmt.Errorf("handler is not a function")
	}
	handler, ok := bus.handlers[topic]
	if !ok {
		handler = []reflect.Value{}
	}
	handler = append(handler, v)
	bus.handlers[topic] = handler
	return nil
}
// Publish 發(fā)布
// 這里異步執(zhí)行，并且不會(huì)等待返回結(jié)果
func (bus *AsyncEventBus) Publish(topic string, args ...interface{}) {
	handlers, ok := bus.handlers[topic]
	if !ok {
		fmt.Println("not found handlers in topic:", topic)
		return
	}
	params := make([]reflect.Value, len(args))
	for i, arg := range args {
		params[i] = reflect.ValueOf(arg)
	}
	for i := range handlers {
		go handlers[i].Call(params)
	}
}

單測(cè)

package eventbus
import (
	"fmt"
	"testing"
	"time"
)
func sub1(msg1, msg2 string) {
	time.Sleep(1 * time.Microsecond)
	fmt.Printf("sub1, %s %s\n", msg1, msg2)
}
func sub2(msg1, msg2 string) {
	fmt.Printf("sub2, %s %s\n", msg1, msg2)
}
func TestAsyncEventBus_Publish(t *testing.T) {
	bus := NewAsyncEventBus()
	bus.Subscribe("topic:1", sub1)
	bus.Subscribe("topic:1", sub2)
	bus.Publish("topic:1", "test1", "test2")
	bus.Publish("topic:1", "testA", "testB")
	time.Sleep(1 * time.Second)
}

毫不意外這個(gè)事件總線，只是個(gè)例子，咱也不能在項(xiàng)目開發(fā)里使用，這篇文章咱們先搞清概念，我其實(shí)前兩天關(guān)注了下，沒有發(fā)現(xiàn)什么好用的事件分發(fā)、事件總線的三方庫(kù)，好在實(shí)現(xiàn)起來也不難，后面我準(zhǔn)備自己寫一個(gè)能用的到時(shí)候分享給大家，最起碼是在學(xué)習(xí)、練習(xí)項(xiàng)目里能使用的吧。

總結(jié)

今天給大家用大白話瞎嘮了一下觀察者模式的原理和實(shí)際怎么應(yīng)用，感覺文章的精髓主要在前半部分，可能有的不你還不能理解，后面我會(huì)再通過后續(xù)文章逐一解釋，其實(shí)這些都是事件驅(qū)動(dòng)和事件源這些模式里的基礎(chǔ)內(nèi)容。

至于這次給出的代碼，其實(shí)沒啥實(shí)戰(zhàn)意義，就是大家先了解一下。Go 里邊關(guān)于事件驅(qū)動(dòng)之類的內(nèi)容，感覺不多，有 Spring 使用經(jīng)驗(yàn)的可以先看看 Spring 提供的@EventListener 注解，需要訂閱者異步執(zhí)行可以配合 @Async 注解使用，至于我上面說的需要保證事件發(fā)布的主體和訂閱者的原子性持久化的話，則是通過@Transitional 和 @TransactionalEventListener 結(jié)合使用來實(shí)現(xiàn)。

以上就是Go語(yǔ)言設(shè)計(jì)模式之實(shí)現(xiàn)觀察者模式解決代碼臃腫的詳細(xì)內(nèi)容，更多關(guān)于Go 觀察者模式的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論