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

1次阅读

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

最近在开发一个电商数据分析系统时，我们遇到了一个典型需求：C# 后台服务需要调用团队数据科学家编写的 Python 预测模型（打包为 model.exe）。这个模型每天要处理 10 万 + 商品的价格波动预测，涉及以下挑战：

Python 模型依赖复杂的 NumPy/Pandas 生态
需传递包含 200+ 维度的商品特征数据
要求 90% 的调用在 500ms 内完成

方案	优点	缺点	适用场景
Process.Start	环境隔离好，Python 版本自由	启动开销大 (300ms+)	调用预打包的独立工具
IronPython	内存级交互，无进程开销	不支持 Python3 和 C 扩展库	简单脚本交互
Python.NET	原生互操作，性能最佳	需要手动管理 GIL 和内存	高频调用场景

/// <summary>
/// 调用 Python 工具并获取 JSON 格式结果
/// </summary>
/// <param name="exePath">Python 可执行文件路径 </param>
/// <param name="args"> 序列化后的 JSON 参数 </param>
/// <param name="timeoutMs"> 超时时间 (毫秒)</param>
public static (string Output, string Error) CallPythonExe(
    string exePath, 
    string args, 
    int timeoutMs = 5000)
{
    using var process = new Process
    {
        StartInfo = new ProcessStartInfo
        {
            FileName = exePath,
            Arguments = args,
            UseShellExecute = false,
            RedirectStandardOutput = true,
            RedirectStandardError = true,
            CreateNoWindow = true
        }
    };

    var output = new StringBuilder();
    var error = new StringBuilder();

    // 异步读取输出流
    process.OutputDataReceived += (_, e) => {if (!string.IsNullOrEmpty(e.Data)) 
            output.AppendLine(e.Data);
    };

    process.ErrorDataReceived += (_, e) => {if (!string.IsNullOrEmpty(e.Data))
            error.AppendLine(e.Data);
    };

    try
    {process.Start();

        // 必须异步读取否则可能死锁
        process.BeginOutputReadLine();
        process.BeginErrorReadLine();

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

        return (output.ToString(), error.ToString());
    }
    finally
    {
        // 确保资源释放
        process.Close();}
}

# model.py
import argparse
import json
import sys

def predict(data):
    # 模拟预测逻辑
    return {"result": sum(data["features"]) / len(data["features"])}

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", type=str, 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 进程
var startInfo = new ProcessStartInfo
{
    UserName = "pyRunner",
    Password = securePassword,
    LoadUserProfile = false,
    WorkingDirectory = "C:\sandbox"
};

// 使用 JSON 序列化代替字符串拼接
var safeArgs = JsonConvert.SerializeObject(new 
{
    Features = features,
    ModelVersion = "v2.1"
});

// 调用时进行 Base64 编码
CallPythonExe("model.exe", $"--input {Base64Encode(safeArgs)}");

// 使用 using 确保 Process 对象释放
// 在 finally 中强制终止残留进程
foreach (var p in Process.GetProcessesByName("python"))
{try { p.Kill(); }
    catch {/* 记录日志 */}
}

// 使用 BlockingCollection 实现进程池
public class PythonProcessPool : IDisposable
{private BlockingCollection<Process> _pool = new();

    public PythonProcessPool(int size, string exePath)
    {for (int i = 0; i < size; i++)
        {_pool.Add(CreateProcess(exePath));
        }
    }

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

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

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