Python Queue多线程处理股票数据异常：无法批量处理145个标的

针对你遇到的结果数量异常、大批量数据挂起问题，以及最多50线程的需求，以下是修复后的实现方案：

核心问题分析 #

1. 结果数量异常：大概率是结束信号被误当作有效结果，或task_done()未正确调用导致队列计数混乱。
2. 程序挂起：工作线程未收到退出信号，一直阻塞等待任务；或join()未正确触发，主线程无限等待。
3. q.put阻塞：任务队列无边界导致内存占用过高，或单个线程处理卡住未释放队列资源。

修复后的完整代码实现 #

import threading
import queue

# 替换为你的股票数据处理逻辑
def process_stock_data(stock_item):
    # 示例：模拟数据处理，返回处理后的结果
    # 实际场景中替换为接口请求、计算等逻辑
    return {"code": stock_item[0], "result": "processed"}

def worker(input_queue, result_queue):
    """工作线程：从任务队列取数据，处理后存入结果队列"""
    while True:
        task = input_queue.get()
        # 收到退出信号，终止线程
        if task is None:
            input_queue.task_done()
            break
        
        # 异常捕获，避免单个线程崩溃影响全局
        try:
            processed_result = process_stock_data(task)
            result_queue.put(processed_result)
        except Exception as e:
            # 记录错误信息，可根据需求调整
            result_queue.put({"code": task[0], "error": str(e)})
        finally:
            # 必须调用task_done，否则队列join会无限阻塞
            input_queue.task_done()

def main():
    # 配置参数
    MAX_WORKERS = 50
    # 模拟145个股票标的数据，替换为你的实际数据获取逻辑（比如从Excel读取A2:C145）
    stock_data = [("STOCK{}".format(i), "data1", "data2") for i in range(145)]
    
    # 初始化任务队列和结果队列
    task_queue = queue.Queue()
    result_queue = queue.Queue()

    # 创建并启动工作线程
    threads = []
    for _ in range(MAX_WORKERS):
        thread = threading.Thread(target=worker, args=(task_queue, result_queue))
        thread.daemon = True  # 守护线程：主线程退出时自动终止
        thread.start()
        threads.append(thread)

    # 向任务队列添加所有股票数据
    for item in stock_data:
        task_queue.put(item)

    # 向每个工作线程发送退出信号（每个线程对应一个None）
    for _ in range(MAX_WORKERS):
        task_queue.put(None)

    # 等待所有任务处理完成
    task_queue.join()

    # 收集所有处理结果
    final_results = []
    while not result_queue.empty():
        final_results.append(result_queue.get())
        result_queue.task_done()
    result_queue.join()

    # 验证结果数量
    print(f"总任务数：{len(stock_data)}，处理结果数：{len(final_results)}")
    # 后续可对final_results进行分析、存储等操作

if __name__ == "__main__":
    main()

关键修复点说明 #

1. 线程退出机制：通过向任务队列发送None作为结束信号，确保每个工作线程能主动终止，避免主线程挂起。
2. 队列计数安全：所有任务（包括结束信号）都调用task_done()，保证queue.join()能正确感知任务完成。
3. 线程数控制：固定创建50个工作线程，无论任务数量多少，都不会超过这个上限，避免资源耗尽。
4. 异常处理：工作线程内捕获异常，防止单个任务崩溃导致整个线程终止，同时记录错误信息便于排查。
5. 守护线程设置：确保主线程退出时，所有工作线程自动终止，避免残留线程。

额外优化建议 #

• 给任务队列设置maxsize（比如maxsize=MAX_WORKERS*2），限制队列最大容量，避免内存占用过高：

  task_queue = queue.Queue(maxsize=MAX_WORKERS*2)
  

• 若处理逻辑涉及网络请求，务必设置超时时间，避免单个线程长期阻塞：

  import requests
  def process_stock_data(stock_item):
      try:
          response = requests.get(f"https://api.example.com/stock/{stock_item[0]}", timeout=10)
          return response.json()
      except requests.exceptions.Timeout:
          return {"code": stock_item[0], "error": "请求超时"}
  

• 若需要共享全局资源（比如数据库连接），使用threading.Lock保证线程安全：

  db_lock = threading.Lock()
  def process_stock_data(stock_item):
      with db_lock:
          # 执行数据库操作
          pass
  

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