C语言-发布订阅模式详解与实践

C语言发布订阅模式详解与实践

1. 什么是发布订阅模式？

发布订阅模式定义了一种一对多的依赖关系，让多个订阅者对象同时监听某一个主题。这个主题在状态发生变化时，会通知所有依赖于它的订阅者对象，使它们能够自动更新。

2. 为什么需要发布订阅模式？

• 实现对象间的松耦合
• 支持广播通信
• 动态订阅和取消订阅
• 事件驱动架构
• 异步消息处理

3. 实际应用场景

• 传感器数据分发
• 消息队列系统
• 事件处理系统
• 日志监控
• 状态更新通知

4. 代码实现

4.1 UML 关系图

4.2 头文件 (pubsub.h)

#ifndef PUBSUB_H
#define PUBSUB_H

#include <stdint.h>
#include <stdbool.h>

// 主题数据结构
typedef struct {
    char name[32];          // 主题名称
    void* data;            // 主题数据
    uint32_t data_size;    // 数据大小
    uint32_t timestamp;    // 时间戳
} Topic;

// 订阅者回调函数类型
typedef void (*SubscriberCallback)(const Topic* topic, void* user_data);

// 订阅者结构
typedef struct {
    char name[32];              // 订阅者名称
    SubscriberCallback callback; // 回调函数
    void* user_data;            // 用户数据
} Subscriber;

// 发布者结构
typedef struct {
    char name[32];              // 发布者名称
    Subscriber* subscribers[16]; // 订阅者列表
    int subscriber_count;       // 订阅者数量
} Publisher;

// 创建发布者
Publisher* create_publisher(const char* name);

// 销毁发布者
void destroy_publisher(Publisher* publisher);

// 订阅主题
bool subscribe(Publisher* publisher, 
              const char* subscriber_name,
              SubscriberCallback callback,
              void* user_data);

// 取消订阅
bool unsubscribe(Publisher* publisher, const char* subscriber_name);

// 发布主题
void publish(Publisher* publisher, const Topic* topic);

#endif // PUBSUB_H

4.3 实现文件 (pubsub.c)

#include "pubsub.h"
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <time.h>

Publisher* create_publisher(const char* name) {
    Publisher* publisher = (Publisher*)malloc(sizeof(Publisher));
    strncpy(publisher->name, name, sizeof(publisher->name) - 1);
    publisher->subscriber_count = 0;
    memset(publisher->subscribers, 0, sizeof(publisher->subscribers));
    return publisher;
}

void destroy_publisher(Publisher* publisher) {
    if (!publisher) return;
    
    // 释放所有订阅者
    for (int i = 0; i < publisher->subscriber_count; i++) {
        free(publisher->subscribers[i]);
    }
    free(publisher);
}

bool subscribe(Publisher* publisher, 
              const char* subscriber_name,
              SubscriberCallback callback,
              void* user_data) {
    if (!publisher || !subscriber_name || !callback) return false;
    
    // 检查是否已达到最大订阅者数量
    if (publisher->subscriber_count >= 16) {
        printf("订阅者数量已达上限\n");
        return false;
    }
    
    // 检查是否已订阅
    for (int i = 0; i < publisher->subscriber_count; i++) {
        if (strcmp(publisher->subscribers[i]->name, subscriber_name) == 0) {
            printf("订阅者 %s 已存在\n", subscriber_name);
            return false;
        }
    }
    
    // 创建新订阅者
    Subscriber* subscriber = (Subscriber*)malloc(sizeof(Subscriber));
    strncpy(subscriber->name, subscriber_name, sizeof(subscriber->name) - 1);
    subscriber->callback = callback;
    subscriber->user_data = user_data;
    
    // 添加到订阅者列表
    publisher->subscribers[publisher->subscriber_count++] = subscriber;
    printf("订阅者 %s 已添加\n", subscriber_name);
    return true;
}

bool unsubscribe(Publisher* publisher, const char* subscriber_name) {
    if (!publisher || !subscriber_name) return false;
    
    for (int i = 0; i < publisher->subscriber_count; i++) {
        if (strcmp(publisher->subscribers[i]->name, subscriber_name) == 0) {
            // 释放订阅者
            free(publisher->subscribers[i]);
            
            // 移动后续订阅者
            for (int j = i; j < publisher->subscriber_count - 1; j++) {
                publisher->subscribers[j] = publisher->subscribers[j + 1];
            }
            
            publisher->subscriber_count--;
            printf("订阅者 %s 已移除\n", subscriber_name);
            return true;
        }
    }
    
    printf("未找到订阅者 %s\n", subscriber_name);
    return false;
}

void publish(Publisher* publisher, const Topic* topic) {
    if (!publisher || !topic) return;
    
    printf("\n发布者 %s 发布主题 %s\n", publisher->name, topic->name);
    
    // 通知所有订阅者
    for (int i = 0; i < publisher->subscriber_count; i++) {
        Subscriber* subscriber = publisher->subscribers[i];
        printf("通知订阅者 %s\n", subscriber->name);
        subscriber->callback(topic, subscriber->user_data);
    }
}

4.4 使用示例 (main.c)

#include "pubsub.h"
#include <stdio.h>

// 温度传感器订阅者回调
void temperature_callback(const Topic* topic, void* user_data) {
    float* threshold = (float*)user_data;
    float temperature = *(float*)topic->data;
    
    printf("温度传感器收到数据: %.1f°C\n", temperature);
    if (temperature > *threshold) {
        printf("警告：温度超过阈值 %.1f°C!\n", *threshold);
    }
}

// 日志记录订阅者回调
void logger_callback(const Topic* topic, void* user_data) {
    printf("日志记录器：主题 %s, 数据大小 %d, 时间戳 %d\n",
           topic->name, topic->data_size, topic->timestamp);
}

int main() {
    // 创建发布者
    Publisher* sensor_publisher = create_publisher("传感器发布者");
    
    // 创建温度阈值
    float temp_threshold = 30.0f;
    
    // 订阅主题
    subscribe(sensor_publisher, "温度监控器", temperature_callback, &temp_threshold);
    subscribe(sensor_publisher, "系统日志", logger_callback, NULL);
    
    // 创建并发布温度数据
    float temp_data[] = {25.5f, 28.3f, 32.7f};
    
    for (int i = 0; i < 3; i++) {
        Topic topic = {
            .name = "temperature",
            .data = &temp_data[i],
            .data_size = sizeof(float),
            .timestamp = (uint32_t)time(NULL)
        };
        publish(sensor_publisher, &topic);
    }
    
    // 取消订阅
    unsubscribe(sensor_publisher, "系统日志");
    
    // 再次发布数据
    float final_temp = 35.2f;
    Topic topic = {
        .name = "temperature",
        .data = &final_temp,
        .data_size = sizeof(float),
        .timestamp = (uint32_t)time(NULL)
    };
    publish(sensor_publisher, &topic);
    
    // 清理资源
    destroy_publisher(sensor_publisher);
    
    return 0;
}

5. 代码分析

5.1 关键设计点

1. 发布者管理订阅者列表
2. 回调机制实现通知
3. 主题数据封装
4. 动态订阅管理

5.2 实现特点

1. 函数指针实现回调
2. 支持用户数据传递
3. 订阅者管理完善
4. 资源管理安全

6. 编译和运行

gcc -c pubsub.c -o pubsub.o
gcc -c main.c -o main.o
gcc pubsub.o main.o -o pubsub_demo

7. 注意事项

1. 订阅者数量限制
2. 内存管理安全
3. 回调函数异常处理
4. 线程安全考虑

8. 改进建议

1. 添加主题过滤
2. 实现异步通知
3. 支持优先级订阅
4. 添加订阅者分组

9. 总结

发布订阅模式通过解耦发布者和订阅者，实现了灵活的消息通知机制。这种模式特别适合处理事件驱动的场景。

参考资料

1. 《设计模式：可复用面向对象软件的基础》
2. 《C语言程序设计》
3. 《事件驱动编程》