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
相关产品推荐
相关产品推荐

