【多进程Topic通信系统设计文档】

多进程Topic通信系统设计文档1. 系统概述1.1 项目背景1.2 系统目标2. 架构设计2.1 整体架构架构图概览核心组件说明1. 客户端层Client Layer2. Topic BrokerTCP通信层3. 队列层Queue Layer4. 处理层Processing Layer5. 结果与配置管理层数据流向设计特点2.2 核心组件2.2.1 SafeQueue安全队列2.2.2 TopicBroker消息代理2.2.3 Workers工作进程2.2.4 ProcessManager进程管理器3. 客户端设计3.1 Data Client数据发布客户端3.2 Config Client配置变更客户端3.3 Result Subscriber结果订阅客户端4. 数据流设计4.1 小数据流4.2 大数据流4.3 配置更新流5. 接口设计5.1 网络协议5.2 消息规范6. 配置管理6.1 系统配置SystemConfig6.2 动态配置Shared Config7. 异常处理策略7.1 队列异常7.2 网络异常7.3 进程异常8. 性能考虑8.1 并发模型8.2 优化策略9. 部署说明9.1 环境要求9.2 启动流程9.3 监控与日志10. 总结代码下载URL1. 系统概述1.1 项目背景本系统是一个基于TCP Socket的多进程发布/订阅消息通信框架实现了进程间的高效数据流转和处理。系统采用生产者-消费者模式通过安全队列机制解耦各组件支持配置动态更新、数据处理和结果反馈。1.2 系统目标实现多进程间的可靠消息通信 支持多种数据类型小数据/大数据的分类处理 提供动态配置更新能力 保证队列操作的安全性防溢出、防阻塞 支持结果的实时订阅和反馈2. 架构设计2.1 整体架构系统采用分层架构设计核心组件包括客户端层、Topic BrokerTCP通信层和数据处理层通过安全队列机制实现解耦和异步通信。架构图概览┌─────────────────────────────────────────────────────────────┐ │ 客户端层 │ ├──────────────┬──────────────┬──────────────────────────────┤ │ data_client │ config_client│ result_subscriber │ └──────┬───────┴──────┬───────┴──────────┬───────────────────┘ │ │ │ └──────────────┼───────────────────┘ ▼ ┌─────────────────────────────────────────────────────────────┐ │ Topic Broker (TCP) │ │ - 订阅管理subscribers dict │ │ - 消息路由按topic分发 │ │ - 结果发布独立线程 │ └──┬────────────┬────────────┬───────────────────────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌──────────┐ │DB Queue │ │Proc Queue│ │Config Q │ └────┬────┘ └────┬────┘ └────┬─────┘ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌──────────┐ │DB Writer│ │Processor│ │Config │ │(TDengine)│ │(×N) │ │Listener │ └─────────┘ └─────────┘ └──────────┘ │ │ └───────────┼──────────────────┐ ▼ ▼ ┌───────────┐ ┌────────────┐ │Result Queue│ │Shared Config│ └─────┬─────┘ └────────────┘ │ ▼ ┌───────────┐ │Broker→订阅者│ └───────────┘核心组件说明1. 客户端层Client Layerdata_client数据生产者负责发送业务数据到系统config_client配置客户端用于动态更新系统配置result_subscriber结果订阅者接收处理结果的实时反馈2. Topic BrokerTCP通信层订阅管理维护订阅者字典subscribers dict记录各topic的订阅者消息路由根据消息topic将数据分发到对应的处理队列结果发布独立线程负责将处理结果推送回订阅者3. 队列层Queue LayerDB Queue数据库写入队列存储需要持久化的数据Proc Queue数据处理队列分发到多个Processor进行并发处理Config Q配置更新队列处理动态配置变更4. 处理层Processing LayerDB Writer数据库写入器将数据写入TDengine时序数据库Processor×N多进程数据处理单元支持水平扩展Config Listener配置监听器实时响应配置变更5. 结果与配置管理层Result Queue结果队列收集各Processor的处理结果Shared Config共享配置确保所有组件配置一致性Broker→订阅者结果推送通道将最终结果返回给订阅者数据流向数据流data_client → Topic Broker → Proc Queue → Processor → Result Queue → Broker→订阅者配置流config_client → Topic Broker → Config Q → Config Listener → Shared Config存储流data_client → Topic Broker → DB Queue → DB Writer → TDengine设计特点解耦设计各组件通过队列解耦提高系统可维护性水平扩展Processor支持多实例部署提升处理能力实时反馈结果订阅机制确保处理结果及时返回配置热更新支持运行时动态调整系统参数安全队列防溢出、防阻塞机制保证系统稳定性2.2 核心组件2.2.1 SafeQueue安全队列基于multiprocessing.Queue的封装提供异常处理和超时机制。特性统一的异常捕获Full/Empty异常超时控制get/put方法日志记录操作成功/失败非阻塞操作支持get_nowait/put_nowait关键方法defput(item,timeoutNone,blockTrue)-booldefget(timeout1.0,defaultNone)-Anydefget_nowait(defaultNone)-Anydefput_nowait(item)-bool2.2.2 TopicBroker消息代理基于TCP Socket的发布/订阅服务器。职责管理客户端连接accept_loop维护订阅关系subscribers字典路由消息到内部队列或外部订阅者结果发布独立线程消息格式{action:subscribe|publish,topic:data|config|result,payload:{...}}Topic路由规则Topic路由目标处理逻辑data (small)DB Queue写入TDenginedata (large)Process Queue处理任务configConfig Queue更新配置result订阅者实时推送2.2.3 Workers工作进程DB Writer Worker从DB Queue读取小数据写入TDengine自动创建子表支持数据持久化Processor Worker从Process Queue读取大数据模拟处理任务随机耗时0.5-2秒读取共享配置快照发送结果到Result QueueConfig Listener Worker从Config Queue读取配置变更更新Shared Config支持动态参数调整2.2.4 ProcessManager进程管理器管理所有进程的生命周期。职责创建和启动所有工作进程管理队列和共享配置处理信号SIGINT/SIGTERM优雅关闭所有组件3. 客户端设计3.1 Data Client数据发布客户端支持发送小数据写入数据库支持发送大数据处理队列支持批量发送混合数据自动订阅result topic接收反馈3.2 Config Client配置变更客户端自定义配置键值对快捷修改batch_size/version查看可配置项列表3.3 Result Subscriber结果订阅客户端实时接收处理结果结构化显示结果信息持续监听模式4. 数据流设计4.1 小数据流Data Client → Broker → DB Queue → DB Writer → TDengine4.2 大数据流Data Client → Broker → Process Queue → Processor → Result Queue → Broker → Subscriber4.3 配置更新流Config Client → Broker → Config Queue → Config Listener → Shared Config5. 接口设计5.1 网络协议传输协议TCP消息格式JSON 换行符分隔端口9999编码UTF-85.2 消息规范订阅消息{action:subscribe,topic:result}发布消息{action:publish,topic:data,payload:{data_type:small,name:test_data,content:{value:test}}}结果消息{topic:result,payload:{worker_id:0,task_name:task_1,status:success,result:processed,process_time:1.23秒,record_count:100,config_snapshot:{...},timestamp:2026-08-02 10:30:00}}6. 配置管理6.1 系统配置SystemConfigdataclassclassSystemConfig:worker_num:int3# 处理器数量broker_host:str127.0.0.1broker_port:int9999td_url:strhttp://127.0.0.1:6041td_user:strroottd_pass:strtaosdatatd_db:strpipe_demoqueue_timeout:float1.0# 队列超时时间max_queue_size:int1000# 最大队列大小6.2 动态配置Shared Configapp_name: 应用名称version: 版本号batch_size: 批处理大小支持动态添加新配置项7. 异常处理策略7.1 队列异常Full异常记录警告日志丢弃数据Empty异常返回默认值继续等待其他异常记录错误日志不影响主流程7.2 网络异常客户端断开移除订阅者清理资源连接拒绝友好提示用户启动Broker发送失败移除断开的客户端7.3 进程异常信号处理优雅关闭所有进程超时强制终止5秒超时后使用terminate()资源清理关闭队列和连接8. 性能考虑8.1 并发模型Broker多线程每个客户端独立线程Workers多进程CPU密集型任务队列进程安全的Queue8.2 优化策略非阻塞I/Oselect.select批量发送支持配置缓存共享字典队列大小限制防溢出9. 部署说明9.1 环境要求Python 3.7TDengine可选用于数据持久化依赖taosrest, multiprocessing9.2 启动流程启动Broker和Workerspython pipe_demon_v1.py启动数据客户端python data_client.py启动配置客户端python config_client.py启动结果订阅者python result_subscriber.py9.3 监控与日志系统运行状态监控队列深度监控处理延迟统计错误日志记录10. 总结本系统通过分层架构设计实现了多进程间的可靠消息通信。核心特点包括解耦设计各组件通过安全队列解耦提高系统可维护性灵活扩展支持水平扩展可根据负载动态调整Processor数量实时反馈完整的发布/订阅机制确保处理结果及时返回配置热更新支持运行时动态调整系统参数异常健壮完善的异常处理机制保证系统稳定性系统适用于需要高并发、低延迟、可扩展的数据处理场景特别适合物联网数据采集、实时计算、分布式任务处理等应用场景。代码下载URL多进程Topic通信系统设计