1. 从单打独斗到协同作战为什么我们需要联合仿真在工程仿真领域尤其是涉及复杂系统比如汽车、航空航天、机器人时我们常常会遇到一个尴尬的局面你手里的“王牌”仿真工具可能只擅长解决整个问题链条中的一环。比如你可能用 Simulink 把控制算法玩得炉火纯青但车辆动力学模型却在 CarSim 里才能得到最精确的模拟或者你的核心算法是用 C/Python 写的高性能计算程序但系统级的逻辑和界面却需要在 Simulink 里搭建和观察。这时候传统的做法要么是把所有模型都“翻译”到同一个平台里费时费力且可能损失精度要么就是分开仿真手动来回倒腾数据不仅效率低下还容易出错更无法实现真正的动态交互。联合仿真Co-Simulation就是为了解决这个痛点而生的。它本质上是一种“专业的人做专业的事然后大家坐下来一起开会”的仿真模式。各个子系统或称为仿真器在各自最擅长的环境中独立运行计算自己的状态但同时通过一个约定的通信机制在特定的时间点上交换数据从而实现整个大系统的协同仿真。想象一下你正在开发一辆自动驾驶汽车。Simulink 里的感知与决策算法需要知道车辆当前的位置和速度来自 CarSim 的动力学模型而 CarSim 需要接收来自 Simulink 的油门、刹车和转向指令。如果两者能实时对话你就能在一个闭环里测试算法的有效性观察车辆在虚拟世界中的真实反应。这就是联合仿真的核心价值在保持各子系统建模独立性和专业性的前提下实现系统级的集成验证与性能评估。要实现这种对话关键就在于“通信”。MATLAB/Simulink 作为工业界广泛使用的平台提供了多种与外部世界交互的接口其中最强大、最灵活的莫过于 S-Function。而要让 S-Function 能与外部应用程序可能是另一个仿真器、一个用 C 写的物理引擎或者一个用 Python 做的 AI 模型稳定、高效地“聊天”我们需要一个可靠、快速且跨语言的通信中间件。ZeroMQ简称 ZMQ正是这样一个“社交达人”它轻量、高速支持多种通信模式完美契合了联合仿真中对通信的苛刻要求。本文将深入探讨如何利用 S-Function 和 ZeroMQ搭建起 Simulink 与外部应用程序之间的通信桥梁手把手带你实现一个可用的联合仿真框架。2. 通信基石ZeroMQ 与 S-Function 的选型与配置在开始敲代码之前我们必须搞清楚手里的“工具”到底能干什么以及为什么选它们。这是避免后续踩坑的关键。2.1 为什么是 ZeroMQ不仅仅是“零延迟”的承诺ZeroMQ 不是一个消息队列服务器而是一个看起来像“智能插座”的并发网络通信库。它封装了底层的网络通信细节如 TCP、IPC提供了套接字风格的 API但比原生套接字抽象层次更高功能更强大。对于联合仿真它有以下几个不可替代的优势无中间件部署简单ZMQ 是库不是服务。你的应用程序链接它就能直接获得通信能力无需额外部署和运维像 RabbitMQ、Kafka 这样的消息代理服务器。这对于仿真环境部署的简洁性至关重要。超高性能其“零拷贝”等设计理念使得它在进程间通信IPC和本地网络通信中延迟极低吞吐量很高能满足仿真中实时或准实时的数据交换需求。灵活的通信模式它提供了请求-应答Req-Rep、发布-订阅Pub-Sub、推-拉Push-Pull等多种模式。在联合仿真中发布-订阅模式尤为常用。Simulink 可以作为发布者Publisher定时向外发布状态量外部程序作为订阅者Subscriber接收这些状态进行计算并将结果作为另一个主题发布回来Simulink 再订阅它。这种模式解耦了通信双方非常灵活。跨语言支持ZMQ 有 C、C、Python、MATLAB通过 MEX等数十种语言的绑定。这意味着你可以用 C 写高性能物理引擎用 Python 做机器学习推理用 MATLAB/Simulink 做控制它们都能通过 ZMQ 无缝对话。传输协议多样支持tcp://、ipc://进程间、inproc://线程间。在联合仿真中如果 Simulink 和外部程序在同一台机器上使用ipc://协议可以获得比 TCP 更低的延迟和更高的吞吐量。注意虽然 ZMQ 性能优异但它不保证消息的绝对可靠传输像 TCP 那样。在 Pub-Sub 模式下如果订阅者启动晚于发布者会丢失一些初始消息。对于仿真我们通常需要确保初始状态同步这需要在应用层设计握手协议来解决。2.2 S-FunctionSimulink 通往外部世界的万能钥匙S-Function系统函数是 Simulink 的扩展机制允许你用 C、C、Fortran、MATLAB 语言编写自定义模块集成到 Simulink 模型中。它就像一个标准的插件接口Simulink 引擎在仿真的每个关键步骤初始化、计算输出、更新状态、计算导数、终止都会调用你编写的回调函数。对于联合仿真通信我们选择用 C/C 编写 S-Function主要原因有三性能C/C 直接操作内存和网络效率最高能与 ZMQ 的 C 语言 API 无缝结合。控制力可以精细控制内存分配、网络连接的生命周期避免 MATLAB MEX 层面可能带来的额外开销和不确定性。兼容性生成的代码更容易被其他 C/C 项目复用也便于最终部署到实时系统或嵌入式设备上。一个基本的 C MEX S-Function 需要实现几个核心的回调函数mdlInitializeSizes: 定义模块的输入端口数、输出端口数、状态数、采样时间等基本信息。mdlInitializeSampleTimes: 定义模块的采样时间。对于联合仿真这里通常设置为继承自驱动它的信号源或者设置为固定的、与外部程序协调好的步长。mdlOutputs: 在每个采样时刻计算模块的输出。在这里我们会从 ZMQ 套接字读取外部程序发送过来的数据并赋值给输出端口。mdlUpdate: 在每个采样时刻更新模块的内部状态。在这里我们通常会发送Simulink 的当前状态输入端口的数据到外部程序。mdlTerminate: 仿真结束时调用用于安全地关闭 ZMQ 套接字和上下文释放资源。2.3 环境准备编译器与库的配置工欲善其事必先利其器。在 Windows 上使用 MATLAB 编写 C MEX S-Function 并链接第三方库需要正确配置 C/C 编译器。安装编译器MATLAB 通常推荐使用 MinGW-w64。你可以通过执行mex -setup命令按照 MATLAB 的提示下载并安装它。确保安装的版本与你的 MATLAB 版本兼容。对于较新的 MATLAB它可能已内置支持或引导你安装指定的版本。获取 ZeroMQ 库Windows最方便的方式是从 ZeroMQ 官网下载预编译的二进制包如zeromq-4.3.4-x64.zip。解压后你会得到include文件夹包含zmq.h等头文件和lib文件夹包含libzmq.lib等库文件。Linux/macOS通常使用包管理器安装如sudo apt-get install libzmq3-dev(Ubuntu) 或brew install zeromq(macOS)。配置 MATLAB 的 MEX 编译选项我们需要告诉 MATLAB 编译器去哪里找 ZMQ 的头文件和库文件。这通过创建一个mex编译脚本或直接设置环境变量来实现。一个更可靠的方法是在你的 S-Function 源文件所在目录创建一个compile_mex.m脚本% compile_mex.m zmq_include_path D:\Libraries\ZeroMQ\include; % 替换为你的 ZMQ include 路径 zmq_lib_path D:\Libraries\ZeroMQ\lib; % 替换为你的 ZMQ lib 路径 mex_cmd sprintf(... mex -I%s -L%s -lzmq sfun_zmq_comms.c, ... zmq_include_path, zmq_lib_path); eval(mex_cmd); disp(S-Function compiled successfully with ZeroMQ.);运行这个脚本MATLAB 就会编译sfun_zmq_comms.c并链接libzmq.lib生成一个.mexw64(Windows) 或.mexa64(Linux) 等后缀的文件这就是可以被 Simulink 直接调用的 S-Function 二进制模块。3. 构建通信桥梁S-Function 与 ZMQ 的集成实现理论准备就绪现在我们来搭建这座桥。我们将实现一个双向通信的 S-Function它从 Simulink 模型接收数据通过 ZMQ 发送给外部程序同时从 ZMQ 接收外部程序的计算结果输出给 Simulink。3.1 定义 S-Function 的接口与数据结构首先在 S-Function 的源文件头部包含必要的头文件并定义我们用于存储 ZMQ 上下文和套接字的结构体。这个结构体将作为 S-Function 的“用户数据”void *userData在回调函数之间传递。/* sfun_zmq_comms.c */ #define S_FUNCTION_NAME sfun_zmq_comms #define S_FUNCTION_LEVEL 2 #include simstruc.h #include zmq.h #include string.h #include stdio.h /* 定义我们的持久化数据结构 */ typedef struct { void* zmq_context; void* zmq_pub_socket; /* 用于发布Simulink数据的套接字 */ void* zmq_sub_socket; /* 用于订阅外部数据的套接字 */ char pub_endpoint[256]; /* 发布地址如 tcp://*:5555 */ char sub_endpoint[256]; /* 订阅地址如 tcp://localhost:5556 */ int is_initialized; /* 标志位防止重复初始化 */ } ZMQCommsData;这里我们计划使用两个套接字一个**发布PUB套接字用于向外发送 Simulink 的模型状态例如控制指令、参考信号一个订阅SUB**套接字用于接收外部程序的计算结果例如被控对象的反馈状态。使用两个独立的套接字和端口可以使数据流更加清晰避免自环等复杂情况。3.2 实现核心回调函数接下来我们逐一实现 S-Function 的关键回调函数。mdlInitializeSizes函数定义了模块的基本属性。static void mdlInitializeSizes(SimStruct *S) { ssSetNumSFcnParams(S, 2); /* 我们有两个可配置参数发布地址和订阅地址 */ if (ssGetNumSFcnParams(S) ! ssGetSFcnParamsCount(S)) { return; /* 参数数量不匹配Simulink会报错 */ } ssSetNumContStates(S, 0); ssSetNumDiscStates(S, 0); /* 定义输入端口接收来自Simulink模型要发送出去的数据 */ if (!ssSetNumInputPorts(S, 1)) return; ssSetInputPortWidth(S, 0, DYNAMICALLY_SIZED); /* 宽度由模型连接决定 */ ssSetInputPortDirectFeedThrough(S, 0, 1); /* 输入直接馈通因为输出可能依赖于当前输入 */ /* 定义输出端口将接收到的外部数据输出给Simulink模型 */ if (!ssSetNumOutputPorts(S, 1)) return; ssSetOutputPortWidth(S, 0, DYNAMICALLY_SIZED); /* 宽度也动态决定需与外部程序约定 */ ssSetNumSampleTimes(S, 1); /* 单个采样时间 */ ssSetNumRWork(S, 0); ssSetNumIWork(S, 0); ssSetNumPWork(S, 1); /* 保留一个指针工作向量用于存储我们的 ZMQCommsData 结构体指针 */ ssSetNumModes(S, 0); ssSetNumNonsampledZCs(S, 0); ssSetOptions(S, SS_OPTION_EXCEPTION_FREE_CODE); }mdlInitializeSampleTimes设置采样时间。在联合仿真中这个时间步长至关重要它决定了 Simulink 与外部程序交换数据的频率。通常设置为固定步长并与外部程序的仿真步长保持一致。static void mdlInitializeSampleTimes(SimStruct *S) { /* 设置为固定步长例如 0.001秒 (1kHz) */ ssSetSampleTime(S, 0, 0.001); ssSetOffsetTime(S, 0, 0.0); }mdlStart函数在仿真开始时调用是我们初始化 ZMQ 上下文和套接字的理想位置。#define MDL_START static void mdlStart(SimStruct *S) { ZMQCommsData* comms_data; const mxArray* param_ptr; char* endpoint_buf; size_t str_len; int rc; /* 为我们的数据结构分配内存 */ comms_data (ZMQCommsData*)malloc(sizeof(ZMQCommsData)); if (comms_data NULL) { ssSetErrorStatus(S, Failed to allocate memory for ZMQ data); return; } memset(comms_data, 0, sizeof(ZMQCommsData)); /* 从模块参数获取发布和订阅地址 */ param_ptr ssGetSFcnParam(S, 0); /* 第一个参数发布地址 */ endpoint_buf mxArrayToString(param_ptr); if (endpoint_buf) { strncpy(comms_data-pub_endpoint, endpoint_buf, sizeof(comms_data-pub_endpoint)-1); mxFree(endpoint_buf); } param_ptr ssGetSFcnParam(S, 1); /* 第二个参数订阅地址 */ endpoint_buf mxArrayToString(param_ptr); if (endpoint_buf) { strncpy(comms_data-sub_endpoint, endpoint_buf, sizeof(comms_data-sub_endpoint)-1); mxFree(endpoint_buf); } /* 初始化 ZMQ 上下文 */ comms_data-zmq_context zmq_ctx_new(); if (!comms_data-zmq_context) { ssSetErrorStatus(S, Failed to create ZMQ context); free(comms_data); return; } /* 创建 PUB 套接字并绑定 */ comms_data-zmq_pub_socket zmq_socket(comms_data-zmq_context, ZMQ_PUB); rc zmq_bind(comms_data-zmq_pub_socket, comms_data-pub_endpoint); if (rc ! 0) { ssSetErrorStatus(S, Failed to bind PUB socket); goto cleanup; } /* 创建 SUB 套接字并连接设置订阅过滤器空字符串表示订阅所有消息 */ comms_data-zmq_sub_socket zmq_socket(comms_data-zmq_context, ZMQ_SUB); rc zmq_connect(comms_data-zmq_sub_socket, comms_data-sub_endpoint); if (rc ! 0) { ssSetErrorStatus(S, Failed to connect SUB socket); goto cleanup; } rc zmq_setsockopt(comms_data-zmq_sub_socket, ZMQ_SUBSCRIBE, , 0); if (rc ! 0) { ssSetErrorStatus(S, Failed to set SUB socket subscription); goto cleanup; } /* 可选设置套接字超时防止仿真卡住。例如设置接收超时为100ms */ int timeout 100; zmq_setsockopt(comms_data-zmq_sub_socket, ZMQ_RCVTIMEO, timeout, sizeof(timeout)); comms_data-is_initialized 1; /* 将结构体指针存入 PWork 向量供其他函数使用 */ ssSetPWorkValue(S, 0, comms_data); return; cleanup: if (comms_data-zmq_sub_socket) zmq_close(comms_data-zmq_sub_socket); if (comms_data-zmq_pub_socket) zmq_close(comms_data-zmq_pub_socket); if (comms_data-zmq_context) zmq_ctx_term(comms_data-zmq_context); free(comms_data); ssSetErrorStatus(S, ZMQ initialization failed); }mdlOutputs函数在每个仿真步长计算输出。这里我们尝试从 SUB 套接字接收外部数据并赋值给输出端口。static void mdlOutputs(SimStruct *S, int_T tid) { ZMQCommsData* comms_data (ZMQCommsData*)ssGetPWorkValue(S, 0); real_T* y ssGetOutputPortRealSignal(S, 0); int_T width ssGetOutputPortWidth(S, 0); zmq_msg_t msg; int rc; size_t data_size; if (!comms_data || !comms_data-is_initialized) { /* 未初始化输出零或默认值 */ for (int i 0; i width; i) y[i] 0.0; return; } /* 尝试接收消息 */ zmq_msg_init(msg); rc zmq_msg_recv(msg, comms_data-zmq_sub_socket, ZMQ_DONTWAIT); /* 非阻塞接收 */ if (rc 0) { /* 成功接收到消息 */ data_size zmq_msg_size(msg); /* 假设接收到的数据是 double 数组且大小与输出端口匹配 */ if (data_size width * sizeof(real_T)) { memcpy(y, zmq_msg_data(msg), data_size); } else { /* 大小不匹配可能是协议错误这里可以记录或报错 */ /* 为了鲁棒性我们只拷贝安全的部分 */ size_t copy_size (data_size width * sizeof(real_T)) ? data_size : width * sizeof(real_T); memcpy(y, zmq_msg_data(msg), copy_size); } } else if (rc -1 errno EAGAIN) { /* 没有新消息保持上一次的输出值或者输出默认值 在联合仿真中通常需要外部程序保持同步所以这里可能是个错误状态。 一个更健壮的做法是在第一次仿真前进行握手确保连接和同步。 这里我们先输出零并在后续讨论同步问题。 */ for (int i 0; i width; i) y[i] 0.0; } else { /* 接收发生其他错误 */ ssSetErrorStatus(S, Error receiving data from ZMQ socket); } zmq_msg_close(msg); }mdlUpdate函数也在每个步长被调用通常用于更新离散状态。我们在这里将 Simulink 的输入数据发送出去。static void mdlUpdate(SimStruct *S, int_T tid) { ZMQCommsData* comms_data (ZMQCommsData*)ssGetPWorkValue(S, 0); const real_T* u ssGetInputPortRealSignal(S, 0); int_T width ssGetInputPortWidth(S, 0); zmq_msg_t msg; int rc; if (!comms_data || !comms_data-is_initialized) { return; } /* 准备要发送的消息 */ rc zmq_msg_init_size(msg, width * sizeof(real_T)); if (rc ! 0) { return; /* 初始化失败 */ } memcpy(zmq_msg_data(msg), u, width * sizeof(real_T)); /* 发送消息 */ rc zmq_msg_send(msg, comms_data-zmq_pub_socket, 0); /* 阻塞发送 */ zmq_msg_close(msg); /* 发送后zmq会接管消息内存但我们仍需调用close */ if (rc -1) { /* 发送失败可以记录日志但通常不终止仿真 */ /* printf(ZMQ send error: %s\n, zmq_strerror(errno)); */ } }最后在mdlTerminate中清理资源。static void mdlTerminate(SimStruct *S) { ZMQCommsData* comms_data (ZMQCommsData*)ssGetPWorkValue(S, 0); if (comms_data) { if (comms_data-zmq_sub_socket) zmq_close(comms_data-zmq_sub_socket); if (comms_data-zmq_pub_socket) zmq_close(comms_data-zmq_pub_socket); if (comms_data-zmq_context) zmq_ctx_term(comms_data-zmq_context); free(comms_data); ssSetPWorkValue(S, 0, NULL); } }别忘了实现mdlCheckParameters等函数来验证参数以及编写sfun_zmq_comms_wrapper.c中要求的其他标准 S-Function 接口函数如mdlSetWorkWidths。完整的代码还需要处理更多的错误检查和边界情况。4. 握手、同步与数据协议让联合仿真稳定可靠有了通信框架下一步是确保两个仿真器能“步调一致”地工作并且能正确理解对方发送的“语言”。这是联合仿真从“能跑通”到“稳定可用”的关键。4.1 初始握手与同步策略直接开始仿真往往会出问题。比如Simulink 的 S-Function 启动了开始发布数据但外部程序可能还没启动或者启动后还没来得及连接和订阅就会丢失最初的几包关键数据如初始状态。因此一个简单的握手协议是必要的。一个常见的做法是使用 ZMQ 的 REQ-REP 模式在仿真开始前进行一次握手Simulink S-Function 在mdlStart中除了创建 PUB/SUB 套接字再创建一个 REP应答套接字绑定到一个固定端口如tcp://*:5557并等待连接。外部程序启动后创建一个 REQ请求套接字连接到 Simulink 的 REP 端口。外部程序发送一个特定的握手消息例如字符串READY。Simulink 的 S-Function 收到READY后回复一个确认消息例如ACK并可以附带一些初始参数如仿真步长、数据维度等。双方收到确认后才正式启动各自的仿真循环。这个握手过程确保了双方在开始交换实时数据前网络连接已经建立并且就基本参数达成一致。在mdlOutputs中我们可以设置一个标志在收到握手确认前输出端口保持为零或初始值避免使用未初始化的数据。4.2 定义清晰的数据交换协议仅仅发送一堆二进制数据是不够的。双方必须对数据的格式、顺序和含义有完全一致的约定。这就是数据协议。数据序列化我们上面例子中直接memcpy了double数组。这只在双方平台字节序Endianness相同、内存对齐方式一致、且real_T在 MATLAB 中就是double的情况下才有效。更稳健的做法是定义一种平台无关的序列化格式。简单方案约定使用网络字节序大端序。发送前将每个double通过htonl/ntohl用于整数转换类的函数转换或者直接使用text格式发送如 CSV 字符串但效率较低。高级方案使用专业的序列化库如Protocol Buffers、FlatBuffers或MessagePack。它们能自动处理字节序、版本兼容并生成多语言代码是大型项目的首选。但这会引入额外的依赖和复杂度。消息结构一个消息除了负载数据最好包含一个“信封”。主题Topic在 Pub-Sub 模式中非常有用。例如Simulink 发布control/steering和control/throttle两个主题外部程序可以按需订阅。这增加了灵活性。时间戳包含仿真时间戳有助于接收方处理可能的延迟、丢包或进行数据对齐。序列号用于检测丢包。数据维度/类型描述负载数据的结构。一个简单的二进制消息结构可以设计为[消息头 (16字节)][负载数据] 消息头 [主题ID (4字节)][时间戳 (8字节)][数据长度 (4字节)]接收方先读取并解析头部再根据“数据长度”读取负载。4.3 仿真步长同步与实时性问题联合仿真中最棘手的问题之一是时间同步。Simulink 以固定的步长如 1ms推进仿真。外部程序可能以不同的频率运行例如一个物理引擎也以 1ms 运行但一个图像渲染器可能以 60Hz 运行。锁步同步Lock-Step这是最严格的方式。双方约定一个主时钟通常是 Simulink。Simulink 在一个仿真步长内完成计算并发送数据后等待收到外部程序针对此步长的计算结果然后才推进到下一个步长。这保证了数据的严格因果性和同步性但速度受限于最慢的一方且任何一方的卡顿都会导致整个仿真暂停。可以通过在mdlOutputs中使用阻塞接收去掉ZMQ_DONTWAIT标志并设置合理超时来实现但超时后的处理逻辑重试、终止、使用旧值需要仔细设计。异步通信双方以各自最快的速度运行和通信不互相等待。这能最大化利用计算资源但会引入时间延迟和数据不一致的风险。例如Simulink 在 t1.0s 时发出的控制指令外部程序可能在 t1.002s 才收到并开始计算返回的结果对应的是 t1.002s 的状态而 Simulink 此时可能已经计算到 t1.001s 了。这对于快速动态系统可能是致命的。折中方案带插值的异步通信。双方仍然异步运行但交换的数据包都带有精确的时间戳。接收方如 Simulink不是直接使用最新收到的数据而是根据自己当前的仿真时间对收到的历史数据进行插值如线性插值得到一个“估计”的当前时刻的值。这需要缓冲区来存储历史数据并增加了算法的复杂性但能有效缓解延迟带来的误差是许多高保真联合仿真工具如 FMI/FMU采用的策略。对于大多数工程应用如果仿真步长一致且网络延迟远小于步长例如步长10ms延迟1ms采用简单的准同步方式往往就够了双方按固定步长运行在每一步完成计算后立即发送数据并非阻塞地尝试接收对方的数据。如果收到新数据就使用没收到就使用上一时刻的数据或零阶保持。这需要在mdlOutputs中实现一个简单的数据缓存机制。5. 实战演练搭建一个 Simulink 与 Python 的简单联合仿真让我们用一个具体的例子把上面的理论串联起来。我们将实现一个经典的控制仿真Simulink 作为控制器计算 PID 控制律一个 Python 程序作为被控对象模拟一个简单的质量-弹簧-阻尼系统。5.1 Python 被控对象仿真程序首先我们用 Python 和 ZMQ 写一个简单的被控对象服务器。它订阅 Simulink 发来的控制力u根据动力学方程计算物体的位置和速度然后将状态发布回去。# plant_simulator.py import zmq import numpy as np import time # 物理参数 m 1.0 # 质量 (kg) c 0.5 # 阻尼系数 (N·s/m) k 10.0 # 弹簧系数 (N/m) dt 0.001 # 仿真步长 (s) # 初始化状态 x 0.0 # 位置 (m) v 0.0 # 速度 (m/s) context zmq.Context() # 创建 SUB 套接字订阅来自 Simulink 的控制指令 sub_socket context.socket(zmq.SUB) sub_socket.connect(tcp://localhost:5555) # 连接 Simulink 的 PUB 端口 sub_socket.setsockopt_string(zmq.SUBSCRIBE, ) # 订阅所有消息 # 创建 PUB 套接字发布状态给 Simulink pub_socket context.socket(zmq.PUB) pub_socket.bind(tcp://*:5556) # 绑定到 Simulink SUB 连接的端口 print(Plant simulator started. Waiting for control input...) sim_time 0.0 try: while True: # 非阻塞接收控制指令 try: # 假设接收到的数据是 8 字节的 double (一个控制力 u) data sub_socket.recv(zmq.NOBLOCK) u np.frombuffer(data, dtypenp.float64)[0] except zmq.Again: # 没有新指令使用上一次的 u (这里初始为0) u 0.0 # 基于当前状态和控制力计算加速度 (动力学方程: m*a c*v k*x u) a (u - c*v - k*x) / m # 前向欧拉积分 (简单演示实际可用更精确的积分器) v_new v a * dt x_new x v * dt # 或用 (v v_new)/2 * dt 更精确 x, v x_new, v_new sim_time dt # 准备要发送的状态数据 [位置, 速度] state_data np.array([x, v], dtypenp.float64).tobytes() # 发布状态 pub_socket.send(state_data) # 为了模拟真实计算耗时可以添加微小延时 # time.sleep(dt * 0.5) # 模拟比实时稍快的计算 except KeyboardInterrupt: print(\nSimulation stopped by user.) finally: sub_socket.close() pub_socket.close() context.term()5.2 Simulink 控制器模型与 S-Function 配置在 Simulink 中我们搭建一个简单的 PID 控制器模型。创建一个Constant模块作为目标位置例如设为 1.0。创建一个PID Controller模块调整 P、I、D 参数例如P50, I1, D5。我们需要一个模块来代表被控对象。这里就使用我们刚刚编写的 S-Function。在模型中拖入一个S-Function模块。双击打开参数设置将S-function name设置为sfun_zmq_comms我们编译后的 MEX 文件名。在S-function parameters中设置两个参数参数1 (发布地址):tcp://*:5555(Python 程序订阅的地址)参数2 (订阅地址):tcp://localhost:5556(Python 程序发布的地址)这个 S-Function 的输入端口将接收 PID 控制器计算出的控制力u。这个 S-Function 的输出端口将输出从 Python 程序接收到的物体状态[位置, 速度]。用一个Demux模块将 S-Function 输出的二维信号拆分成position和velocity。将position信号反馈给 PID 控制器的输入端与目标位置做差形成闭环。用Scope模块连接position和velocity信号用于观察响应曲线。在运行仿真前确保已经使用前面编写的compile_mex.m脚本成功编译了sfun_zmq_comms.c生成了sfun_zmq_comms.mexw64文件并且该文件位于 MATLAB 当前路径或搜索路径中。先启动 Python 被控对象程序 (python plant_simulator.py)。这样当 Simulink 启动时ZMQ 连接已经可以建立。在 Simulink 中将求解器设置为固定步长步长与代码中的dt(0.001秒) 保持一致。开始仿真。你应该能在 Scope 中看到物体位置逐渐跟踪到目标值 1.0 的曲线。5.3 调试与常见问题排查第一次运行很可能不会一帆风顺。以下是一些常见问题及排查思路S-Function 编译失败错误‘zmq.h’ file not found检查compile_mex.m中的zmq_include_path是否正确指向了 ZMQ 的include文件夹。错误cannot open file ‘libzmq.lib’检查zmq_lib_path是否正确并确认库文件是.lib格式Windows。Linux/macOS 下是-lzmq链接.so或.dylib。错误LNK2019: unresolved external symbol...通常是链接错误确保链接的库版本32/64位与你的 MATLAB 和编译器匹配。MATLAB 默认是 64 位。仿真运行时连接失败Simulink 报错Failed to bind PUB socket检查端口5555是否已被其他程序占用。可以尝试更换端口号。Python 端报错Connection refused确保先启动 Python 程序绑定到5556再启动 Simulink 仿真。因为 Simulink 的 SUB 套接字是去“连接” Python 绑定的地址。没有数据交换Scope 显示为零检查 Simulink S-Function 模块的输入是否确实有信号连接控制力u。在 Python 程序中添加打印语句确认它是否收到了数据以及计算出的状态是否正常。在 S-Function 的mdlOutputs和mdlUpdate函数中添加调试输出使用mexPrintf注意不要在实时线程中频繁调用看是否在执行发送和接收。使用网络调试工具如netstat -an | findstr 5555或tcpdump查看端口是否有数据流量。仿真结果不稳定或发散步长不匹配确认 Simulink 的固定步长与 Python 程序中的dt完全一致。微小的差异会随着时间累积导致相位误差可能引发数值不稳定。数据不同步这可能是最根本的问题。如果 Python 计算太慢Simulink 在下一步已经用旧的状态计算出了新的控制力。考虑在 S-Function 中实现简单的锁步逻辑在mdlOutputs中如果本次没有收到新数据可以不更新输出保持上一次的值实现一个零阶保持器而不是输出零。这需要修改mdlOutputs增加一个静态变量或 DiscState 来存储上一次接收到的数据。动力学方程或控制器参数问题检查 Python 中的物理模型和 Simulink 中的 PID 参数是否合理。可以先在 Simulink 内用标准的 Transfer Fcn 或 State-Space 模块模拟被控对象确保控制器工作正常再切换到 ZMQ 联合仿真。通过这个简单的例子你已经构建了一个可工作的联合仿真原型。在此基础上你可以扩展数据协议添加时间戳、实现更复杂的同步机制、替换更逼真的被控对象模型如 CarSim 接口或者将 Python 端替换为任何其他支持 ZMQ 的语言编写的程序从而构建起强大的跨平台、跨语言的协同仿真系统。