如何从子线程停止asyncio事件循环 解决cryptofeed FeedHandler停止报错

问题原因 #

你的推测完全正确，asyncio的事件循环和创建它的线程是强绑定的，只能在对应线程内直接调用事件循环相关方法。cryptofeed的FeedHandler.stop()内部会调用asyncio.get_event_loop()获取当前线程的事件循环，而你在子线程执行这个方法时，子线程没有绑定任何事件循环，所以直接抛出异常。

解决方案 #

核心思路是将停止操作提交到事件循环所属的主线程中执行，asyncio官方提供了线程安全的方法实现跨线程提交任务：

1. 提前在主线程中保存事件循环的引用
2. 子线程中使用call_soon_threadsafe（提交同步方法）或run_coroutine_threadsafe（提交协程），将stop操作交给主线程的事件循环执行

修改后完整代码 #

from cryptofeed import FeedHandler
from cryptofeed.feed import Feed
from cryptofeed.defines import L2_BOOK, BID, ASK
from cryptofeed.exchange.ftx import FTX
from threading import Thread
from time import sleep
import asyncio


class Executor:
    def __init__(self,  feed: Feed, coin_symbol: str, fut_symbol: str):
        self.coin_symbol = coin_symbol
        self.fut_symbol = fut_symbol
        self.feed = feed

        self.fh = FeedHandler()
        self.loop = None # 存储主线程的事件循环引用

        self._coin_top_book: dict = {}
        self._fut_top_book: dict = {}

    async def _book_update(self, feed, symbol, book, timestamp, receipt_timestamp):
        if symbol == self.coin_symbol:
            self._coin_top_book[BID] = book[BID].peekitem(-1)
            self._coin_top_book[ASK] = book[ASK].peekitem(0)
        elif symbol == self.fut_symbol:
            self._fut_top_book[BID] = book[BID].peekitem(-1)
            self._fut_top_book[ASK] = book[ASK].peekitem(0)

    def start_feed(self):
        self.fh.add_feed(self.feed(symbols=[self.fut_symbol, self.coin_symbol], channels=[L2_BOOK],
                                   callbacks={L2_BOOK: self._book_update}))
        # 主线程获取事件循环并存储
        self.loop = asyncio.get_event_loop()
        # 将循环传入feedhandler的run方法
        self.fh.run(loop=self.loop)

    def shoot(self):
        # 等待订单簿初始化
        while len(self._coin_top_book) == 0 or len(self._fut_top_book) == 0:
            sleep(1)

        for i in range(5):
            print(self._coin_top_book)
            sleep(1)  # 执行业务逻辑

        # 核心修改：将stop操作提交到主线程的事件循环执行
        self.loop.call_soon_threadsafe(self.fh.stop)

    def run(self):
        th1 = Thread(target=self.shoot)
        th1.start()

        self.start_feed()


def main():
    g = Executor(feed=FTX, coin_symbol='SOL-USD', fut_symbol='SOL-PERP')
    g.run()


if __name__ == '__main__':
    main()

补充说明 #

如果后续版本中FeedHandler.stop()改为了协程方法，只需要把停止部分的代码替换为：

asyncio.run_coroutine_threadsafe(self.fh.stop(), self.loop)

即可正常执行。

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