使用子进程实现 C++ 与 Python 交互式控制台
引言
在软件开发中，有时我们需要在C++应用程序中嵌入Python解释器，实现两种语言间的交互操作。本文介绍一种基于 Windows 匿名管道的技术方案，通过重定向标准输入/输出实现 C++ 对 Python 解释器的实时控制。核心代码仅 200 行，却能构建完整的交互式控制台。
一、技术核心：匿名管道与进程通信
Windows 管道（CreatePipe）是进程间通信的关键设施。我们通过以下步骤建立通信通道：
- 创建双向管道
CreatePipe(&hChildStd_OUT_Rd_, &hChildStd_OUT_Wr_, &saAttr, 0);
SetHandleInformation(hChildStd_OUT_Rd_, HANDLE_FLAG_INHERIT, 0);
- 输出管道：子进程（Python）写 → 父进程（C++）读；
- 输入管道：父进程写 → 子进程读；
SetHandleInformation确保父进程独享管道控制权。
- 重定向 Python 的标准流
在STARTUPINFO中配置标准流重定向：
STARTUPINFO siStartInfo;
siStartInfo.hStdInput = hChildStd_IN_Rd_; // Python 从此读命令
siStartInfo.hStdOutput = hChildStd_OUT_Wr_; // Python 向此写结果
siStartInfo.dwFlags |= STARTF_USESTDHANDLES;
二、核心类 ScriptInteractor 剖析
2.1 进程启动与管道绑定
通过 CreateProcess 启动 Python 交互环境：
void start_process() {wchar_t cmdLine[] = L"python.exe -i"; // 启动交互式解释器CreateProcess(NULL, cmdLine, ..., &siStartInfo, &piProcInfo);CloseHandle(hChildStd_OUT_Wr_); // 父进程无需写端CloseHandle(hChildStd_IN_Rd_); // 父进程无需读端
}
关键点：
CREATE_NO_WINDOW隐藏控制台窗口；- 关闭冗余句柄防止资源泄漏。
2.2 异步输出捕获线程
使用独立线程实时捕获Python输出：
void read_output() {while (running_.load()) {PeekNamedPipe(hChildStd_OUT_Rd_, ..., &dwAvailable, ...);if (dwAvailable > 0) {ReadFile(hChildStd_OUT_Rd_, chBuf, ...);std::cout << chBuf; // 实时打印Python输出}std::this_thread::sleep_for(50ms);}
}
PeekNamedPipe非阻塞检查数据，避免线程死锁；- 原子标志 running_实现安全线程退出。
2.3 命令发送与退出控制
void send_script(const std::string& data) {WriteFile(hChildStd_IN_Wr_, data.c_str(), data.size(), ...);
}
void quit() {send_script("exit()\n"); // 发送Python退出指令running_ = false; // 停止输出线程
}
三、主循环：交互式控制台实现
int main() {SetConsoleOutputCP(CP_UTF8); // 支持中文输出ScriptInteractor interactor;while (true) {std::string line;std::getline(std::cin, line);if (line == "exit()") break;interactor.send_script(line + "\n"); // 发送命令}interactor.quit(); // 优雅退出
}
- 用户输入直接转发至 Python；
- 退出时自动清理管道和线程资源。
四、完整源码
ScriptInteractor.hpp
#include <string>
#include <thread>
#include <atomic>
#include <mutex>
#include <iostream>
#include <windows.h>class ScriptInteractor {
public:ScriptInteractor() :hProcess_{}, hChildStd_IN_Rd_{}, hChildStd_IN_Wr_{}, hChildStd_OUT_Rd_{}, hChildStd_OUT_Wr_{} {create_pipes();start_process();// 启动输出读取线程running_.store(true, std::memory_order_release);outputThread_ = std::thread(&ScriptInteractor::read_output, this);}~ScriptInteractor() {// 通知线程退出并等待线程结束running_.store(false, std::memory_order_release);if (outputThread_.joinable()) {outputThread_.join();}// 关闭所有句柄close_handles();}void send_script(const std::string& data) {DWORD dwWritten;WriteFile(hChildStd_IN_Wr_, data.c_str(), static_cast<DWORD>(data.size()), &dwWritten, NULL);}void quit() {running_.store(false, std::memory_order_release);// 发送退出命令send_script("exit()\n");}private:// 创建管道void create_pipes() {SECURITY_ATTRIBUTES saAttr{};saAttr.nLength = sizeof(SECURITY_ATTRIBUTES);saAttr.bInheritHandle = TRUE;saAttr.lpSecurityDescriptor = NULL;// 创建输出管道if (!CreatePipe(&hChildStd_OUT_Rd_, &hChildStd_OUT_Wr_, &saAttr, 0)) {std::cerr << "CreatePipe failed: " << GetLastError() << std::endl;return;}// 确保读取端不被继承if (!SetHandleInformation(hChildStd_OUT_Rd_, HANDLE_FLAG_INHERIT, 0)) {std::cerr << "SetHandleInformation failed: " << GetLastError() << std::endl;return;}// 创建输入管道if (!CreatePipe(&hChildStd_IN_Rd_, &hChildStd_IN_Wr_, &saAttr, 0)) {std::cerr << "CreatePipe failed: " << GetLastError() << std::endl;return;}// 确保写入端不被继承if (!SetHandleInformation(hChildStd_IN_Wr_, HANDLE_FLAG_INHERIT, 0)) {std::cerr << "SetHandleInformation failed: " << GetLastError() << std::endl;return;}}// 启动进程void start_process() {PROCESS_INFORMATION piProcInfo;STARTUPINFO siStartInfo;ZeroMemory(&piProcInfo, sizeof(PROCESS_INFORMATION));ZeroMemory(&siStartInfo, sizeof(STARTUPINFO));siStartInfo.cb = sizeof(STARTUPINFO);siStartInfo.hStdError = hChildStd_OUT_Wr_;siStartInfo.hStdOutput = hChildStd_OUT_Wr_;siStartInfo.hStdInput = hChildStd_IN_Rd_;siStartInfo.dwFlags |= STARTF_USESTDHANDLES;// 创建Python进程wchar_t cmdLine[] = L"python.exe -i";BOOL bSuccess = CreateProcess(NULL, // 应用程序名cmdLine, // 命令行NULL, // 进程安全属性NULL, // 线程安全属性TRUE, // 继承句柄CREATE_NO_WINDOW, // 创建标志NULL, // 环境变量NULL, // 当前目录&siStartInfo, // STARTUPINFO&piProcInfo // PROCESS_INFORMATION);if (!bSuccess) {std::cerr << "CreateProcess failed: " << GetLastError() << std::endl;return;}// 保存进程句柄hProcess_ = piProcInfo.hProcess;// 关闭不需要的句柄CloseHandle(piProcInfo.hThread);CloseHandle(hChildStd_OUT_Wr_);CloseHandle(hChildStd_IN_Rd_);}void read_output() {constexpr int BUFFER_SIZE = 4096;DWORD dwRead;CHAR chBuf[BUFFER_SIZE]{};while (running_.load(std::memory_order_acquire)) {// 检查是否有数据可读DWORD dwAvailable;if (!PeekNamedPipe(hChildStd_OUT_Rd_, NULL, 0, NULL, &dwAvailable, NULL) || dwAvailable == 0) {std::this_thread::sleep_for(std::chrono::milliseconds(50));continue;}// 读取数据BOOL bSuccess = ReadFile(hChildStd_OUT_Rd_, chBuf, BUFFER_SIZE - 1, &dwRead, NULL);if (!bSuccess || dwRead == 0) {if (GetLastError() == ERROR_BROKEN_PIPE) {running_.store(false, std::memory_order_release);break;}continue;}// 处理读取到的数据chBuf[dwRead] = '\0';std::cout << chBuf;}}void close_handles() {if (hProcess_) {CloseHandle(hProcess_);}if (hChildStd_IN_Wr_) {CloseHandle(hChildStd_IN_Wr_);}if (hChildStd_OUT_Rd_) {CloseHandle(hChildStd_OUT_Rd_);}}private:// Windows API句柄HANDLE hProcess_;HANDLE hChildStd_IN_Rd_;HANDLE hChildStd_IN_Wr_;HANDLE hChildStd_OUT_Rd_;HANDLE hChildStd_OUT_Wr_;// 线程管理std::thread outputThread_;std::atomic_bool running_;
};
main.cpp
#include "ScriptInteractor.hpp"int main() {// 设置控制台编码为UTF-8SetConsoleOutputCP(CP_UTF8);ScriptInteractor interactor;std::cout << "Python Interactive Console (Windows API)" << std::endl;std::cout << ">>> ";while (true) {std::string line;std::getline(std::cin, line);if (line == "exit()" || line == "quit()") {break;}// 发送命令给Python进程interactor.send_script(line + "\n");}interactor.quit();return 0;
}
五、跨平台解决方案
5.1 Boost.Process 实现方案
Boost.Process是Boost库中用于跨平台进程管理的组件，提供了统一的API来处理不同操作系统的进程管理差异。
核心实现思路：
#include <boost/process.hpp>
#include <boost/process/async.hpp>
#include <boost/asio.hpp>namespace bp = boost::process;class BoostScriptInteractor {
public:BoostScriptInteractor() : io_context_(), outputThread_(), running_(true) {create_process();start_async_read();}void send_script(const std::string& data) {if (stdin_) {stdin_->write(data.c_str(), data.size());stdin_->flush();}}void quit() {running_ = false;send_script("exit()\n");if (child_) {child_->terminate();child_->wait();}}private:void create_process() {// 创建管道stdin_ = std::make_unique<bp::opstream>();stdout_ = std::make_unique<bp::ipstream>();// 启动Python进程child_ = std::make_unique<bp::child>("python -i", // 命令行bp::std_in < *stdin_,bp::std_out > *stdout_,io_context_);}void start_async_read() {outputThread_ = std::thread([this]() {std::string line;while (running_) {if (stdout_ && std::getline(*stdout_, line)) {std::cout << line << std::endl;} else {std::this_thread::sleep_for(std::chrono::milliseconds(50));}}});}private:boost::asio::io_context io_context_;std::unique_ptr<bp::child> child_;std::unique_ptr<bp::opstream> stdin_;std::unique_ptr<bp::ipstream> stdout_;std::thread outputThread_;std::atomic_bool running_;
};
5.2 Qt QProcess 实现方案
Qt 的 QProcess 类提供了强大的跨平台进程管理能力，特别适合GUI应用程序。
核心实现思路：
#include <QProcess>
#include <QThread>class QtScriptInteractor : public QObject {Q_OBJECT
public:QtScriptInteractor(QObject *parent = nullptr) : QObject(parent) {process_.setProgram("python");process_.setArguments(QStringList() << "-i");// 连接信号槽connect(&process_, &QProcess::readyReadStandardOutput, this, &QtScriptInteractor::onReadyRead);connect(&process_, &QProcess::readyReadStandardError, this, &QtScriptInteractor::onReadyRead);// 启动进程process_.start();process_.waitForStarted();}void send_script(const std::string& data) {if (process_.state() == QProcess::Running) {process_.write(data.c_str(), data.size());}}void quit() {send_script("exit()\n");process_.closeWriteChannel();process_.waitForFinished();}public slots:void onReadyRead() {QByteArray output = process_.readAllStandardOutput();QByteArray error = process_.readAllStandardError();if (!output.isEmpty()) {std::cout << output.constData();}if (!error.isEmpty()) {std::cerr << error.constData();}}private:QProcess process_;
};