如何实现C#与Python应用的高效通信（无法修改Python应用）

架构老林

2026-7-12

问题场景与优化方案

问题背景

我有一个C#应用，需要通过执行命令从一个无法修改的Python CLI应用获取信息。最初的实现是为每个命令创建一个Process实例，后来为了简化流程，编写了一个Python Wrapper脚本，让C#仅创建一个Process与之交互，但Wrapper每次执行命令仍会在Python中新建进程返回结果。这种方式虽有提升，但效率仍不理想，希望通过后台常驻进程、第三方工具或库等方式优化通信效率。

现有实现代码

C# 执行代码

using System;
using System.Diagnostics;
using System.Text;
using System.Threading.Tasks;

namespace MainApp
{
    public class CommandManager
    {
        public string GetHelpCommand()
        {
            return "/usr/phyPROGRAM.py -help";
        }
    }

    public class PythonExecutor : IDisposable
    {
        private Process pythonProcess;
        private StringBuilder outputBuilder;
        private TaskCompletionSource<string> tcs;

        public PythonExecutor()
        {
            InitializePythonProcess();
        }

        private void InitializePythonProcess()
        {
            pythonProcess = new Process
            {
                StartInfo = {
                    FileName = "python3",
                    Arguments = "/usr/wrapper.py",
                    UseShellExecute = false,
                    RedirectStandardInput = true,
                    RedirectStandardOutput = true,
                    CreateNoWindow = true
                }
            };

            outputBuilder = new StringBuilder();
            pythonProcess.OutputDataReceived += (sender, e) =>
            {
                if (e.Data == "##END##")
                {
                    tcs.SetResult(outputBuilder.ToString());
                    outputBuilder.Clear();
                }
                else if (e.Data != null)
                {
                    outputBuilder.AppendLine(e.Data);
                }
            };

            pythonProcess.Start();
            pythonProcess.BeginOutputReadLine();
        }

        public Task<string> ExecuteCommand(string command)
        {
            tcs = new TaskCompletionSource<string>();
            pythonProcess.StandardInput.WriteLine(command);
            return tcs.Task;
        }

        public void Dispose()
        {
            pythonProcess.StandardInput.WriteLine("exit");
            pythonProcess.WaitForExit();
            pythonProcess.Dispose();
        }
    }

    class Program
    {
        static async Task Main(string[] args)
        {
            Console.ReadLine();
            CommandManager cmdManager = new CommandManager();

            using PythonExecutor executor = new PythonExecutor();
            string command1 = cmdManager.GetHelpCommand();
            string result1 = await executor.ExecuteCommand(command1);
            Console.WriteLine(result1);
           
            executor.Dispose();
        }
    }
}

Python Wrapper 脚本

import sys
import subprocess

def execute_main_script(command):
    try:
        # 调用目标Python CLI命令并返回输出
        result = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT)
        return result.decode('utf-8')
    except subprocess.CalledProcessError as e:
        return e.output.decode('utf-8')

while True:
    command = sys.stdin.readline().strip()  # 从标准输入读取命令
    if command == "exit":
        break
    else:
        output = execute_main_script(command)
        print(output, end='',flush=True)  # 将输出写入标准输出
        print("##END##",flush=True)

优化方案

1. 改造Wrapper为常驻进程，复用Python CLI实例

核心思路是让Wrapper启动一次目标Python CLI并保持进程常驻，通过标准输入输出和它持续交互，避免每次命令都新建进程。这是实现成本最低、效果最明显的优化方式。

修改后的Python Wrapper示例：

import sys
import subprocess
import shlex

def main():
    # 启动目标Python CLI，保持输入输出流打开
    cli_process = subprocess.Popen(
        ["python3", "/usr/phyPROGRAM.py"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1  # 启用行缓冲，确保实时输出
    )

    def read_until_prompt():
        """读取CLI输出直到出现预设的提示符，判断命令执行完成"""
        output = []
        while True:
            line = cli_process.stdout.readline()
            if not line:
                break
            output.append(line)
            # 替换为目标CLI实际使用的提示符（比如">>> "或自定义标识）
            if line.endswith("CLI_READY>"):
                break
        return ''.join(output)

    # 初始化：读取CLI启动后的初始输出，直到提示符出现
    read_until_prompt()

    while True:
        command = sys.stdin.readline().strip()
        if command == "exit":
            # 发送退出命令给CLI，等待进程终止后退出Wrapper
            cli_process.stdin.write("exit\n")
            cli_process.stdin.flush()
            cli_process.wait()
            break
        elif command:
            # 将命令发送给常驻的CLI进程
            cli_process.stdin.write(f"{command}\n")
            cli_process.stdin.flush()
            # 读取命令执行结果直到提示符
            result = read_until_prompt()
            # 移除结果末尾的提示符，返回给C#
            result = result.rsplit("CLI_READY>", 1)[0]
            print(result, end='', flush=True)
            print("##END##", flush=True)

if __name__ == "__main__":
    main()

注意：需要根据目标Python CLI的实际交互逻辑调整read_until_prompt中的结束判断条件，如果CLI没有固定提示符，可约定命令执行完成后输出特定标记（比如##COMMAND_DONE##）来替代。

对应的C#代码无需大幅修改，保持现有PythonExecutor的通信逻辑即可。

2. 使用命名管道（Named Pipes）实现可靠通信

命名管道比标准输入输出更稳定，适合复杂的跨进程交互场景，能避免输出截断、缓冲异常等问题。

C# 端（命名管道客户端）

using System.IO.Pipes;
using System.Text;
using System.Threading.Tasks;

public class PythonPipeClient : IDisposable
{
    private NamedPipeClientStream _pipeClient;
    private StreamReader _reader;
    private StreamWriter _writer;
    private TaskCompletionSource<string> _tcs;

    public PythonPipeClient()
    {
        // 连接到Python端创建的命名管道
        _pipeClient = new NamedPipeClientStream(".", "PythonCLIPipe", PipeDirection.InOut, PipeOptions.Asynchronous);
        _pipeClient.Connect();
        _reader = new StreamReader(_pipeClient, Encoding.UTF8);
        _writer = new StreamWriter(_pipeClient, Encoding.UTF8) { AutoFlush = true };
        
        // 异步监听管道输出
        _ = ReadPipeOutputAsync();
    }

    private async Task ReadPipeOutputAsync()
    {
        string line;
        var outputBuilder = new StringBuilder();
        while ((line = await _reader.ReadLineAsync()) != null)
        {
            if (line == "##END##")
            {
                _tcs?.SetResult(outputBuilder.ToString());
                outputBuilder.Clear();
            }
            else
            {
                outputBuilder.AppendLine(line);
            }
        }
    }

    public Task<string> ExecuteCommand(string command)
    {
        _tcs = new TaskCompletionSource<string>();
        _writer.WriteLine(command);
        return _tcs.Task;
    }

    public void Dispose()
    {
        _writer.WriteLine("exit");
        _reader.Dispose();
        _writer.Dispose();
        _pipeClient.Dispose();
    }
}

Python 端（命名管道服务端）

import sys
import subprocess
from multiprocessing.connection import Listener

def main():
    # 启动常驻的目标Python CLI进程
    cli_process = subprocess.Popen(
        ["python3", "/usr/phyPROGRAM.py"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1
    )

    def read_until_prompt():
        output = []
        while True:
            line = cli_process.stdout.readline()
            if not line:
                break
            output.append(line)
            if line.endswith("CLI_READY>"):
                break
        return ''.join(output)

    read_until_prompt()

    # 创建命名管道监听连接
    listener = Listener(address=('./pipes', 'PythonCLIPipe'), authkey=b'secret_auth_key')
    conn = listener.accept()

    while True:
        command = conn.recv()
        if command == "exit":
            # 终止CLI进程并关闭管道
            cli_process.stdin.write("exit\n")
            cli_process.stdin.flush()
            cli_process.wait()
            conn.close()
            listener.close()
            break
        else:
            cli_process.stdin.write(f"{command}\n")
            cli_process.stdin.flush()
            result = read_until_prompt()
            result = result.rsplit("CLI_READY>", 1)[0]
            conn.send(result)
            conn.send("##END##")

if __name__ == "__main__":
    main()

3. 使用gRPC实现类型安全的跨语言通信

如果需要长期维护、支持复杂交互或高并发，gRPC是更专业的选择。但需要编写.proto定义文件，生成C#客户端和Python服务端代码，Python服务端作为中间层调用目标CLI。

大致流程：

编写.proto文件，定义调用CLI的服务和方法
使用gRPC工具生成C#客户端和Python服务端代码
Python服务端启动后常驻，接收gRPC请求，调用目标CLI并返回结果
C#客户端通过gRPC调用Python服务端

这种方式实现成本较高，但扩展性和可靠性最好。

4. 其他轻量选项

ZeroMQ：轻量级消息队列，支持多种通信模式，比gRPC简单，比命名管道灵活
共享内存：适合大数据量传输，但实现复杂，需要处理同步和数据序列化问题

总结

优先选择方案1，仅需修改Wrapper脚本即可大幅减少进程创建开销，实现成本最低；如果需要更可靠的通信或复杂场景，再考虑方案2（命名管道）或方案3（gRPC）。

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

如何实现C#与Python应用的高效通信（无法修改Python应用）

问题背景 #

现有实现代码 #

C# 执行代码 #

Python Wrapper 脚本 #

优化方案 #

1. 改造Wrapper为常驻进程，复用Python CLI实例 #

2. 使用命名管道（Named Pipes）实现可靠通信 #