Python 3.11与2.7独立脚本间快速共享角度数据问题求助

针对你用Python3.11 Flask脚本（实时视频生成角度）和Python2.7机器人控制脚本的高频通信需求（每帧调用sendAngles），之前的socket/文件写入卡顿主要是因为同步阻塞或低效的IO操作，下面是几个高效的解决方案：

1. UDP非阻塞Socket（推荐，平衡速度与实现复杂度） #

UDP无连接、低开销，适合实时性优先、允许少量丢帧的场景（机器人角度每帧更新，少量丢帧不影响控制）。关键是用非阻塞模式避免主进程卡顿。

Python3.11 发送端（Flask脚本） #

import socket
import pickle

# 全局初始化一次，不要在sendAngles内重复创建
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
robot_addr = ("127.0.0.1", 9999)  # 机器人脚本的IP和端口
udp_socket.setblocking(False)  # 全局设置非阻塞

def sendAngles(videomodus, liste):
    # 用pickle打包数据，指定protocol=2保证Python2/3兼容
    data = pickle.dumps({"mode": videomodus, "angles": liste}, protocol=2)
    try:
        udp_socket.sendto(data, robot_addr)
    except BlockingIOError:
        # 发送缓冲区满时直接跳过，不阻塞视频处理流程
        pass

Python2.7 接收端（机器人脚本） #

import socket
import pickle
import time

udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_socket.bind(("0.0.0.0", 9999))
udp_socket.setblocking(False)

def handle_robot_angles(videomodus, liste):
    # 这里写你的机器人控制逻辑
    pass

while True:
    try:
        data, _ = udp_socket.recvfrom(4096)  # 缓冲区大小根据数据量调整
        angle_data = pickle.loads(data)
        handle_robot_angles(angle_data["mode"], angle_data["angles"])
    except socket.error:
        # 无数据时跳过，避免阻塞循环
        pass
    time.sleep(0.001)  # 降低CPU占用，可根据需求调整

2. 共享内存（本地通信最快方案） #

如果两个脚本在同一台机器上，共享内存是零IO开销的最优解，直接操作内存传递数据，完全不会卡顿。

Python3.11 写入端 #

import mmap
import struct
import os

# 全局初始化共享内存，只执行一次
SHM_NAME = "/robot_angle_shm"
SHM_SIZE = 1024  # 根据数据大小调整（比如每个角度占4字节，计算总长度）

# 创建/打开共享内存
shm_fd = os.open(SHM_NAME, os.O_CREAT | os.O_RDWR)
os.ftruncate(shm_fd, SHM_SIZE)
shm = mmap.mmap(shm_fd, SHM_SIZE, mmap.MAP_SHARED, mmap.PROT_WRITE)

def sendAngles(videomodus, liste):
    # 用struct打包二进制数据，避免Python版本兼容问题
    buffer = bytearray()
    buffer.append(videomodus)  # 用1字节存储模式标记
    buffer.extend(struct.pack("I", len(liste)))  # 用4字节存角度列表长度
    for angle in liste:
        buffer.extend(struct.pack("f", angle))  # 每个角度用单精度浮点（4字节）存储
    # 写入共享内存，覆盖旧数据
    shm.seek(0)
    shm.write(buffer.ljust(SHM_SIZE, b"\x00"))  # 填充到共享内存大小

Python2.7 读取端 #

import mmap
import struct
import os
import time

SHM_NAME = "/robot_angle_shm"
SHM_SIZE = 1024

# 打开共享内存
shm_fd = os.open(SHM_NAME, os.O_RDWR)
shm = mmap.mmap(shm_fd, SHM_SIZE, mmap.MAP_SHARED, mmap.PROT_READ)

def handle_robot_angles(videomodus, liste):
    # 机器人控制逻辑
    pass

while True:
    shm.seek(0)
    buffer = shm.read(SHM_SIZE)
    # 解析二进制数据
    videomodus = ord(buffer[0])
    length = struct.unpack("I", buffer[1:5])[0]
    angles = []
    for i in range(length):
        offset = 5 + i * 4
        angle = struct.unpack("f", buffer[offset:offset+4])[0]
        angles.append(angle)
    handle_robot_angles(videomodus, angles)
    time.sleep(0.001)

3. 异步TCP非阻塞（可靠传输备选） #

如果必须保证每帧数据都不丢失，可改用异步TCP非阻塞模式，避免同步IO导致的卡顿。

Python3.11 发送端 #

import asyncio
import pickle
import threading

tcp_writer = None

async def init_tcp_connection():
    global tcp_writer
    try:
        _, writer = await asyncio.open_connection("127.0.0.1", 8888)
        tcp_writer = writer
    except Exception:
        await asyncio.sleep(1)
        await init_tcp_connection()  # 连接断开时自动重连

async def send_data_async(data):
    global tcp_writer
    if tcp_writer:
        try:
            tcp_writer.write(data)
            await tcp_writer.drain()
        except Exception:
            tcp_writer.close()
            await tcp_writer.wait_closed()
            tcp_writer = None

# 在Flask启动时启动异步连接线程
threading.Thread(target=lambda: asyncio.run(init_tcp_connection()), daemon=True).start()

def sendAngles(videomodus, liste):
    data = pickle.dumps({"mode": videomodus, "angles": liste}, protocol=2)
    asyncio.create_task(send_data_async(data))

Python2.7 接收端 #

import socket
import select
import pickle
import time

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("0.0.0.0", 8888))
server_socket.listen(1)
server_socket.setblocking(False)

inputs = [server_socket]
client_socket = None

def handle_robot_angles(videomodus, liste):
    # 机器人控制逻辑
    pass

while True:
    readable, _, _ = select.select(inputs, [], [], 0.001)
    for sock in readable:
        if sock is server_socket:
            client_socket, _ = server_socket.accept()
            client_socket.setblocking(False)
            inputs.append(client_socket)
        else:
            try:
                data = sock.recv(4096)
                if not data:
                    inputs.remove(sock)
                    sock.close()
                    client_socket = None
                else:
                    angle_data = pickle.loads(data)
                    handle_robot_angles(angle_data["mode"], angle_data["angles"])
            except Exception:
                pass
    time.sleep(0.001)

内容的提问来源于stack exchange，提问作者Alex