C#与Python互操作实战：如何安全高效调用Python EXE工具

1次阅读

共计 3341 个字符，预计需要花费 9 分钟才能阅读完成。

最近在开发一个电商数据分析系统时，遇到了一个典型场景：团队用 Python 开发了一套基于 Pandas 的商品推荐模型，但后端服务是用 C# 编写的 ASP.NET Core 应用。需要将 Python 模型集成到 C# 服务中，每天定时处理百万级用户行为数据。

先看三种主流实现方式的对比：

方案	优点	缺点
Process 调用 EXE	隔离性好，支持任意 Python 版本	启动开销大，IPC 成本高
IronPython 引擎	内存内直接执行，无进程开销	仅支持 Python 2.7，不兼容 C 扩展
Python.NET	原生互操作，性能最佳	需要匹配 Python 和.NET 运行时版本

/// <summary>
/// 执行 Python 脚本并获取输出
/// </summary>
/// <param name="pythonExePath">Python 解释器路径 </param>
/// <param name="scriptPath"> 脚本路径 </param>
/// <param name="arguments"> 命令行参数 </param>
/// <param name="timeoutMs"> 超时时间 (毫秒)</param>
public static (string Output, string Error) ExecutePythonScript(
    string pythonExePath, 
    string scriptPath, 
    string arguments = "", 
    int timeoutMs = 30000)
{using var process = new Process();
    process.StartInfo = new ProcessStartInfo
    {
        FileName = pythonExePath,
        Arguments = $"\"{scriptPath}\"{arguments}",
        RedirectStandardOutput = true,
        RedirectStandardError = true,
        UseShellExecute = false,
        CreateNoWindow = true
    };

    var outputBuilder = new StringBuilder();
    var errorBuilder = new StringBuilder();

    process.OutputDataReceived += (_, e) => {if (e.Data != null) outputBuilder.AppendLine(e.Data);
    };
    process.ErrorDataReceived += (_, e) => {if (e.Data != null) errorBuilder.AppendLine(e.Data);
    };

    try
    {process.Start();
        process.BeginOutputReadLine();
        process.BeginErrorReadLine();

        if (!process.WaitForExit(timeoutMs))
        {process.Kill();
            throw new TimeoutException($"Python 执行超时 ({timeoutMs}ms)");
        }

        return (outputBuilder.ToString(), errorBuilder.ToString());
    }
    finally 
    {if (!process.HasExited)
            process.Kill();}
}

推荐使用 argparse 标准库处理命令行参数：

# model_handler.py
import argparse
import json
import sys

def predict(input_data):
    # 模型预测逻辑
    return {"result": "sample_output"}

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    args = parser.parse_args()

    try:
        data = json.loads(args.input)
        result = predict(data)
        print(json.dumps(result))
    except Exception as e:
        print(str(e), file=sys.stderr)
        sys.exit(1)

使用单独的低权限账号运行 Python 进程
通过 Windows Job Object 限制资源使用量

// 创建 Job 对象限制 CPU 和内存
var job = new JobObject();
job.SetLimitCpuPercentage(50);
job.SetLimitMemoryBytes(1024 * 1024 * 500); // 500MB
job.AssignProcess(process.Handle);

永远不要拼接字符串构造命令行
使用 JSON Base64 编码传递复杂参数

var input = new {UserId = 123, Items = new[] {"A", "B"} };
var base64Input = Convert.ToBase64String(Encoding.UTF8.GetBytes(JsonSerializer.Serialize(input)));

// Python 侧解码:
// import base64; data = json.loads(base64.b64decode(args.input))

确保 Process 对象始终被 Dispose
使用 using 语句包裹文件流操作
在 finally 块中强制结束残留进程

public class PythonProcessPool : IDisposable
{
    private readonly BlockingCollection<Process> _pool;
    private readonly string _pythonPath;

    public PythonProcessPool(int poolSize, string pythonPath)
    {
        _pythonPath = pythonPath;
        _pool = new BlockingCollection<Process>(poolSize);

        for (int i = 0; i < poolSize; i++)
        {_pool.Add(CreateKeepAliveProcess());
        }
    }

    private Process CreateKeepAliveProcess()
    {var process = new Process();
        // ... 初始化配置
        process.Start();
        return process;
    }

    public Process Acquire() => _pool.Take();

    public void Release(Process process)
    {if (!process.HasExited)
            _pool.Add(process);
    }

    public void Dispose()
    {foreach (var p in _pool)
            p.Kill();}
}