GoF設(shè)計(jì)模式之觀察者模式

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

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

簡(jiǎn)介

現(xiàn)在有 2 個(gè)服務(wù)，Service A 和 Service B，通過(guò) REST 接口通信；Service A 在某個(gè)業(yè)務(wù)場(chǎng)景下調(diào)用 Service B 的接口完成一個(gè)計(jì)算密集型任務(wù)，假設(shè)接口為 http://service_b/api/v1/domain；該任務(wù)運(yùn)行時(shí)間很長(zhǎng)，但 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 在請(qǐng)求 Service B 接口時(shí)帶上 callback uri，比如 http://service_b/api/v1/domain?callbackuri=http://service_a/api/v1/domain，Service B 收到請(qǐng)求后立即返回 HTTP Status 200 Ok，等任務(wù)完成后再調(diào)用 Service A callback uri 進(jìn)行通知。

   

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

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

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 是一對(duì)多的關(guān)系，當(dāng) Subject 狀態(tài)變更時(shí)，所有的 Observer 都會(huì)被通知到。

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

場(chǎng)景上下文

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

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

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

在上述socket - http - app 三層模型中，對(duì) socket 和 http，socket 是 Subject，http 是 Observer；對(duì) 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)的上下文對(duì)象
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對(duì)象
//socketImplSocket的默認(rèn)實(shí)現(xiàn)
typesocketImplstruct{
//關(guān)鍵點(diǎn)4:在Subject中持有Observer的集合
listeners[]SocketListener
}

//關(guān)鍵點(diǎn)5:為Subject定義注冊(cè)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)的上下文對(duì)象。上述例子為Handle方法，上下文對(duì)象為Packet。
3. 定義 Subject 對(duì)象，上述例子為socketImpl對(duì)象。當(dāng)然，也可以先將 Subject 抽象為接口，比如上述例子中的Socket接口，但大多數(shù)情況下都不是必須的。
4. 在 Subject 對(duì)象中，持有 Observer 接口的集合，上述例子為listeners屬性。讓 Subject 依賴 Observer 接口，能夠使 Subject 與具體的 Observer 實(shí)現(xiàn)解耦，提升代碼的可擴(kuò)展性。
5. 為 Subject 對(duì)象定義注冊(cè) Observer 的方法，上述例子為AddListener方法。
6. 當(dāng) Subject 狀態(tài)變更時(shí)，遍歷 Observer 集合，并調(diào)用它們的狀態(tài)更變處理方法，上述例子為Receive方法。

擴(kuò)展

發(fā)布-訂閱模式

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

從前文的觀察者模式實(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 引用，它們之間通常通過(guò)Broker來(lái)完成解耦。也即，Publisher 不知道有哪些 Subscriber，也不知道 Subscriber 的數(shù)量：

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

下面，我們通過(guò)簡(jiǎn)單的分布式應(yīng)用系統(tǒng)（示例代碼工程）中的 mq 模塊，展示一個(gè)簡(jiǎn)單的發(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對(duì)象
//memoryMq內(nèi)存消息隊(duì)列，通過(guò)channel實(shí)現(xiàn)
typememoryMqstruct{
//關(guān)鍵點(diǎn)3: Broker中維持一個(gè)隊(duì)列的map，其中key為topic，value為queue，go語(yǔ)言通常用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選擇對(duì)應(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選擇對(duì)應(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對(duì)象。
2. 定義 Broker 對(duì)象，Broker 是緩存消息的地方，上述例子為memoryMq對(duì)象。
3. 在 Broker 中維持一個(gè)隊(duì)列的 map，其中 key 為 topic，value 為 queue，go 語(yǔ)言通常用 chan 來(lái)實(shí)現(xiàn) queue，上述例子為queues屬性。
4. 為 Broker 定義 produce 方法，根據(jù)消息中的 topic 選擇對(duì)應(yīng)的 queue 發(fā)布消息，上述例子為Produce方法。
5. 為 Broker 定義 consume 方法，根據(jù) topic 選擇對(duì)應(yīng)的 queue 消費(fèi)消息，上述例子為Consume方法。

Push 模式 VS Pull 模式

實(shí)現(xiàn)觀察者模式和發(fā)布-訂閱模式時(shí)，都會(huì)涉及到Push 模式或Pull 模式的選取。所謂 Push 模式，指的是 Subject/Publisher 直接將消息推送給 Observer/Subscriber；所謂 Pull 模式，指的是 Observer/Subscriber 主動(dòng)向 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)前無(wú)消息可處理，將導(dǎo)致 Observer/Subscriber 空輪詢，可以采用類似 Kafka 的解決方案：讓 Observer/Subscriber 阻塞一定時(shí)長(zhǎng)，讓出 CPU，避免長(zhǎng)期無(wú)效的 CPU 空轉(zhuǎn)。

典型應(yīng)用場(chǎng)景

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

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

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

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

• 支持廣播通信。

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

缺點(diǎn)

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

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

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

文章配圖

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

審核編輯：湯梓紅