基于觀察者模式設(shè)計(jì)的框架-REB，使代碼模塊化

設(shè)計(jì)模式里面的觀察者模式，一直是作者想去設(shè)計(jì)一套框架來(lái)闡述這一個(gè)模式，因此REB(Rice Event Broker)就是為了完成觀察者模式的一個(gè)框架。

觀察者模式

聊REB之前，我們聊聊觀察者模式帶給我們特性，他能對(duì)我們框架設(shè)計(jì)提供什么好處。

什么是觀察者模式

觀察者模式(Observer Pattern)是一種行為設(shè)計(jì)模式，用于定義對(duì)象之間的一對(duì)多依賴關(guān)系，使得一個(gè)對(duì)象的狀態(tài)變化會(huì)通知其所有依賴者并自動(dòng)更新它們的狀態(tài)。這個(gè)模式涉及兩種主要類型的對(duì)象：

被觀察者：也稱為主題或可觀察者，是一個(gè)對(duì)象，它維護(hù)一組觀察者（或依賴者）并提供方法來(lái)添加、刪除和通知這些觀察者。當(dāng)被觀察者的狀態(tài)發(fā)生變化時(shí)，它會(huì)通知所有已注冊(cè)的觀察者。

觀察者：觀察者是依賴于被觀察者的對(duì)象，它們實(shí)現(xiàn)一個(gè)接口或抽象類，包含一個(gè)更新方法（通常稱為update），用于接收并處理被觀察者的狀態(tài)變化通知。

觀察者模式工作流程

被觀察者注冊(cè)觀察者：被觀察者維護(hù)一個(gè)觀察者列表，并提供注冊(cè)（添加）和注銷（刪除）觀察者的方法。

被觀察者狀態(tài)變化：當(dāng)被觀察者的狀態(tài)發(fā)生變化，它會(huì)遍歷其觀察者列表，調(diào)用每個(gè)觀察者的更新方法，將狀態(tài)變化通知給它們。

觀察者響應(yīng)：每個(gè)觀察者在接收到通知后會(huì)執(zhí)行自己的更新邏輯，以響應(yīng)被觀察者的狀態(tài)變化。

觀察者模式優(yōu)勢(shì)

「解耦性：」觀察者模式可以幫助降低對(duì)象之間的耦合度。被觀察者和觀察者之間的關(guān)系是松散的，它們可以獨(dú)立演化，而不會(huì)影響彼此的具體實(shí)現(xiàn)。

「可擴(kuò)展性：」你可以輕松地添加新的觀察者，而不需要修改被觀察者的代碼。這種擴(kuò)展性使你能夠動(dòng)態(tài)地增加或刪除觀察者，以滿足不同的需求。

「通知機(jī)制：」觀察者模式允許被觀察者通知觀察者，從而使觀察者能夠在適當(dāng)?shù)臅r(shí)候進(jìn)行響應(yīng)。這可以幫助確保數(shù)據(jù)的一致性，因?yàn)橛^察者會(huì)立即知道被觀察者的狀態(tài)變化。

「分布式事件處理：」觀察者模式常用于實(shí)現(xiàn)分布式事件處理系統(tǒng)，其中多個(gè)觀察者可以遠(yuǎn)程訂閱和接收事件通知。

「可重用性：」觀察者模式可以在不同的應(yīng)用中重復(fù)使用，因?yàn)樗且粋€(gè)通用的設(shè)計(jì)模式，不受特定應(yīng)用領(lǐng)域的限制。

「靈活性：」觀察者模式可以用于許多不同的場(chǎng)景，如用戶界面更新、事件處理、數(shù)據(jù)同步等，使得代碼更加靈活和可維護(hù)。

「支持一對(duì)多關(guān)系：」觀察者模式支持一對(duì)多的依賴關(guān)系，這意味著一個(gè)被觀察者可以同時(shí)通知多個(gè)觀察者，從而實(shí)現(xiàn)多個(gè)對(duì)象之間的協(xié)同工作。

觀察者模式例子

物聯(lián)網(wǎng)協(xié)議MQTT：MQTT（Message Queuing Telemetry Transport，消息隊(duì)列遙測(cè)傳輸協(xié)議），是一種基于發(fā)布/訂閱（publish/subscribe）模式的“輕量級(jí)”通訊協(xié)議。

Android的EventBus：EventBus是一個(gè)基于發(fā)布者/訂閱者模式的事件總線框架。

REB框架設(shè)計(jì)

REB框架圖

REB框架說(shuō)明

REB框架分為3層：osal(OS抽象層)，REB核心層(包含發(fā)布者，觀察者，中間人)，應(yīng)用層(調(diào)用REB的模塊或應(yīng)用)。

osal(OS抽象層)：為了能讓此框架應(yīng)用于不同的操作系統(tǒng)，且不用修改框架本身，所以提供os適配層。

REB核心層(包含發(fā)布者，觀察者，中間人)：框架的三大角色，它們?nèi)呋ハ嘁蕾嚒?/p>

publisher(發(fā)布者)：REB框架的發(fā)布者支持4種接口：默認(rèn)發(fā)送接口，默認(rèn)發(fā)送完釋放數(shù)據(jù)內(nèi)存釋放接口，緊急發(fā)送接口，緊急發(fā)送完數(shù)據(jù)內(nèi)存釋放接口。

observer(觀察者)：REB框架的觀察者支持3種接口：信號(hào)接收接口，回調(diào)接收接口，線程接收接口。

broker(中間人)：REB框架的中間人支持兩種接口：觀察者只觀察一次接口，觀察者觀察多次接口。

應(yīng)用層(調(diào)用REB的模塊或應(yīng)用)：上層應(yīng)用或者模塊，相互獨(dú)立，互不依賴。

REB是以事件為導(dǎo)向，事件類型由主事件類型和次事件類型組成，事件類型占用32個(gè)位，主事件類型占高16位，次事件類型占低16位。一般：以網(wǎng)絡(luò)為例：主事件類型為：net_type，次事件類型為：link_up，link_down等。

REB目錄結(jié)構(gòu)

├─adapter
│├─cmsis
│|├─reb_mutex.c//cmsismutex適配層
│|├─reb_queue.c//cmsisqueue適配層
│|├─reb_sem.c//cmsissem適配層
│|└─reb_task.c//cmsistask適配層
│└─rtthread
│├─reb_mutex.c//rtthreadmutex適配層
│├─reb_queue.c//rtthreadqueue適配層
│├─reb_sem.c//rtthreadsem適配層
│└─reb_task.c//rtthreadtask適配層
├─example
│└─reb_rtt_example.c//rtthread平臺(tái)實(shí)例
├─include
│├─reb_broker.h//reb中間人的頭文件
│├─reb_cfg.h//reb參數(shù)配置文件
│├─reb_def.h//reb框架通用接口定義
│├─reb_observer.h//reb觀察者的頭文件
│└─reb_publisher.h//reb發(fā)布者的頭文件
└─src
├─reb_broker.c//reb中間人的源文件
├─reb_publisher.c//reb觀察者的源文件
└─reb_observer.c//reb發(fā)布者的源文件

REB接口說(shuō)明

broker接口

創(chuàng)建broker

在使用該框架時(shí)，必須要通過(guò)此接口創(chuàng)建broker，它是發(fā)布者和觀察者的中間人。

reb_statusbroker_create(void);

刪除broker

當(dāng)不再使用該框架時(shí)，可以調(diào)用此接口刪除broker。

reb_statusbroker_delete(void);

關(guān)聯(lián)觀察者到broker中，并只觀察一次

我創(chuàng)建的觀察者之后，需要通過(guò)此接口將觀察者關(guān)聯(lián)到broker中。當(dāng)發(fā)布者發(fā)布事件，可以通過(guò)broker找到對(duì)用的觀察者。使用該接口觀察者只觀察一次事件。

reb_statusbroker_observer_attach_once(observer_base*obs);

關(guān)聯(lián)觀察者到broker中，并只觀察多次

我創(chuàng)建的觀察者之后，需要通過(guò)此接口將觀察者關(guān)聯(lián)到broker中。當(dāng)發(fā)布者發(fā)布事件，可以通過(guò)broker找到對(duì)用的觀察者。使用該接口觀察者只觀察多次事件。

reb_statusbroker_observer_attach(observer_base*obs);

從broker中脫離觀察者

reb_statusbroker_observer_detach(observer_base*obs);

observer接口

創(chuàng)建信號(hào)模式的觀察者

該接口是創(chuàng)建信號(hào)模式的觀察者，它只接收事件信號(hào)，不傳輸數(shù)據(jù)的。

observer_base*observer_signal_create(uint16_ttype,uint16_tsub_type);

信號(hào)模式的觀察者，等待同步信號(hào)

該接口是信號(hào)模式的觀察者，用戶層需要通過(guò)一個(gè)任務(wù)監(jiān)聽(tīng)觀察事件的同步信號(hào)接口。

reb_statusobserver_signal_wait(observer_base*base,reb_time_ttimeout);

創(chuàng)建回調(diào)模式的觀察者

該接口是創(chuàng)建回調(diào)模式的觀察者，當(dāng)事件產(chǎn)生時(shí)，broker會(huì)通過(guò)回調(diào)的方式通知觀察者事件的到來(lái)。

observer_base*observer_callback_create(uint16_ttype,
uint16_tsub_type,
obs_callback_cbcb,
void*arg);

創(chuàng)建任務(wù)模式的觀察者

該接口是創(chuàng)建任務(wù)模式的觀察者，當(dāng)事件產(chǎn)生時(shí)，broker會(huì)通過(guò)創(chuàng)建一個(gè)線程，然后由獨(dú)立的線程將事件通知給觀察者。

observer_base*observer_task_create(uint16_ttype,
uint16_tsub_type,
obs_task_cbrun,
void*arg,
uint32_tstack_size,
uint32_tprio);

從broker中脫離觀察者

reb_statusobserver_delete(observer_base*base);

publisher接口

創(chuàng)建發(fā)布者工廠

該接口是創(chuàng)建發(fā)布者工廠，提供事件隊(duì)列，使發(fā)布消息處于非阻塞式發(fā)送

reb_statuspublisher_factory_create(pub_notifynotify);

發(fā)布者默認(rèn)發(fā)送消息

該接口是發(fā)布者發(fā)布事件接口，它是采用先進(jìn)先出的方式發(fā)送消息

reb_statuspublisher_send(uint16_ttype,uint16_tsub_type,
uint32_tdata,reb_time_ttimeout);

發(fā)布者默認(rèn)發(fā)送消息，發(fā)送完成之后把消息緩沖刪除

該接口是發(fā)布者發(fā)布事件接口，它是采用先進(jìn)先出的方式發(fā)送消息，并且將消息發(fā)送給所有觀察者之后，數(shù)據(jù)的內(nèi)存會(huì)執(zhí)行釋放。

reb_statuspublisher_send_with_free(uint16_ttype,uint16_tsub_type,
uint32_tdata,reb_time_ttimeout);

發(fā)布者發(fā)送緊急消息

該接口是發(fā)布者發(fā)布事件接口，它是采用插隊(duì)的方式發(fā)送消息，它會(huì)將發(fā)布的消息插入消息隊(duì)列的頭部。

reb_statuspublisher_urgent_send(uint16_ttype,uint16_tsub_type,
uint32_tdata,reb_time_ttimeout);

發(fā)布者發(fā)送緊急消息，發(fā)送完成之后把消息緩沖刪除

該接口是發(fā)布者發(fā)布事件接口，它它是采用插隊(duì)的方式發(fā)送消息，它會(huì)將發(fā)布的消息插入消息隊(duì)列的頭部。并且將消息發(fā)送給所有觀察者之后，數(shù)據(jù)的內(nèi)存會(huì)執(zhí)行釋放。

reb_statuspublisher_urgent_send_with_free(uint16_ttype,uint16_tsub_type,
uint32_tdata,reb_time_ttimeout);

REB驗(yàn)證

創(chuàng)建三個(gè)不同模式的觀察者，并關(guān)聯(lián)到broker中。

通過(guò)多次不發(fā)布事件，查看觀察者是否能接收到事件。

#include"rtthread.h"
#include"reb_broker.h"
#include"reb_observer.h"
#include"reb_publisher.h"

observer_base*obs_signal;
observer_base*obs_call;
observer_base*obs_task;

voidsig_thread_handle(void*arg)//信號(hào)模式觀察者監(jiān)聽(tīng)同步信號(hào)
{
while(1){
if(observer_signal_wait(obs_signal,RT_WAITING_FOREVER)==REB_OK){
rt_kprintf("signal:recvsuccessrn");
}
}
}

voidobs_callback(uint32_tevent,uint32_tdata,void*arg)//回調(diào)模式觀察者處理函數(shù)
{
rt_kprintf("call:event:0x%08x,data:%srn",event,(char*)data);
}

voidobs_task_fun(uint32_tevent,uint32_tdata,void*arg)//任務(wù)模式觀察者任務(wù)處理函數(shù)
{
rt_kprintf("task:event:0x%08x,data:%srn",event,(char*)data);
}

intreb_init(void)
{
rt_thread_tsignal_thread=NULL;

broker_create();//broker創(chuàng)建

obs_signal=observer_signal_create(1,REB_ALL_MINOR_TYPE);//創(chuàng)建信號(hào)模式觀察者

signal_thread=rt_thread_create("sig_thread",sig_thread_handle,NULL,1024,10,20);//創(chuàng)建線程，等待信號(hào)模式下的事件
rt_thread_startup(signal_thread);

obs_call=observer_callback_create(1,REB_ALL_MINOR_TYPE,obs_callback,NULL);//創(chuàng)建回調(diào)模式觀察者

obs_task=observer_task_create(1,REB_ALL_MINOR_TYPE,obs_task_fun,NULL,1024,15);//創(chuàng)建任務(wù)模式觀察者

broker_observer_attach(obs_signal);//關(guān)聯(lián)信號(hào)模式觀察者
broker_observer_attach(obs_call);//關(guān)聯(lián)回調(diào)模式觀察者
broker_observer_attach_once(obs_task);//關(guān)聯(lián)任務(wù)模式觀察者

returnRT_EOK;
}
INIT_COMPONENT_EXPORT(reb_init);

intreb_test(void)
{
char*data="RiceChen";
publisher_send(1,1,(int)data,1000);//發(fā)布事件
publisher_send(1,2,(int)data,1000);//發(fā)布事件
publisher_send(1,3,(int)data,1000);//發(fā)布事件
}
MSH_CMD_EXPORT(reb_test,RiceEventbrokertest);

REB總結(jié)

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

REB提供了多種的模式的觀察者，可以根據(jù)需求選擇不同模式的觀察者。

REB的發(fā)布者可以支持緊急事件發(fā)布和非緊急事件發(fā)送，保證了事件的及時(shí)響應(yīng)。

REB增加了OSAL層，使其不依賴于任何的平臺(tái)，可以很方便的移植到其他平臺(tái)。

REB缺點(diǎn)：

REB的事件處理采用串行的方式，當(dāng)事件積累多了，負(fù)載會(huì)比較大(后續(xù)優(yōu)化)。

REB已經(jīng)被合并到RT-THREAD軟件包中，并且得到RT-THREAD技術(shù)總監(jiān)的認(rèn)可。開(kāi)源鏈接：https://github.com/RiceChen0/reb

?

審核編輯 黃宇