C语言实现设计模式—订阅者/发布者模式（Pub-Sub Pattern）

2023-10-11 14:40
文章标签 语言 实现 设计模式 模式 订阅 pattern sub pub 发布者

本文主要是介绍C语言实现设计模式—订阅者/发布者模式（Pub-Sub Pattern），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

  • C语言实现订阅者/发布者模式（Pub-Sub Pattern）
    • 模式介绍
    • 模式的优点
    • 订阅者/发布者模式实现
      • 节点定义
      • 发布者/订阅者接口注册
      • 统一调度实现
      • 外层接口

C语言实现订阅者/发布者模式（Pub-Sub Pattern）

模式介绍

发布订阅模式中，称为发布者的消息发送者不会将消息直接发送给订阅者，这意味着发布者和订阅者不知道彼此的存在。在发布者和订阅者之间存在第三个组件，称为消息代理或调度中心或中间件，它维持着发布者和订阅者之间的联系，过滤所有发布者传入的消息并相应地分发它们给订阅者。

订阅者把自己想订阅的事件注册到调度中心，当该事件触发时候，发布者发布该事件到调度中心，由调度中心统一调度订阅者注册到调度中心的处理代码。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qJwesHLW-1597830017185)(C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20200819170525741.png)]

模式的优点

  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 user_mode_def{USER_MODE_PUB,            /* Publisher */USER_MODE_SUB             /* Subscribe */  
}user_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   user_list;   rt_mq_t     rt_mq;user_mode   user;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_user_create(mq_magic_t *pmq, const char *name, uint8_t flag);
rt_err_t rz_mq_magic_send(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_send_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, user_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, user_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 false;}/* _find */if (NULL != _find_publisher(t, (pub_mq_magic_t *)pmq)) {rt_kprintf("%s():don't register (%x) repeated.\r\n", __FUNCTION__, pmq);return false;}pmq->user = USER_MODE_PUB;pmq->topic = topic;/* insert list */rt_list_insert_after(&t->pub_list, &pmq->user_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 false;}/* _find */if (NULL != _find_subscriber(t, (sub_mq_magic_t *)pmq)) {rt_kprintf("%s():don't register (%x) repeated.\r\n", __FUNCTION__, pmq);return false;}pmq->user = USER_MODE_SUB;pmq->topic = topic;/* insert list */rt_list_insert_after(&t->sub_list, &pmq->user_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;/* use 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;}return 0;
}/* base to topic, push meassage to ubscriber who subscibe this topic */
rt_err_t mq_magic_send(rz_mq_magic_t *t, topic_t topic, mq_magic_frame_t *buff)
{rt_list_t *pos;mq_magic_t *pmq;bool is_success = false;rt_list_for_each(pos, &t->sub_list) {pmq = (sub_mq_magic_t *)rt_container_of(pos, sub_mq_magic_t, user_list);if (topic == pmq->topic) {if (RT_EOK != rt_mq_send(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 service */
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, user_list);topic = mq_magic_recv(pmq, &mq_magic_data, RT_WAITING_NO);if (topic > 0) {mq_magic_send(t, topic, &mq_magic_data);/* send finish, reset topic */topic = 0;memset(&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_user_create(mq_magic_t *pmq, const char *name, uint8_t flag)
{if (NULL == pmq) {rt_kprintf("%s():(NULL == pmq) invalid.\r\n", __FUNCTION__);return false;}/* 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__);return false;}return true;}rt_err_t rz_mq_magic_send(mq_magic_t *t, mq_magic_frame_t *buffer)
{if (RT_EOK != rt_mq_send(t->rt_mq, buffer, sizeof(mq_magic_frame_t))) {//rt_kprintf("%s(): rt_mq_send failed.\r\n", __FUNCTION__);return RT_ERROR;}return RT_EOK;
}rt_err_t rz_mq_magic_recv(mq_magic_t *t, void *buffer, int32_t timeout)
{if (RT_EOK != rt_mq_recv(t->rt_mq, buffer, sizeof(mq_magic_frame_t), timeout)) {//rt_kprintf("%s(): rz_mq_magic_recv failed.\r\n", __FUNCTION__);return RT_ERROR;}return RT_EOK;}/*timeout: uint is ms
*/
rt_err_t rz_mq_magic_send_wait_receive(pub_mq_magic_t *ppub, sub_mq_magic_t *psub, mq_magic_frame_t *buffer, int32_t timeout)
{
#define WAITTIME_BY_ONCE (100)//uint32_t timeout_num = 0;//uint32_t wait_count = 0;if (RT_EOK != rt_mq_send(ppub->rt_mq, buffer, sizeof(mq_magic_frame_t))) {rt_kprintf("%s(): rt_mq_send failed.\r\n", __FUNCTION__);return RT_ERROR;}return RT_EOK;#if 0timeout_num = timeout / WAITTIME_BY_ONCE;if (0 != (timeout % WAITTIME_BY_ONCE)) {timeout_num++;}memset(buffer, 0, sizeof(mq_magic_frame_t));do {wait_count++;if (RT_EOK == rt_mq_recv(psub->rt_mq, buffer, sizeof(mq_magic_frame_t), RT_WAITING_NO))return RT_EOK;rt_thread_mdelay(WAITTIME_BY_ONCE);} while (wait_count < timeout_num);return RT_ERROR;#endif
}/* init */
int rz_mq_magic_init(void)
{rz_mq_magic_t *prz_mq_magic = rz_mq_magic_obj_get();if (NULL != prz_mq_magic) {prz_mq_magic->init(prz_mq_magic);}return RT_EOK;}INIT_APP_EXPORT(rz_mq_magic_init);

这篇关于C语言实现设计模式—订阅者/发布者模式（Pub-Sub Pattern）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/188758。

相关文章

Java实现字节字符转bcd编码

Java实现字节字符转bcd编码

《Java实现字节字符转bcd编码》BCD是一种将十进制数字编码为二进制的表示方式,常用于数字显示和存储,本文将介绍如何在Java中实现字节字符转BCD码的过程,需要的小伙伴可以了解下... 目录前言BCD码是什么Java实现字节转bcd编码方法补充总结前言BCD码（Binary-Coded Decima
阅读更多...
SpringBoot全局域名替换的实现

SpringBoot全局域名替换的实现

《SpringBoot全局域名替换的实现》本文主要介绍了SpringBoot全局域名替换的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录 项目结构⚙️ 配置文件application.yml️ 配置类AppProperties.Ja
阅读更多...
Python实现批量CSV转Excel的高性能处理方案

Python实现批量CSV转Excel的高性能处理方案

《Python实现批量CSV转Excel的高性能处理方案》在日常办公中,我们经常需要将CSV格式的数据转换为Excel文件,本文将介绍一个基于Python的高性能解决方案,感兴趣的小伙伴可以跟随小编一... 目录一、场景需求二、技术方案三、核心代码四、批量处理方案五、性能优化六、使用示例完整代码七、小结一、
阅读更多...
Java实现将HTML文件与字符串转换为图片

Java实现将HTML文件与字符串转换为图片

《Java实现将HTML文件与字符串转换为图片》在Java开发中,我们经常会遇到将HTML内容转换为图片的需求,本文小编就来和大家详细讲讲如何使用FreeSpire.DocforJava库来实现这一功... 目录前言核心实现：html 转图片完整代码场景 1：转换本地 HTML 文件为图片场景 2：转换 H
阅读更多...
C#使用Spire.Doc for .NET实现HTML转Word的高效方案

C#使用Spire.Doc for .NET实现HTML转Word的高效方案

《C#使用Spire.Docfor.NET实现HTML转Word的高效方案》在Web开发中,HTML内容的生成与处理是高频需求,然而,当用户需要将HTML页面或动态生成的HTML字符串转换为Wor... 目录引言一、html转Word的典型场景与挑战二、用 Spire.Doc 实现 HTML 转 Word1
阅读更多...
C#实现一键批量合并PDF文档

C#实现一键批量合并PDF文档

《C#实现一键批量合并PDF文档》这篇文章主要为大家详细介绍了如何使用C#实现一键批量合并PDF文档功能,文中的示例代码简洁易懂,感兴趣的小伙伴可以跟随小编一起学习一下... 目录前言效果展示功能实现1、添加文件2、文件分组（书签）3、定义页码范围4、自定义显示5、定义页面尺寸6、PDF批量合并7、其他方法
阅读更多...
SpringBoot实现不同接口指定上传文件大小的具体步骤

SpringBoot实现不同接口指定上传文件大小的具体步骤

《SpringBoot实现不同接口指定上传文件大小的具体步骤》：本文主要介绍在SpringBoot中通过自定义注解、AOP拦截和配置文件实现不同接口上传文件大小限制的方法,强调需设置全局阈值远大于... 目录一  springboot实现不同接口指定文件大小1.1 思路说明1.2 工程启动说明二 具体实施2
阅读更多...
Python实现精确小数计算的完全指南

Python实现精确小数计算的完全指南

《Python实现精确小数计算的完全指南》在金融计算、科学实验和工程领域,浮点数精度问题一直是开发者面临的重大挑战,本文将深入解析Python精确小数计算技术体系,感兴趣的小伙伴可以了解一下... 目录引言：小数精度问题的核心挑战一、浮点数精度问题分析1.1 浮点数精度陷阱1.2 浮点数误差来源二、基础解决
阅读更多...
Java实现在Word文档中添加文本水印和图片水印的操作指南

Java实现在Word文档中添加文本水印和图片水印的操作指南

《Java实现在Word文档中添加文本水印和图片水印的操作指南》在当今数字时代,文档的自动化处理与安全防护变得尤为重要,无论是为了保护版权、推广品牌,还是为了在文档中加入特定的标识,为Word文档添加... 目录引言Spire.Doc for Java：高效Word文档处理的利器代码实战：使用Java为Wo
阅读更多...
Java实现远程执行Shell指令

Java实现远程执行Shell指令

《Java实现远程执行Shell指令》文章介绍使用JSch在SpringBoot项目中实现远程Shell操作,涵盖环境配置、依赖引入及工具类编写,详解分号和双与号执行多指令的区别... 目录软硬件环境说明编写执行Shell指令的工具类总结jsch(Java Secure Channel)是SSH2的一个纯J
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2