Python多进程Pipe如何清空管道并接收最新发送的数据？

你的需求是让发送端每秒持续发送数据，接收端每4秒读取一次，且每次只获取管道中最新的那条数据，清空之前堆积的旧数据。

修改后的代码 #

from time import sleep
from multiprocessing import Process
from multiprocessing import Pipe

# 发送端函数
def sender(connection):
    print('Sender: Running', flush=True)
    # 生成1000条数据
    for i in range(1000):
        value = i + 1
        sleep(1)
        connection.send(value)
        print("Sender sent ", value)
    # 发送结束信号
    connection.send(None)
    print('Sender: Done', flush=True)

# 接收端函数
def receiver(connection):
    print('Receiver: Running', flush=True)
    while True:
        latest_item = None
        # 循环清空管道，只保留最后一条数据
        while connection.poll():
            latest_item = connection.recv()
            # 如果收到结束信号，直接处理
            if latest_item is None:
                break
        # 检查是否是结束信号
        if latest_item is None:
            print(f'>receiver got {latest_item}', flush=True)
            break
        # 输出最新数据
        print(f'>receiver got {latest_item}', flush=True)
        sleep(4)
    print('Receiver: Done', flush=True)

if __name__ == '__main__':
    # 创建管道
    conn1, conn2 = Pipe()
    # 启动发送进程
    sender_process = Process(target=sender, args=(conn2,))
    sender_process.start()
    # 启动接收进程
    receiver_process = Process(target=receiver, args=(conn1,))
    receiver_process.start()
    # 等待进程结束
    sender_process.join()
    receiver_process.join()

关键修改说明 #

1. 在接收端的循环中，新增了while connection.poll()的循环：
   • connection.poll()会无阻塞检查管道中是否有可读取的数据
   • 循环读取所有堆积的数据，每次更新latest_item，最终保留最后一条有效数据
2. 单独处理结束信号None，确保收到后能正确退出循环，避免遗漏结束逻辑

运行效果 #

修改后会得到你期望的输出：

Sender: Running
Receiver: Running
Sender sent  1
>receiver got 1
Sender sent  2
Sender sent  3
Sender sent  4
>receiver got 4
Sender sent  5
Sender sent  6
Sender sent  7
Sender sent  8
>receiver got 8
Sender sent  9
Sender sent  10
Sender sent  11
Sender sent  12
>receiver got 12
...

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