在ThreadPoolExecutor线程中运行cryptofeed提示无事件循环错误如何解决？

问题原因 #

Python 3.7+版本中，仅主线程会默认初始化asyncio事件循环，ThreadPoolExecutor创建的工作子线程默认没有绑定任何事件循环实例。cryptofeed的FeedHandler.run()方法内部会直接调用asyncio.get_event_loop()获取当前线程的事件循环，子线程中不存在对应实例就会抛出你遇到的RuntimeError。你之前把FeedHandler初始化放到子线程里还是报错，就是因为子线程还是没有提前创建事件循环。

解决方法 #

在子线程的执行函数开头，先手动为当前子线程创建并绑定事件循环即可，修改后的代码如下：

import asyncio
import concurrent.futures
from cryptofeed import FeedHandler
from cryptofeed.exchanges import Gateio
from cryptofeed.backends.postgres import TradePostgresGateio

def threadable():    
    # 手动为当前子线程创建并设置事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    f = FeedHandler()
    f.add_feed(Gateio(channels=[TRADES], symbols=list_tmp, callbacks={ TRADES: TradePostgresGateio(**postgres_cfg)}))
    f.run()

executor = concurrent.futures.ThreadPoolExecutor(16)
job2 = executor.submit(threadable)
# 注意：job2.result()会阻塞主线程直到任务结束，如果你只是要后台运行不需要调用result()
# 如果需要终止任务，可以调用job2.cancel()

注意事项 #

• 每个运行cryptofeed FeedHandler的子线程都需要单独绑定自己的事件循环，不要跨线程共享事件循环实例
• 如果你不需要多线程并行跑多个不同交易所的数据流，完全可以不用ThreadPoolExecutor，直接在主线程跑FeedHandler即可，cryptofeed本身是asyncio异步实现，单线程就能处理大量并发数据流
• 调用job2.result()会阻塞主线程等待子线程任务返回，如果你只是需要后台运行不需要等待结果，不需要调用该方法

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