上一篇：【Go實(shí)現(xiàn)】實(shí)踐GoF的23種設(shè)計(jì)模式：裝飾者模式

簡單的分布式應(yīng)用系統(tǒng)（示例代碼工程）：https://github.com/ruanrunxue/Practice-Design-Pattern--Go-Implementation

簡介

現(xiàn)在有 2 個(gè)服務(wù)，Service A 和 Service B，通過 REST 接口通信；Service A 在某個(gè)業(yè)務(wù)場景下調(diào)用 Service B 的接口完成一個(gè)計(jì)算密集型任務(wù)，假設(shè)接口為 http://service_b/api/v1/domain；該任務(wù)運(yùn)行時(shí)間很長，但 Service A 不想一直阻塞在接口調(diào)用上。為了滿足 Service A 的要求，通常有 2 種方案：

1. Service A 隔一段時(shí)間調(diào)用一次 Service B 的接口，如果任務(wù)還沒完成，就返回 HTTP Status 102 Processing；如果已完成，則返回 HTTP Status 200 Ok。

   

2. Service A 在請求 Service B 接口時(shí)帶上 callback uri，比如 http://service_b/api/v1/domain?callbackuri=http://service_a/api/v1/domain，Service B 收到請求后立即返回 HTTP Status 200 Ok，等任務(wù)完成后再調(diào)用 Service A callback uri 進(jìn)行通知。

   

方案 1 須要輪詢接口，輪詢太頻繁會導(dǎo)致資源浪費(fèi)，間隔太長又會導(dǎo)致任務(wù)完成后 Service A 無法及時(shí)感知。顯然，方案 2 更加高效，因此也被廣泛應(yīng)用。

方案 2 用到的思想就是本文要介紹的觀察者模式（Observer Pattern），GoF 對它的定義如下：

Define a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically.

我們將觀察者稱為Observer，被觀察者（或主體）稱為Subject，那么Subject 和 Observer 是一對多的關(guān)系，當(dāng) Subject 狀態(tài)變更時(shí)，所有的 Observer 都會被通知到。

UML 結(jié)構(gòu)

場景上下文

在簡單的分布式應(yīng)用系統(tǒng)（示例代碼工程）中，應(yīng)用之間通過 network 模塊來通信，其中通信模型采用觀察者模式：

從上圖可知，App 直接依賴 http 模塊，而 http 模塊底層則依賴 socket 模塊：

1. 在App2初始化時(shí)，先向 http 模塊注冊一個(gè)request handler，處理App1發(fā)送的 http 請求。
2. http 模塊會將request handler轉(zhuǎn)換為packet handler注冊到 socket 模塊上。
3. App 1發(fā)送 http 請求，http 模塊將請求轉(zhuǎn)換為socket packet發(fā)往App 2的 socket 模塊。
4. App 2的 socket 模塊收到 packet 后，調(diào)用packet handler處理該報(bào)文；packet handler又會調(diào)用App 2注冊的request handler處理該請求。

在上述socket - http - app 三層模型中，對 socket 和 http，socket 是 Subject，http 是 Observer；對 http 和 app，http 是 Subject，app 是 Observer。

代碼實(shí)現(xiàn)

因?yàn)樵谟^察者模式的實(shí)現(xiàn)上，socket 模塊和 http 模塊類似，所以，下面只給出 socket 模塊的實(shí)現(xiàn)：

//demo/network/socket.go
packagenetwork

//關(guān)鍵點(diǎn)1:定義Observer接口
//SocketListenerSocket報(bào)文監(jiān)聽者
typeSocketListenerinterface{
//關(guān)鍵2:為Observer定義更新處理方法，入?yún)橄嚓P(guān)的上下文對象
Handle(packet*Packet)error
}

//Subject接口
//Socket網(wǎng)絡(luò)通信Socket接口
typeSocketinterface{
//Listen在endpoint指向地址上起監(jiān)聽
Listen(endpointEndpoint)error
//Close關(guān)閉監(jiān)聽
Close(endpointEndpoint)
//Send發(fā)送網(wǎng)絡(luò)報(bào)文
Send(packet*Packet)error
//Receive接收網(wǎng)絡(luò)報(bào)文
Receive(packet*Packet)
//AddListener增加網(wǎng)絡(luò)報(bào)文監(jiān)聽者
AddListener(listenerSocketListener)
}

//關(guān)鍵點(diǎn)3:定義Subject對象
//socketImplSocket的默認(rèn)實(shí)現(xiàn)
typesocketImplstruct{
//關(guān)鍵點(diǎn)4:在Subject中持有Observer的集合
listeners[]SocketListener
}

//關(guān)鍵點(diǎn)5:為Subject定義注冊O(shè)bserver的方法
func(s*socketImpl)AddListener(listenerSocketListener){
s.listeners=append(s.listeners,listener)
}

//關(guān)鍵點(diǎn)6:當(dāng)Subject狀態(tài)變更時(shí)，遍歷Observers集合，調(diào)用它們的更新處理方法
func(s*socketImpl)Receive(packet*Packet){
for_,listener:=ranges.listeners{
listener.Handle(packet)
}
}

...

總結(jié)實(shí)現(xiàn)觀察者模式的幾個(gè)關(guān)鍵點(diǎn)：

1. 定義 Observer 接口，上述例子中為SocketListener接口。
2. 為 Observer 接口定義狀態(tài)更新的處理方法，其中方法入?yún)橄嚓P(guān)的上下文對象。上述例子為Handle方法，上下文對象為Packet。
3. 定義 Subject 對象，上述例子為socketImpl對象。當(dāng)然，也可以先將 Subject 抽象為接口，比如上述例子中的Socket接口，但大多數(shù)情況下都不是必須的。
4. 在 Subject 對象中，持有 Observer 接口的集合，上述例子為listeners屬性。讓 Subject 依賴 Observer 接口，能夠使 Subject 與具體的 Observer 實(shí)現(xiàn)解耦，提升代碼的可擴(kuò)展性。
5. 為 Subject 對象定義注冊 Observer 的方法，上述例子為AddListener方法。
6. 當(dāng) Subject 狀態(tài)變更時(shí)，遍歷 Observer 集合，并調(diào)用它們的狀態(tài)更變處理方法，上述例子為Receive方法。

擴(kuò)展

發(fā)布-訂閱模式

與觀察者模式相近的，是發(fā)布-訂閱模式（Pub-Sub Pattern），很多人會把兩者等同，但它們之間還是有些差異。

從前文的觀察者模式實(shí)現(xiàn)中，我們發(fā)現(xiàn) Subject 持有 Observer 的引用，當(dāng)狀態(tài)變更時(shí)，Subject 直接調(diào)用 Observer 的更新處理方法完成通知。也就是，Subject 知道有哪些 Observer，也知道 Observer 的數(shù)量：

在發(fā)布-訂閱模式中，我們將發(fā)布方稱為Publisher，訂閱方稱為Subscriber，不同于觀察者模式，Publisher 并不直接持有 Subscriber 引用，它們之間通常通過Broker來完成解耦。也即，Publisher 不知道有哪些 Subscriber，也不知道 Subscriber 的數(shù)量：

發(fā)布-訂閱模式被廣泛應(yīng)用在消息中間件的實(shí)現(xiàn)上，比如 Apache Kafka 基于 Topic 實(shí)現(xiàn)了發(fā)布-訂閱模式，發(fā)布方稱為 Producer，訂閱方稱為 Consumer。

下面，我們通過簡單的分布式應(yīng)用系統(tǒng)（示例代碼工程）中的 mq 模塊，展示一個(gè)簡單的發(fā)布-訂閱模式實(shí)現(xiàn)，在該實(shí)現(xiàn)中，我們將 Publisher 的 produce 方法和 Subscriber 的 consume 方法都合并到 Broker 中：

//demo/mq/memory_mq.go

//關(guān)鍵點(diǎn)1:定義通信雙方交互的消息，攜帶topic信息
//Message消息隊(duì)列中消息定義
typeMessagestruct{
topicTopic
payloadstring
}

//關(guān)鍵點(diǎn)2:定義Broker對象
//memoryMq內(nèi)存消息隊(duì)列，通過channel實(shí)現(xiàn)
typememoryMqstruct{
//關(guān)鍵點(diǎn)3: Broker中維持一個(gè)隊(duì)列的map，其中key為topic，value為queue，go語言通常用chan實(shí)現(xiàn)。
queuessync.Map//key為Topic，value為chan*Message，每個(gè)topic單獨(dú)一個(gè)隊(duì)列
}

//關(guān)鍵點(diǎn)4:為Broker定義Produce方法，根據(jù)消息中的topic選擇對應(yīng)的queue發(fā)布消息
func(m*memoryMq)Produce(message*Message)error{
record,ok:=m.queues.Load(message.Topic())
if!ok{
q:=make(chan*Message,10000)
m.queues.Store(message.Topic(),q)
record=q
}
queue,ok:=record.(chan*Message)
if!ok{
returnerrors.New("model'stypeisnotchan*Message")
}
queue<-?message
?returnnil
}

//關(guān)鍵點(diǎn)5:為Broker定義Consume方法，根據(jù)topic選擇對應(yīng)的queue消費(fèi)消息
func(m*memoryMq)Consume(topicTopic)(*Message,error){
record,ok:=m.queues.Load(topic)
if!ok{
q:=make(chan*Message,10000)
m.queues.Store(topic,q)
record=q
}
queue,ok:=record.(chan*Message)
if!ok{
returnnil,errors.New("model'stypeisnotchan*Message")
}
return<-queue,?nil
}

客戶端使用時(shí)，直接調(diào)用memoryMq的Produce方法和Consume方法完成消息的生產(chǎn)和消費(fèi)：

//發(fā)布方
funcpublisher(){
msg:=NewMessage("test","helloworld")
err:=MemoryMqInstance().Produce(msg)
assert.Nil(t,err)
}

//訂閱方
funcsubscriber(){
result,err:=MemoryMqInstance().Consume("test")
assert.Nil(err)
assert.Equal(t,"helloworld",result.payload)
}

總結(jié)實(shí)現(xiàn)發(fā)布-訂閱模式的幾個(gè)關(guān)鍵點(diǎn)：

1. 定義通信雙方交互的消息，攜帶 topic 信息，上述例子為Message對象。
2. 定義 Broker 對象，Broker 是緩存消息的地方，上述例子為memoryMq對象。
3. 在 Broker 中維持一個(gè)隊(duì)列的 map，其中 key 為 topic，value 為 queue，go 語言通常用 chan 來實(shí)現(xiàn) queue，上述例子為queues屬性。
4. 為 Broker 定義 produce 方法，根據(jù)消息中的 topic 選擇對應(yīng)的 queue 發(fā)布消息，上述例子為Produce方法。
5. 為 Broker 定義 consume 方法，根據(jù) topic 選擇對應(yīng)的 queue 消費(fèi)消息，上述例子為Consume方法。

Push 模式 VS Pull 模式

實(shí)現(xiàn)觀察者模式和發(fā)布-訂閱模式時(shí)，都會涉及到Push 模式或Pull 模式的選取。所謂 Push 模式，指的是 Subject/Publisher 直接將消息推送給 Observer/Subscriber；所謂 Pull 模式，指的是 Observer/Subscriber 主動向 Subject/Publisher 拉取消息：

Push 模式和 Pull 模式的選擇，取決于通信雙方處理消息的速率大小。

如果 Subject/Publisher 方生產(chǎn)消息的速率要比 Observer/Subscriber 方處理消息的速率小，可以選擇 Push 模式，以求得更高效、及時(shí)的消息傳遞；相反，如果 Subject/Publisher 方產(chǎn)生消息的速率要大，就要選擇 Pull 模式，由 Observer/Subscriber 方?jīng)Q定消息的消費(fèi)速率，否則可能導(dǎo)致 Observer/Subscriber 崩潰。

Pull 模式有個(gè)缺點(diǎn)，如果當(dāng)前無消息可處理，將導(dǎo)致 Observer/Subscriber 空輪詢，可以采用類似 Kafka 的解決方案：讓 Observer/Subscriber 阻塞一定時(shí)長，讓出 CPU，避免長期無效的 CPU 空轉(zhuǎn)。

典型應(yīng)用場景

• 需要監(jiān)聽某個(gè)狀態(tài)的變更，且在狀態(tài)變更時(shí)，通知到監(jiān)聽者。
• web 框架。很多 web 框架都用了觀察者模式，用戶注冊請求 handler 到框架，框架收到相應(yīng)請求后，調(diào)用 handler 完成處理邏輯。
• 消息中間件。如 Kafka、RocketMQ 等。

優(yōu)缺點(diǎn)

優(yōu)點(diǎn)

• 消息通信雙方解耦。觀察者模式通過依賴接口達(dá)到松耦合；發(fā)布-訂閱模式則通過 Broker 達(dá)到解耦目的。

• 支持廣播通信。

• 可基于 topic 來達(dá)到指定消費(fèi)某一類型消息的目的。

缺點(diǎn)

• 通知 Observer/Subscriber 的順序是不確定的，應(yīng)用程序不應(yīng)該依賴通知順序來保證業(yè)務(wù)邏輯的正確性。
• 廣播通信場景，需要 Observer/Subscriber 自己去判斷是否需要處理該消息，否則容易導(dǎo)致unexpected update。

與其他模式的關(guān)聯(lián)

觀察者模式和發(fā)布-訂閱模式中的 Subject 和 Broker，通常都會使用單例模式來確保它們?nèi)治ㄒ弧?/p>

文章配圖

可以在用Keynote畫出手繪風(fēng)格的配圖中找到文章的繪圖方法。

審核編輯：湯梓紅