Python：数据管道无数据传入时如何退出循环？

我需要编写Python 2.7脚本，打印CAN Bus数据管道/var/run/regen/can0/rx的前10行数据。现有代码在管道有数据时运行正常，但无数据时会卡在循环中无法退出；尝试用定时器实现退出也没成功，请问该如何检测管道是否有数据传入？

现有代码： #

#!/usr/bin/python2.7
import os
import re
with open('/var/run/regen/can0/rx', 'r') as pse:  #/var/run/regen/can0/rx is the data pipe
        count = 0
        for line in pse:
            if " " in line:
                break          #want to exit the loop when there is no data
            else:
                while count < 10:
                    for line in pse:
                        words = list(filter(None, re.split(r"[()\[\]\s]\s*", line)))
                        print(words)
                        count += 1
                        break
            break

尝试的定时器代码： #

#!/usr/bin/python2.7
import os
import time
t_end = time.time() + 0.2
while time.time() < t_end:
    with open('/var/run/regen/can0/rx', 'r') as pse:
    # --- remaining code ----

解决方案： #

管道（FIFO）的核心特性是：以只读模式打开时，若没有写端进程打开，open()会直接返回；但如果有写端打开却无数据，读取操作会一直阻塞。要解决阻塞退出问题，推荐用select监听管道的可读状态并设置超时，这是最可靠的方法。

示例代码：

#!/usr/bin/python2.7
import os
import re
import select

PIPE_PATH = '/var/run/regen/can0/rx'
MAX_LINES = 10
TIMEOUT = 0.5  # 无数据时等待超时时间，单位秒

count = 0
with open(PIPE_PATH, 'r') as pse:
    while count < MAX_LINES:
        # 监听管道是否可读，超时返回空列表
        ready, _, _ = select.select([pse], [], [], TIMEOUT)
        if ready:
            line = pse.readline()
            if not line:
                # 管道写端已关闭，无更多数据
                break
            # 处理数据
            words = list(filter(None, re.split(r"[()\[\]\s]\s*", line)))
            print(words)
            count += 1
        else:
            # 超时未收到数据，退出循环
            break

代码说明：

1. select.select([pse], [], [], TIMEOUT)：监听管道文件对象的可读事件，设置超时时间。如果在超时内有数据，ready列表会包含该文件对象；否则返回空列表。
2. 读取到空行（not line）说明管道的所有写端已关闭，直接退出循环。
3. 逻辑清晰，既保证能读取前10行数据，又能在无数据时及时退出。

原代码问题分析：

• 嵌套循环逻辑混乱，for line in pse会一直阻塞等待数据，没有超时机制；
• 定时器代码把open()放在循环内，每次打开都会重新触发管道的阻塞等待，反而加剧问题。

补充方案：检查管道写端是否存在 #

如果需要提前判断管道是否有数据来源，可以用psutil检查是否有进程以写模式打开该管道：

#!/usr/bin/python2.7
import os
import re
import select
import psutil

PIPE_PATH = '/var/run/regen/can0/rx'
MAX_LINES = 10
TIMEOUT = 0.5

def has_active_writers(pipe):
    """检查管道是否有活跃的写进程"""
    for proc in psutil.process_iter(['pid', 'open_files']):
        try:
            for file in proc.open_files():
                if file.path == pipe and ('w' in file.mode or 'a' in file.mode):
                    return True
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return False

count = 0
with open(PIPE_PATH, 'r') as pse:
    while count < MAX_LINES:
        if not has_active_writers(PIPE_PATH):
            print("无活跃写进程，退出")
            break
        ready, _, _ = select.select([pse], [], [], TIMEOUT)
        if ready:
            line = pse.readline()
            if not line:
                break
            words = list(filter(None, re.split(r"[()\[\]\s]\s*", line)))
            print(words)
            count += 1
        else:
            break

注意：此方法需要安装psutil（pip install psutil），且对权限有要求，部分进程可能无法被遍历。

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