C语⾔实现设计模式—订阅者发布者模式（Pub-SubPattern）
⽂章⽬录
C语⾔实现订阅者/发布者模式（Pub-Sub Pattern）
模式介绍
发布订阅模式中，称为发布者的消息发送者不会将消息直接发送给订阅者，这意味着发布者和订阅者不知道彼此的存在。在发布者和订阅者之间存在第三个组件，称为消息代理或调度中⼼或中间件，它维持着发布者和订阅者之间的联系，过滤所有发布者传⼊的消息并相应地分发它们给订阅者。
订阅者把⾃⼰想订阅的事件注册到调度中⼼，当该事件触发时候，发布者发布该事件到调度中⼼，由调度中⼼统⼀调度订阅者注册到调度中⼼的处理代码。
模式的优点
1. 发布者与订阅者松耦合，甚⾄不需要知道它们的存在。由于主题才是关注的焦点，发布者和订阅者可以对系统拓扑结构保持⼀⽆所
知。各⾃继续正常操作⽽⽆需顾及对⽅。
2. 不需要关⼼不同的组件是如何组合在⼀起的，只要他们共同遵守⼀份协议即可。
3. 异步的消息传递有助于应⽤程序在增加的负载下继续平稳运⾏，并且可以更有效地处理间歇性故障。发布者/订阅者模式更像是⼀种跨
应⽤程序模式。发布服务器和订阅服务器可以驻留在两个不同的应⽤程序中。它们中的每⼀个都通过消息代理或消息队列进⾏通信。
订阅者/发布者模式实现
节点定义
#ifndef __MQ_MAGIC_H__
#define __MQ_MAGIC_H__
#include<stdbool.h>
#include"lw_oopc.h"
#include<rtthread.h>
typedef int32_t topic_t;
typedef enum ur_mode_def{
USER_MODE_PUB,/* Publisher */
USER_MODE_SUB /* Subscribe */
}ur_mode;
typedef struct mq_magic_frame_def{
topic_t head;
int32_t value;
}mq_magic_frame_t, pmq_magic_frame_t;
typedef struct mq_magic_def{
rt_list_t ur_list;
rt_mq_t rt_mq;
ur_mode ur;
topic_t topic;
}mq_magic_t,*pmq_magic_t;屏闪
typedef mq_magic_t pub_mq_magic_t;
typedef mq_magic_t sub_mq_magic_t;
typedef struct rz_mq_magic_t rz_mq_magic_t;
CLASS(rz_mq_magic_t)
{
rt_device_t tid_device;
rt_list_t pub_list;
rt_list_t sub_list;
/* 发布者注册接⼝ */
bool (*publisher_register)(rz_mq_magic_t *t, mq_magic_t *pmq, topic_t topic);
上大学网
/* 订阅者注册接⼝ */
bool (*subscriber_register)(rz_mq_magic_t *t, mq_magic_t *pmq, topic_t topic);
void(*init)(rz_mq_magic_t *t);
};
rz_mq_magic_t *rz_mq_magic_obj_get(void);
bool mq_magic_ur_create(mq_magic_t *pmq,const char*name, uint8_t flag);
rt_err_t rz_mq_magic_nd(mq_magic_t *t, mq_magic_frame_t *buffer);穴居狼蛛
跳台阶
rt_err_t rz_mq_magic_recv(mq_magic_t *t,void*buffer, int32_t timeout);
rt_err_t rz_mq_magic_nd_wait_receive(pub_mq_magic_t *ppub, sub_mq_magic_t *psub, mq_magic_frame_t *buffer, int32_t timeout); int rz_mq_magic_init(void);
#endif
发布者/订阅者接⼝注册
static pub_mq_magic_t *_find_publisher(rz_mq_magic_t *t, pub_mq_magic_t *pmq)
{
rt_list_t *pos;
pub_mq_magic_t *ppub;
rt_list_for_each(pos,&t->pub_list){
ppub =(pub_mq_magic_t *)rt_container_of(pos, pub_mq_magic_t, ur_list);
if(ppub == pmq){
return ppub;
}
}
return NULL;
}
护士辞职报告static sub_mq_magic_t *_find_subscriber(rz_mq_magic_t *t, sub_mq_magic_t *pmq) {
rt_list_t *pos;
pub_mq_magic_t *psub;
rt_list_for_each(pos,&t->sub_list){
psub =(sub_mq_magic_t *)rt_container_of(pos, sub_mq_magic_t, ur_list);
if(psub == pmq){
return psub;
}
}
return NULL;
}
static bool publisher_register(rz_mq_magic_t *t, mq_magic_t *pmq, topic_t topic)
{
if(NULL== pmq){
rt_kprintf("%s():(NULL == pmq) invalid.\r\n", __FUNCTION__);
return fal;
}
/
* _find */
if(NULL!=_find_publisher(t,(pub_mq_magic_t *)pmq)){
rt_kprintf("%s():don't register (%x) repeated.\r\n", __FUNCTION__, pmq);
return fal;
}
pmq->ur = USER_MODE_PUB;
pmq->topic = topic;
/* inrt list */
rt_list_inrt_after(&t->pub_list,&pmq->ur_list);
return true;
}
static bool subscriber_register(rz_mq_magic_t *t, mq_magic_t *pmq, topic_t topic) {
if(NULL== pmq){
rt_kprintf("%s():(NULL == pmq) invalid.\r\n", __FUNCTION__);
return fal;
}
/* _find */
if(NULL!=_find_subscriber(t,(sub_mq_magic_t *)pmq)){
rt_kprintf("%s():don't register (%x) repeated.\r\n", __FUNCTION__, pmq);
return fal;
}
pmq->ur = USER_MODE_SUB;
pmq->topic = topic;
/* inrt list */
rt_list_inrt_after(&t->sub_list,&pmq->ur_list);
return true;
}
统⼀调度实现
#define RTHREAD_PRIORITY_MQ_MAGIC 28
#define THREAD_STACK_SIZE_MQ_MAGIC 1024
#define THREAD_TIMESLICE_MQ_MAGIC 5
static rz_mq_magic_t *rz_mq_magic_obj =NULL;
/* u singleton pattern */
rz_mq_magic_t *rz_mq_magic_obj_get(void)
{
if(NULL== rz_mq_magic_obj){
rz_mq_magic_obj =rz_mq_magic_t_new();
if(NULL== rz_mq_magic_obj){
rt_kprintf("rz_mq_magic_t_new failed\r\n");
}
}
return(rz_mq_magic_t *)rz_mq_magic_obj;
}
/* receive meassage, then return topic */
topic_t mq_magic_recv(mq_magic_t *t, mq_magic_frame_t *buff, int32_t timeout) {
if(RT_EOK ==rt_mq_recv(t->rt_mq, buff,sizeof(mq_magic_frame_t), timeout)){ return t->topic;
}
return0;
}
/* ba to topic, push meassage to ubscriber who subscibe this topic */
rt_err_t mq_magic_nd(rz_mq_magic_t *t, topic_t topic, mq_magic_frame_t *buff) {
rt_list_t *pos;
mq_magic_t *pmq;
bool is_success = fal;
rt_list_for_each(pos,&t->sub_list){
pmq =(sub_mq_magic_t *)rt_container_of(pos, sub_mq_magic_t, ur_list);
if(topic == pmq->topic){
if(RT_EOK !=rt_mq_nd(pmq->rt_mq, buff,sizeof(mq_magic_frame_t))){ continue;
}
is_success = true;
}
}
if(!is_success){
return RT_ERROR;
}
return RT_EOK;
}
/* main work : message relay rvice */
static void mq_magic_broker_process(void*parameter)
{
rz_mq_magic_t *t =(rz_mq_magic_t *)parameter;
rt_list_t *pos;
mq_magic_t *pmq;
mq_magic_frame_t mq_magic_data;
topic_t topic;
while(1){
/* check all publishers */
rt_list_for_each(pos,&t->pub_list){
pmq =(pub_mq_magic_t *)rt_container_of(pos, pub_mq_magic_t, ur_list); topic =mq_magic_recv(pmq,&mq_magic_data, RT_WAITING_NO);
培优辅差工作总结if(topic >0){
mq_magic_nd(t, topic,&mq_magic_data);
/* nd finish, ret topic */
topic =0;
memt(&mq_magic_data,0,sizeof(mq_magic_data));
}
}
rt_thread_delay(10);
}
}
static void mq_magic_init(rz_mq_magic_t *t)
{
rt_list_init(&t->pub_list);
七大未解之谜rt_list_init(&t->sub_list);
rt_thread_t tid = RT_NULL;
tid =rt_thread_create("mq_magic",
mq_magic_broker_process, t,
THREAD_STACK_SIZE_MQ_MAGIC,
RTHREAD_PRIORITY_MQ_MAGIC,
THREAD_TIMESLICE_MQ_MAGIC);
if(RT_NULL != tid){
rt_thread_startup(tid);
}
}
CTOR(rz_mq_magic_t)
FUNCTION_SETTING(publisher_register, publisher_register);
FUNCTION_SETTING(subscriber_register, subscriber_register); FUNCTION_SETTING(init, mq_magic_init);
END_CTOR
外层接⼝
/* ---------------------------------------------- */
bool mq_magic_ur_create(mq_magic_t *pmq,const char*name, uint8_t flag) {
if(NULL== pmq){
rt_kprintf("%s():(NULL == pmq) invalid.\r\n", __FUNCTION__);
return fal;
}
/* create rt_mq */
pmq->rt_mq =rt_mq_create(name,sizeof(mq_magic_frame_t),10, flag);
if(NULL== pmq->rt_mq){
rt_kprintf("%s():rt_mq_create failed.\r\n", __FUNCTION__);
proud名词return fal;
}
return true;
}
