Python脚本如何通过管道传输进程数据

wen python案例 2026-06-27 253

本文目录导读：

subprocess 模块（推荐）
os.popen()（较旧方式）
使用 shell 管道操作符
双向管道通信
实战示例：日志监控系统
数据流处理示例
错误处理最佳实践
重要提示：

Python 提供了多种方式通过管道传输进程数据,主要包括以下几种：

subprocess 模块（推荐）

管道输出到变量

import subprocess
# 获取命令输出
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
print(result.stdout)  # 标准输出
print(result.stderr)  # 错误输出
# 或者使用 PIPE
result = subprocess.run(['grep', 'python'], input='hello\npython\nworld', 
                       capture_output=True, text=True)
print(result.stdout)  # 输出: python

管道连接多个进程

import subprocess
# 方式1：使用 PIPE 连接进程
p1 = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['grep', 'py'], stdin=p1.stdout, stdout=subprocess.PIPE)
p1.stdout.close()  # 允许 p1 接收 SIGPIPE 信号
output = p2.communicate()[0]
print(output.decode())
# 方式2：使用 subprocess.PIPE 和 subprocess.run
result1 = subprocess.run(['echo', 'hello world'], capture_output=True, text=True)
result2 = subprocess.run(['grep', 'hello'], input=result1.stdout, 
                        capture_output=True, text=True)
print(result2.stdout)

os.popen()（较旧方式）

import os
# 读取命令输出
with os.popen('ls -l') as f:
    output = f.read()
    print(output)
# 写入到命令
with os.popen('grep python', 'w') as f:
    f.write('hello python\n')
    f.write('hello world\n')

使用 shell 管道操作符

import subprocess
# 直接在 shell 中使用管道
result = subprocess.run('ls -l | grep py | wc -l', 
                       shell=True, capture_output=True, text=True)
print(f"Python 文件数量: {result.stdout.strip()}")

双向管道通信

import subprocess
import time
# 创建双向管道
proc = subprocess.Popen(['cat'],
                       stdin=subprocess.PIPE,
                       stdout=subprocess.PIPE,
                       stderr=subprocess.PIPE,
                       text=True)
# 发送数据
proc.stdin.write("Hello\n")
proc.stdin.write("World\n")
proc.stdin.flush()
# 接收数据
output = proc.stdout.readline()
print(f"收到: {output.strip()}")
# 关闭并等待
proc.stdin.close()
proc.wait()

实战示例：日志监控系统

import subprocess
import time
import sys
def monitor_logs():
    """监控系统日志并过滤特定关键词"""
    # 模拟 tail -f 命令
    tail_proc = subprocess.Popen(
        ['tail', '-f', '/var/log/syslog'],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        bufsize=1  # 行缓冲
    )
    # 创建 grep 进程过滤关键词
    grep_proc = subprocess.Popen(
        ['grep', '--line-buffered', 'ERROR\|WARNING'],
        stdin=tail_proc.stdout,
        stdout=subprocess.PIPE,
        text=True,
        bufsize=1
    )
    # 关闭 tail 的 stdout 副本
    tail_proc.stdout.close()
    try:
        # 实时读取过滤后的输出
        for line in iter(grep_proc.stdout.readline, ''):
            if line:
                print(f"[{time.strftime('%H:%M:%S')}] {line.strip()}")
    except KeyboardInterrupt:
        print("\n停止监控")
    finally:
        tail_proc.terminate()
        grep_proc.terminate()
# 运行监控（需要权限）
# monitor_logs()

数据流处理示例

import subprocess
import json
def process_data_pipeline():
    """使用管道处理数据流"""
    # 创建数据生成器
    data = [
        {"name": "Alice", "score": 95},
        {"name": "Bob", "score": 82},
        {"name": "Charlie", "score": 78}
    ]
    # 转换数据为 JSON 并通过管道传给 jq 处理
    json_str = '\n'.join(json.dumps(item) for item in data)
    # 使用 jq 过滤数据
    result = subprocess.run(
        ['jq', '.name, .score'],
        input=json_str,
        capture_output=True,
        text=True
    )
    print(result.stdout)
# 运行示例
process_data_pipeline()

错误处理最佳实践

import subprocess
import sys
def safe_pipeline():
    """安全的管道操作"""
    try:
        # 创建进程管道
        p1 = subprocess.Popen(
            ['command1'],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )
        p2 = subprocess.Popen(
            ['command2'],
            stdin=p1.stdout,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )
        # 关闭管道以避免死锁
        p1.stdout.close()
        # 获取输出
        stdout, stderr = p2.communicate()
        # 检查错误
        if p2.returncode != 0:
            print(f"错误: {stderr.decode()}", file=sys.stderr)
            return None
        return stdout.decode()
    except FileNotFoundError:
        print("命令未找到", file=sys.stderr)
        return None
    except Exception as e:
        print(f"管道错误: {e}", file=sys.stderr)
        return None