Python与C#双向通信：如何基于命名管道实现稳定同步数据传输

命名管道本身完全支持你需要的频繁双向交互场景，当前遇到的问题全部来自实现逻辑设计缺陷：

• 你采用了短连接模式，每次收发消息都新建管道实例、用完即销毁，才导致需要频繁创建服务端/客户端，两端启动不同步就会连接报错崩溃
• 没有使用命名管道的全双工特性，一套已经建立的管道连接本身就支持双向读写，不需要两端各启动一套服务端、客户端
• 用using关键字包裹了管道实例，代码块执行完会自动释放管道资源，自然会出现读取完数据管道就自动关闭的问题

核心设计思路 #

仅保留单侧作为命名管道服务端，另一侧作为客户端，一次连接建立后维持长连接，全程复用该连接做双向读写即可，不需要两边都实现服务端逻辑。

优化要点 #

• 管道实例全局持有，不要用完就释放，单独开启后台线程做阻塞读取，避免卡住主线程
• 定义统一的消息边界，你当前代码用的\n分隔符即可，只要保证业务消息内不会出现该分隔符就能解决粘包问题
• 客户端增加连接重试逻辑，启动后循环尝试连接服务端，直到连接成功，避免服务端未就绪时连接崩溃
• 可增加简单心跳机制，两端定期发送空心跳包，长时间未收到心跳就判定连接断开，自动触发重连逻辑

C#端（作为服务端，维持长连接） #

// 全局持有管道相关实例，不要用using自动释放
private static NamedPipeServerStream _pipeServer;
private static StreamReader _reader;
private static StreamWriter _writer;
private static Thread _readThread;

// 启动服务端，只需调用一次
public static void StartPipeServer(string pipeName = "CSServer")
{
    _pipeServer = new NamedPipeServerStream(pipeName, PipeDirection.InOut, 1, PipeTransmissionMode.Message);
    // 等待Python端连接
    _pipeServer.WaitForConnection();
    _reader = new StreamReader(_pipeServer);
    _writer = new StreamWriter(_pipeServer) { AutoFlush = true };
    
    // 开启独立线程循环读取消息
    _readThread = new Thread(ReadLoop);
    _readThread.IsBackground = true;
    _readThread.Start();
}

// 读取消息的循环逻辑
private static void ReadLoop()
{
    while (_pipeServer.IsConnected)
    {
        try
        {
            var message = _reader.ReadLine();
            if (message == null) break;
            // 此处处理收到的Python端消息，处理完成直接调用SendMessage即可返回结果
            Console.WriteLine($"收到Python端消息：{message}");
        }
        catch
        {
            break;
        }
    }
    // 连接断开后可在此处触发重连逻辑
}

// 发送消息给Python端，随时可调用
public static void SendMessage(string payload)
{
    if (_pipeServer.IsConnected)
    {
        _writer.WriteLine(payload);
    }
}

Python端（作为客户端，维持长连接） #

import win32pipe
import win32file
import threading
import time

class LongConnPipeClient:
    def __init__(self, pipe_name="CSServer"):
        self.pipe_path = r'\\.\pipe\\' + pipe_name
        self.pipe_handle = None
        self.running = False
        self.read_thread = None

    # 启动连接，支持重试
    def connect(self, retry_interval=1):
        while True:
            try:
                self.pipe_handle = win32file.CreateFile(
                    self.pipe_path,
                    win32file.GENERIC_READ | win32file.GENERIC_WRITE,
                    0,
                    None,
                    win32file.OPEN_EXISTING,
                    0,
                    None
                )
                # 设为消息读取模式，和C#端配置对齐
                win32pipe.SetNamedPipeHandleState(
                    self.pipe_handle,
                    win32pipe.PIPE_READMODE_MESSAGE,
                    None,
                    None
                )
                self.running = True
                self.read_thread = threading.Thread(target=self._read_loop, daemon=True)
                self.read_thread.start()
                break
            except Exception as e:
                print(f"连接失败，{retry_interval}秒后重试：{e}")
                time.sleep(retry_interval)

    # 后台读取消息循环
    def _read_loop(self):
        while self.running:
            try:
                err, data = win32file.ReadFile(self.pipe_handle, 4096)
                if err == 0 and data:
                    message = data.decode('utf-8').strip()
                    # 此处处理收到的C#端消息，处理完成直接调用send_message即可返回结果
                    print(f"收到C#端消息：{message}")
            except:
                self.running = False
                break
        # 连接断开后可在此处触发重连逻辑

    # 发送消息给C#端，随时可调用
    def send_message(self, payload):
        if self.running:
            win32file.WriteFile(self.pipe_handle, (payload + '\n').encode('utf-8'))

    def close(self):
        self.running = False
        win32file.CloseHandle(self.pipe_handle)

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