使用Python多进程队列实时处理流式数据遇打印问题求助

你的代码存在几个关键问题，导致计算结果无法正常打印：

1. 子进程循环逻辑错误：while not Queue.empty()只会在启动时检查一次队列状态，此时队列还未存入数据，子进程会直接退出，根本不会等待后续新增的数据。
2. 队列参数传递错误：你定义的队列实例是Q，但启动子进程时传入的是Queue（multiprocessing的队列类），这会导致子进程无法访问正确的队列实例。
3. 数据采集与处理的并行逻辑未落地：主进程需要持续运行数据采集逻辑，同时子进程要保持运行状态以实时处理队列数据。

修正后的完整代码 #

import multiprocessing as mp
import datetime as dt
import time

def process_queue_data(queue):
    # 无限循环，持续监听队列，get()方法会阻塞直到有数据进入
    while True:
        try:
            data = queue.get()
            # 替换为你的实际计算逻辑，这里用示例计算展示
            result_of_computation = f"[{dt.datetime.now()}] 处理数据：{data} | 示例计算结果：数据键值数量={len(data)}"
            print(result_of_computation)
        except Exception as e:
            print(f"数据处理出错：{e}")
            break

def collect_live_data(queue):
    # 模拟实时数据采集逻辑：每1秒生成一组数据存入队列
    count = 0
    while True:
        time.sleep(1)
        # 替换为你从网站获取的真实数据
        live_data = {
            "timestamp": dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "data_id": count,
            "content": "实时采集的网站数据"
        }
        queue.put(live_data)
        print(f"[{dt.datetime.now()}] 已存入队列：{live_data}")
        count += 1

if __name__ == "__main__":
    # 创建队列实例
    q = mp.Queue()

    # 启动数据处理子进程，设置为守护进程（主进程结束时自动终止）
    process1 = mp.Process(target=process_queue_data, args=(q,))
    process1.daemon = True
    process1.start()

    # 主进程运行数据采集逻辑（也可单独启动子进程运行采集逻辑）
    collect_live_data(q)

修改说明 #

• 子进程循环改为无限阻塞监听：用queue.get()替代队列空状态检查，该方法会自动阻塞等待新数据，确保子进程持续运行并实时处理队列内容。
• 修正队列参数传递：启动子进程时传入队列实例q，而非队列类Queue。
• 完善并行逻辑：数据采集逻辑在主进程持续运行，子进程同步处理队列数据，两者完全并行。
• 守护进程设置：将处理队列的子进程设为守护进程，避免主进程结束后残留僵尸进程。

修改后，只要首个数据存入队列，子进程就会立即取出计算并打印结果，同时主进程持续采集实时数据，满足你的需求。

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