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

觀察者模式

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

什么是觀察者模式

觀察者模式(Observer Pattern)是一種行為設計模式，用于定義對象之間的一對多依賴關系，使得一個對象的狀態變化會通知其所有依賴者并自動更新它們的狀態。這個模式涉及兩種主要類型的對象：

被觀察者：也稱為主題或可觀察者，是一個對象，它維護一組觀察者（或依賴者）并提供方法來添加、刪除和通知這些觀察者。當被觀察者的狀態發生變化時，它會通知所有已注冊的觀察者。

觀察者：觀察者是依賴于被觀察者的對象，它們實現一個接口或抽象類，包含一個更新方法（通常稱為update），用于接收并處理被觀察者的狀態變化通知。

觀察者模式工作流程

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

被觀察者狀態變化：當被觀察者的狀態發生變化，它會遍歷其觀察者列表，調用每個觀察者的更新方法，將狀態變化通知給它們。

觀察者響應：每個觀察者在接收到通知后會執行自己的更新邏輯，以響應被觀察者的狀態變化。

觀察者模式優勢

「解耦性：」觀察者模式可以幫助降低對象之間的耦合度。被觀察者和觀察者之間的關系是松散的，它們可以獨立演化，而不會影響彼此的具體實現。

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

「通知機制：」觀察者模式允許被觀察者通知觀察者，從而使觀察者能夠在適當的時候進行響應。這可以幫助確保數據的一致性，因為觀察者會立即知道被觀察者的狀態變化。

「分布式事件處理：」觀察者模式常用于實現分布式事件處理系統，其中多個觀察者可以遠程訂閱和接收事件通知。

「可重用性：」觀察者模式可以在不同的應用中重復使用，因為它是一個通用的設計模式，不受特定應用領域的限制。

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

「支持一對多關系：」觀察者模式支持一對多的依賴關系，這意味著一個被觀察者可以同時通知多個觀察者，從而實現多個對象之間的協同工作。

觀察者模式例子

物聯網協議MQTT：MQTT（Message Queuing Telemetry Transport，消息隊列遙測傳輸協議），是一種基于發布/訂閱（publish/subscribe）模式的“輕量級”通訊協議。

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

REB框架設計

REB框架圖

REB框架說明

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

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

REB核心層(包含發布者，觀察者，中間人)：框架的三大角色，它們三者互相依賴。

publisher(發布者)：REB框架的發布者支持4種接口：默認發送接口，默認發送完釋放數據內存釋放接口，緊急發送接口，緊急發送完數據內存釋放接口。

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

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

應用層(調用REB的模塊或應用)：上層應用或者模塊，相互獨立，互不依賴。

REB是以事件為導向，事件類型由主事件類型和次事件類型組成，事件類型占用32個位，主事件類型占高16位，次事件類型占低16位。一般：以網絡為例：主事件類型為：net_type，次事件類型為：link_up，link_down等。

REB目錄結構

├─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平臺實例
├─include
│├─reb_broker.h//reb中間人的頭文件
│├─reb_cfg.h//reb參數配置文件
│├─reb_def.h//reb框架通用接口定義
│├─reb_observer.h//reb觀察者的頭文件
│└─reb_publisher.h//reb發布者的頭文件
└─src
├─reb_broker.c//reb中間人的源文件
├─reb_publisher.c//reb觀察者的源文件
└─reb_observer.c//reb發布者的源文件

REB接口說明

broker接口

接口	說明
broker_create	創建broker
broker_delete	刪除broker
broker_observer_attach_once	關聯觀察者到broker中，并只觀察一次
broker_observer_attach	關聯觀察者到broker中，并只觀察多次
broker_observer_detach	從broker中脫離觀察者

創建broker

在使用該框架時，必須要通過此接口創建broker，它是發布者和觀察者的中間人。

reb_statusbroker_create(void);

「參數」	「描述」
--	--
「返回」	——
REB_OK	broker創建成功
REB_ERROR	broker創建失敗

刪除broker

當不再使用該框架時，可以調用此接口刪除broker。

reb_statusbroker_delete(void);

「參數」	「描述」
--	--
「返回」	——
REB_OK	broker刪除成功
REB_ERROR	broker刪除失敗

關聯觀察者到broker中，并只觀察一次

我創建的觀察者之后，需要通過此接口將觀察者關聯到broker中。當發布者發布事件，可以通過broker找到對用的觀察者。使用該接口觀察者只觀察一次事件。

reb_statusbroker_observer_attach_once(observer_base*obs);

「參數」	「描述」
obs	觀察者對象
「返回」	——
REB_OK	關聯觀察者到broker中，成功
REB_ERROR	關聯觀察者到broker中，失敗

關聯觀察者到broker中，并只觀察多次

我創建的觀察者之后，需要通過此接口將觀察者關聯到broker中。當發布者發布事件，可以通過broker找到對用的觀察者。使用該接口觀察者只觀察多次事件。

reb_statusbroker_observer_attach(observer_base*obs);

「參數」	「描述」
obs	觀察者對象
「返回」	——
REB_OK	關聯觀察者到broker中，成功
REB_ERROR	關聯觀察者到broker中，失敗

從broker中脫離觀察者

reb_statusbroker_observer_detach(observer_base*obs);

「參數」	「描述」
obs	觀察者對象
「返回」	——
REB_OK	觀察者從broker中脫離，成功
REB_ERROR	觀察者從broker中脫離，失敗

observer接口

接口	說明
observer_signal_create	創建信號模式的觀察者，只接收事件信號，不傳輸數據的觀察者
observer_signal_wait	信號模式的觀察者，等待同步信號
observer_callback_create	創建回調模式的觀察者
observer_task_create	創建任務模式的觀察者
observer_delete	刪除觀察者

創建信號模式的觀察者

該接口是創建信號模式的觀察者，它只接收事件信號，不傳輸數據的。

observer_base*observer_signal_create(uint16_ttype,uint16_tsub_type);

「參數」	「描述」
type	觀察者觀察的主事件類型
sub_type	觀察者觀察的次事件類型
「返回」	——
obs	觀察者創建成功
NULL	觀察者創建失敗

信號模式的觀察者，等待同步信號

該接口是信號模式的觀察者，用戶層需要通過一個任務監聽觀察事件的同步信號接口。

reb_statusobserver_signal_wait(observer_base*base,reb_time_ttimeout);

「參數」	「描述」
base	觀察者對象
timeout	觀察事件的超時事件
「返回」	——
REB_OK	觀察到對應事件
OTHER	觀察失敗

創建回調模式的觀察者

該接口是創建回調模式的觀察者，當事件產生時，broker會通過回調的方式通知觀察者事件的到來。

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

「參數」	「描述」
type	觀察者觀察的主事件類型
sub_type	觀察者觀察的次事件類型
cb	事件產生時，回調的接口函數
arg	回調函數的用戶數據
「返回」	——
obs	觀察者創建成功
NULL	觀察者創建失敗

創建任務模式的觀察者

該接口是創建任務模式的觀察者，當事件產生時，broker會通過創建一個線程，然后由獨立的線程將事件通知給觀察者。

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

「參數」	「描述」
type	觀察者觀察的主事件類型
sub_type	觀察者觀察的次事件類型
run	事件產生時，線程的處理函數
arg	線程處理函數的用戶數據
stack_size	線程的?？臻g大小
prio	線程的優先級
「返回」	——
obs	觀察者創建成功
NULL	觀察者創建失敗

從broker中脫離觀察者

reb_statusobserver_delete(observer_base*base);

「參數」	「描述」
base	觀察者對象
「返回」	——
REB_OK	觀察者刪除成功
REB_ERROR	觀察者刪除失敗

publisher接口

接口	說明
publisher_factory_create	創建發布者工廠
publisher_send	發布者默認發送消息
publisher_send_with_free	發布者默認發送消息，發送完成之后把消息緩沖刪除
publisher_urgent_send	發布者發送緊急消息
publisher_urgent_send_with_free	發布者發送緊急消息，發送完成之后把消息緩沖刪除

創建發布者工廠

該接口是創建發布者工廠，提供事件隊列，使發布消息處于非阻塞式發送

reb_statuspublisher_factory_create(pub_notifynotify);

「參數」	「描述」
notify	事件通知回調，當發布者發布消息之后，通過回調通知broker
「返回」	——
REB_OK	發布者工廠創建成功
REB_ERROR	發布者工廠創建失敗

發布者默認發送消息

該接口是發布者發布事件接口，它是采用先進先出的方式發送消息

reb_statuspublisher_send(uint16_ttype,uint16_tsub_type,
uint32_tdata,reb_time_ttimeout);

「參數」	「描述」
type	發布消息的主事件類型
sub_type	發布消息的次事件類型
data	發布消息的數據
timeout	發布消息的超時時間
「返回」	——
REB_OK	發布消息成功
OTHER	發布消息失敗

發布者默認發送消息，發送完成之后把消息緩沖刪除

該接口是發布者發布事件接口，它是采用先進先出的方式發送消息，并且將消息發送給所有觀察者之后，數據的內存會執行釋放。

reb_statuspublisher_send_with_free(uint16_ttype,uint16_tsub_type,
uint32_tdata,reb_time_ttimeout);

「參數」	「描述」
type	發布消息的主事件類型
sub_type	發布消息的次事件類型
data	發布消息的數據
timeout	發布消息的超時時間
「返回」	——
REB_OK	發布消息成功
OTHER	發布消息失敗

發布者發送緊急消息

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

reb_statuspublisher_urgent_send(uint16_ttype,uint16_tsub_type,
uint32_tdata,reb_time_ttimeout);

「參數」	「描述」
type	發布消息的主事件類型
sub_type	發布消息的次事件類型
data	發布消息的數據
timeout	發布消息的超時時間
「返回」	——
REB_OK	發布消息成功
OTHER	發布消息失敗

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

該接口是發布者發布事件接口，它它是采用插隊的方式發送消息，它會將發布的消息插入消息隊列的頭部。并且將消息發送給所有觀察者之后，數據的內存會執行釋放。

reb_statuspublisher_urgent_send_with_free(uint16_ttype,uint16_tsub_type,
uint32_tdata,reb_time_ttimeout);

「參數」	「描述」
type	發布消息的主事件類型
sub_type	發布消息的次事件類型
data	發布消息的數據
timeout	發布消息的超時時間
「返回」	——
REB_OK	發布消息成功
OTHER	發布消息失敗

REB驗證

創建三個不同模式的觀察者，并關聯到broker中。

通過多次不發布事件，查看觀察者是否能接收到事件。

#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)//信號模式觀察者監聽同步信號
{
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)//回調模式觀察者處理函數
{
rt_kprintf("call:event:0x%08x,data:%srn",event,(char*)data);
}

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

intreb_init(void)
{
rt_thread_tsignal_thread=NULL;

broker_create();//broker創建

obs_signal=observer_signal_create(1,REB_ALL_MINOR_TYPE);//創建信號模式觀察者

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

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

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

broker_observer_attach(obs_signal);//關聯信號模式觀察者
broker_observer_attach(obs_call);//關聯回調模式觀察者
broker_observer_attach_once(obs_task);//關聯任務模式觀察者

returnRT_EOK;
}
INIT_COMPONENT_EXPORT(reb_init);

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

REB總結

REB優點：

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

REB的發布者可以支持緊急事件發布和非緊急事件發送，保證了事件的及時響應。

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

REB缺點：

REB的事件處理采用串行的方式，當事件積累多了，負載會比較大(后續優化)。

REB已經被合并到RT-THREAD軟件包中，并且得到RT-THREAD技術總監的認可。開源鏈接：https://github.com/RiceChen0/reb

?

審核編輯 黃宇