如何实现Simulink/MATLAB与Visual Components中Python API的稳定TCP/IP实时通信？

阿华AIGC实验室

2026-5-15

解决Simulink与Visual Components Python API的稳定实时TCP连接问题

看起来你在Simulink和Visual Components（VC）的TCP连接上踩了几个典型的坑——连接超时、数据批量涌来、端口不匹配，这些其实都是TCP角色错位和仿真时序不同步导致的，我来给你一步步解决：

1. 修正TCP角色与端口问题，解决连接超时

你当前的VC代码是作为TCP服务端监听5005端口，但如果Simulink的TCP/IP Receive模块也设为Server模式，就会出现双方都在等对方发起连接的死锁，这就是打开Receive模块时连接超时的核心原因。

正确角色分配：

让Simulink做TCP客户端，主动连接VC的服务端
确保VC的监听端口（比如你用的5005）和Simulink的连接端口完全一致，同时排查端口是否被其他进程占用：
- Windows：netstat -ano | findstr :5005
- Linux/macOS：lsof -i :5005

优化VC服务端代码，提升稳定性

原来的代码超时设置和循环逻辑没和VC仿真步长绑定，容易卡死或超时。修改后让TCP操作和VC仿真帧同步：

#!/usr/bin/env python
from vcScript import *
import socket

# 全局连接对象
conn = None

def createTCPIPServer(TCP_IP, TCP_PORT):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 允许端口快速复用，避免重启时端口占用
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind((TCP_IP, TCP_PORT))
    s.listen(1)
    print(f"Waiting for Simulink connection on {TCP_IP}:{TCP_PORT}...")
    conn, addr = s.accept()
    print('Connected to:', addr)
    # 设置非阻塞模式，避免VC仿真被TCP阻塞卡死
    conn.setblocking(0)
    return conn

def readTCP(conn):
    BUFFER_SIZE = 16  # 根据你的单步数据大小调整，别设成1（容易截断）
    try:
        data = conn.recv(BUFFER_SIZE)
        return data.strip() if data else None
    except BlockingIOError:
        # 无数据时的正常异常，直接返回None不中断仿真
        return None

def closeTCP(conn):
    if conn:
        print("Closing connection...")
        conn.close()
        print("Connection closed")

def OnSimulationStart():
    global conn
    # 仿真启动时初始化服务端
    conn = createTCPIPServer('127.0.0.1', 5005)

def OnSimulationStep():
    global conn
    if conn:
        data = readTCP(conn)
        if data:
            print(f"Received step data: {data}")
            # 在这里添加你的VC模型逻辑，比如更新部件位置/状态
            # 示例：component = getComponent("YourRobot")
            # component.setPosition(...)
            # 可选：向Simulink发送确认信号，确保步长同步
            conn.sendall(b"ACK")
        elif data is None:
            # 无数据，继续下一个仿真步
            pass
        else:
            # 连接断开，清理资源
            closeTCP(conn)
            conn = None

def OnSimulationStop():
    global conn
    closeTCP(conn)

2. 配置Simulink实现分步实时发送

要解决数据批量接收的问题，必须让Simulink的发送节奏和仿真步长绑定：

使用TCP/IP Send模块，设置为Client模式，目标IP填127.0.0.1，端口5005
启用Simulink的Fixed-Step Solver（固定步长求解器），步长设置和VC的仿真步长保持一致（VC默认是0.01秒，可在VC仿真设置中调整）
确保数据生成模块（比如From Workspace、Constant）是按步长离散输出，而非连续输出
保留TCP/IP Receive模块，同样设为Client模式，用来接收VC的ACK信号，确保每一步数据发送完成后再进入下一步，避免数据堆积

3. 额外排查连接中断的技巧

避免阻塞调用：VC的Python脚本是单线程的，必须用非阻塞TCP模式（setblocking(0)），否则会卡住整个仿真导致连接超时
增加心跳机制：如果长时间无数据传输，TCP可能被系统断开，可以每10个仿真步发送一次心跳包（比如conn.sendall(b"PING")），Simulink收到后回复PONG维持连接
异常捕获：在TCP操作中捕获ConnectionResetError、OSError等异常，连接异常时自动重新初始化服务端，避免脚本崩溃