如何通过multiprocessing.Pipe实现两个独立Python脚本通信？

问题背景 #

我知道这是论坛常见问题，但没找到适配自身情况的答案。我需要让两个独立脚本通过multiprocessing.Pipe实现通信。

现有code1.py：

import time
#...

def code_one(pipe):
    for _ in range(100):
        pipe.send("data")
        time.sleep(1)

if __name__ == '__main__':
    code_one(parent_pipe)

code2.py：

import time
#...

def code_two(pipe):
    while True:
        if pipe.poll():
            data = pipe.recv()
            print(data)

if __name__ == '__main__':
    code_two(child_pipe)

我的问题是：需要共享parent_pipe, child_pipe = Pipe()，但不知如何操作。尝试将其放在第三个.py文件中，却无法导入该对象。

补充疑问 #

1. 这种需求在Python中是否可行？
2. 除multiprocessing外，还有哪些可用方法？
3. 或许Pipe不是合适工具，Listener才是？

额外说明 #

我尝试过写一个main.py来启动进程，示例如下：

import multiprocessing

import code1
import code2

parent_pipe, child_pipe= multiprocessing.Pipe()

if __name__ == '__main__':
    p1 = multiprocessing.Process(target=code1.code_one, args=(child_pipe,))
    p2 = multiprocessing.Process(target=code2.code_two, args=(parent_pipe,))

    p1.start()
    p2.start()

但我希望通过命令行参数分别后台运行程序并共享资源，比如：

nohup python3 my_package.py --send &
nohup python3 my_package.py --recieve &

对应的尝试代码：

import multiprocessing
from code1 import code_one
from code2 import code_two

import argparse

if __name__ == '__main__':

    parser = argparse.ArgumentParser()
    parser.add_argument('-s', '--send', action='store_true')
    parser.add_argument('-r', '--recieve', action='store_true')
    args = parser.parse_args()

    # Create a pipe
    parent_pipe, child_pipe = multiprocessing.Pipe()

    # Spawn process for code_one
    p1 = multiprocessing.Process(target=code_one, args=(parent_pipe,))
    if args.send:
        p1.start()

    # Spawn process for code_two
    p2 = multiprocessing.Process(target=code_two, args=(child_pipe,))
    if args.recieve:
        p2.start()

    # Join the processes (optional, based on your use case)
    #p1.join()
    #p2.join()

解决方案 #

疑问解答 #

1. multiprocessing.Pipe无法实现独立脚本通信：Pipe是为同一父进程创建的子进程设计的，管道两端由父进程传递给子进程。而两次独立python命令启动的是完全独立的进程，没有共同父进程（除系统init进程），无法共享Pipe对象——导入第三方文件的管道也不行，因为每个进程会重新执行导入文件，创建全新的管道实例。

2. 替代的跨进程通信方法：

   • Socket套接字：用TCP/UDP实现本地或跨机器通信，本地可使用Unix域套接字提升效率
   • 共享内存：通过multiprocessing.Array/Value或mmap模块传递简单数据
   • 消息队列：使用multiprocessing.Manager管理队列，或第三方工具如Redis队列
   • 文件锁+文件：实现简单但效率低，适合小数据量场景
   • multiprocessing.Listener/Client：基于Socket的封装，专为跨进程通信设计，使用便捷

3. Listener/Client是适配场景的最优选择：针对你需要独立启动两个进程的需求，Listener/Client无需共享父进程，直接通过网络套接字建立连接，完全符合你的使用场景。

具体实现示例 #

以下是基于Listener/Client的完整实现：

1. 修改code1.py（发送端）

import time
from multiprocessing.connection import Client

def code_one():
    # 连接接收端，本地用localhost+端口，也可替换为Unix域套接字路径
    conn = Client(('localhost', 6000))
    try:
        for _ in range(100):
            conn.send("data")
            print("已发送data")
            time.sleep(1)
    finally:
        conn.close()

if __name__ == '__main__':
    code_one()

2. 修改code2.py（接收端）

import time
from multiprocessing.connection import Listener

def code_two():
    # 启动监听，地址需与发送端一致
    listener = Listener(('localhost', 6000))
    conn = listener.accept()
    print("连接已建立")
    try:
        while True:
            if conn.poll():
                data = conn.recv()
                print(f"收到数据：{data}")
            time.sleep(0.1)
    finally:
        conn.close()
        listener.close()

if __name__ == '__main__':
    code_two()

3. 编写my_package.py（命令行入口）

import argparse
import subprocess
import sys

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('-s', '--send', action='store_true')
    parser.add_argument('-r', '--recieve', action='store_true')
    args = parser.parse_args()

    if args.send:
        subprocess.Popen([sys.executable, 'code1.py'])
    if args.recieve:
        subprocess.Popen([sys.executable, 'code2.py'])

if __name__ == '__main__':
    main()

运行方式

先启动接收端：

nohup python3 my_package.py --recieve &

再启动发送端：

nohup python3 my_package.py --send &

其他可选方案 #

如果需要共享复杂数据结构，可使用Redis队列（需先安装pip install redis）：

# code1.py发送端
import time
import redis

r = redis.Redis(host='localhost', port=6379, db=0)

def code_one():
    for _ in range(100):
        r.rpush('my_queue', 'data')
        time.sleep(1)

# code2.py接收端
import time
import redis

r = redis.Redis(host='localhost', port=6379, db=0)

def code_two():
    while True:
        data = r.lpop('my_queue')
        if data:
            print(data.decode())
        time.sleep(0.1)

该方案支持跨机器通信，且进程重启后可保留未处理数据。

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