Python3.x下基于多线程队列的流式数据并行处理方案咨询

核心架构选型 #

该场景属于典型的IO密集型任务，采用分层队列+多线程生产者消费者模式完全可以满足需求，Python标准库自带的queue.Queue本身是线程安全的，无需额外处理并发锁问题，开发成本很低。

分层队列设计 #

• 原始数据缓存队列：存储从API流式拉取到的(用户ID, 原始文本)对，设置合理maxsize防止内存溢出，根据每次拉取100组的节奏，设置为200即可，队列满时拉取线程自动阻塞等待，避免上游速度远超下游处理能力。
• 待查询任务队列：存储特征提取后确认需要回复的(用户ID, 搜索特征)对，同样设置最大长度。
• 结果缓存队列：存储最终要返回给用户的(用户ID, 匹配数据)对。

线程角色拆分 #

每个模块独立运行，互不干扰：

• 数据拉取线程：仅负责从API拉取原始数据，写入原始数据队列即可，不需要关心后续处理逻辑。
• 特征提取线程池：多线程并行从原始数据队列取数据做特征提取，判断是否需要回复，需要的就写入待查询队列，不需要的直接丢弃。线程数可以根据特征提取的计算量调整，一般5~10个足够。
• 数据库查询线程池：多线程并行从待查询队列取搜索特征，拉取匹配数据后写入结果队列。数据库查询是纯IO操作，线程数可以开高一点，比如10~20个都没问题。
• 结果推送线程：仅负责从结果队列取数据，返回给对应用户ID。

关键优化点 #

• 如果特征提取逻辑是CPU密集型，多线程会受GIL限制，可以把特征提取模块换成多进程实现，用multiprocessing.Queue做进程间通信即可。
• 新增全局停止标志位，需要停机时先设置标志位，等待所有队列中已有的任务处理完再退出，避免数据丢失。
• 可以定期打印各队列的长度，方便调整线程数配置：如果原始数据队列长期处于满的状态，就增加特征提取线程数；如果待查询队列长期满，就增加数据库查询线程数。

最小可运行代码框架 #

import queue
import threading
import time

# 队列初始化，均设置最大长度防止内存溢出
raw_data_queue = queue.Queue(maxsize=200)
pending_query_queue = queue.Queue(maxsize=100)
result_queue = queue.Queue(maxsize=200)
# 全局停止标志位，用于优雅停机
shutdown_flag = threading.Event()

def api_fetch_worker():
    """数据拉取线程：仅负责从API拉原始数据"""
    while not shutdown_flag.is_set():
        # 模拟每次拉取100组(id, 文本)对
        batch_data = [(idx, f"user_raw_text_{idx}") for idx in range(100)]
        for item in batch_data:
            raw_data_queue.put(item)
        time.sleep(1)  # 模拟API拉取间隔

def feature_extract_worker():
    """特征提取线程：并行处理原始文本，筛选需要回复的任务"""
    while not shutdown_flag.is_set():
        try:
            user_id, raw_text = raw_data_queue.get(timeout=1)
            # 这里替换为你自己的特征提取逻辑
            extracted_feature = f"feature_{raw_text}"
            # 模拟20%概率需要回复
            if hash(user_id) % 5 == 0:
                pending_query_queue.put((user_id, extracted_feature))
            raw_data_queue.task_done()
        except queue.Empty:
            continue

def db_search_worker():
    """数据库查询线程：并行拉取匹配数据"""
    while not shutdown_flag.is_set():
        try:
            user_id, search_feature = pending_query_queue.get(timeout=1)
            # 这里替换为你自己的数据库查询逻辑
            matched_data = f"matched_result_{search_feature}"
            result_queue.put((user_id, matched_data))
            pending_query_queue.task_done()
        except queue.Empty:
            continue

def result_push_worker():
    """结果推送线程：返回数据给对应用户"""
    while not shutdown_flag.is_set():
        try:
            user_id, new_data = result_queue.get(timeout=1)
            # 这里替换为你自己的返回逻辑
            print(f"推送至用户{user_id}: {new_data}")
            result_queue.task_done()
        except queue.Empty:
            continue

if __name__ == "__main__":
    # 启动各模块线程
    threading.Thread(target=api_fetch_worker, daemon=True).start()
    # 启动5个特征提取线程
    for _ in range(5):
        threading.Thread(target=feature_extract_worker, daemon=True).start()
    # 启动10个数据库查询线程
    for _ in range(10):
        threading.Thread(target=db_search_worker, daemon=True).start()
    threading.Thread(target=result_push_worker, daemon=True).start()

    # 模拟运行30秒后优雅停机，可根据需求修改运行逻辑
    try:
        time.sleep(30)
    finally:
        shutdown_flag.set()
        # 等待所有队列中已有任务处理完成再退出
        raw_data_queue.join()
        pending_query_queue.join()
        result_queue.join()

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