Python循环中使用线程触发RuntimeError，求优化方案

背景 #

我正在做研究项目，基于币安API设计交易策略，程序基本流程：

• 等待API推送的K线数据（不同交易对/时段等待时长不固定，因此用threading.Condition()的wait()和notify()方法）
• 数据可用时触发notify()，随后执行技术分析
• 通过while循环维持程序持续运行，但每次迭代都会启动新线程
• 测试时处理约12000条数据后报错：

File "/usr/lib/python3.8/threading.py", line 852, in start
_start_new_thread(self._bootstrap, ())
RuntimeError: can't start new thread

推测是每次循环创建新线程耗尽了系统资源，需要更优的数据等待方式或线程管理方案。

现有测试代码 #

数据读取部分 #

tc_kline_received = threading.Condition()

for file in list_of_files:
    with open(file, "r") as csv_file:
        file_reader = reader(csv_file)
        for line in file_reader:
            btc_busd_5_min.append(line)

数据收集线程函数 #

def collect_kline():
    global btc_busd_5_min
    global kline_btc_busd_5_min
    global klines_dict
    kline_btc_busd_5_min = btc_busd_5_min.pop(0)
    klines_dict["btc_busd_5_min"] = kline_btc_busd_5_min
    with tc_kline_received:
        tc_kline_received.notify()

主循环 #

while True:
    with tc_kline_received:
        t_collect_kline = threading.Thread(target=collect_kline)
        t_collect_kline.start()
        tc_kline_received.wait()
        t_collect_kline.join()   
    insert_kline_to_db(klines_dict)
    create_ta_db(klines_dict)

解决方案 #

1. 复用单线程替代循环创建新线程 #

核心问题是每次循环新建线程，系统线程数量存在上限（单进程通常几千到一万级别），迭代12000次后资源耗尽。可以启动一个常驻线程负责数据收集，无需每次循环新建：

import threading
import time

# 保留原有全局变量
tc_kline_received = threading.Condition()
btc_busd_5_min = []
kline_btc_busd_5_min = None
klines_dict = {}

def collect_kline_loop():
    """常驻线程：持续从列表取数据并通知主进程"""
    global btc_busd_5_min, kline_btc_busd_5_min, klines_dict
    while True:
        # 列表为空时短暂休眠避免CPU空转（真实API场景改为等待推送）
        while not btc_busd_5_min:
            time.sleep(0.1)
        # 读取并存储数据
        kline_btc_busd_5_min = btc_busd_5_min.pop(0)
        klines_dict["btc_busd_5_min"] = kline_btc_busd_5_min
        # 通知主进程处理数据
        with tc_kline_received:
            tc_kline_received.notify()

# 仅启动一次常驻线程
t_collect = threading.Thread(target=collect_kline_loop, daemon=True)
t_collect.start()

# 主循环：等待数据并处理
while True:
    with tc_kline_received:
        tc_kline_received.wait()
    insert_kline_to_db(klines_dict)
    create_ta_db(klines_dict)

2. 用线程安全队列替代全局列表+Condition #

queue.Queue自带线程安全的阻塞等待机制，比手动管理Condition+全局变量更简洁可靠：

import threading
import queue

# 用线程安全队列替代全局列表
kline_queue = queue.Queue()

def collect_kline_loop():
    """常驻线程：从CSV读取数据放入队列（真实场景改为监听API推送）"""
    for file in list_of_files:
        with open(file, "r") as csv_file:
            file_reader = reader(csv_file)
            for line in file_reader:
                kline_queue.put(line)

def process_kline():
    """主逻辑：从队列取数据并处理"""
    while True:
        # 队列为空时自动阻塞等待
        kline_data = kline_queue.get()
        klines_dict = {"btc_busd_5_min": kline_data}
        insert_kline_to_db(klines_dict)
        create_ta_db(klines_dict)
        kline_queue.task_done()

# 启动数据收集线程
t_collect = threading.Thread(target=collect_kline_loop, daemon=True)
t_collect.start()

# 启动数据处理逻辑
process_kline()

3. 真实API场景优化建议 #

后续对接币安真实API时，推荐：

• 使用币安WebSocket推送替代轮询，实时接收K线数据，避免线程空转
• 用queue.Queue做数据缓冲，解耦数据接收与处理逻辑
• 用固定线程池（concurrent.futures.ThreadPoolExecutor）处理技术分析和数据库写入，避免无限制创建线程

关键总结 #

• 避免循环创建线程：线程的创建销毁有开销，系统线程数量有限，复用线程或使用线程池是最优解
• 优先使用线程安全队列：queue.Queue内置阻塞、同步机制，比手动管理Condition+全局变量更稳定

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