如何在独立线程运行asyncio库cryptofeed的FeedHandler并解决事件循环报错

错误原因 #

• Python 3.7及以上版本中，子线程默认不会自动创建和绑定asyncio事件循环，cryptofeed的FeedHandler.run()方法内部会调用asyncio.get_event_loop()获取当前线程绑定的事件循环，在ThreadPoolExecutor创建的子线程中找不到对应的事件循环，因此抛出RuntimeError。
• 原代码逻辑存在两个隐含问题：await self.start_feed()会一直等待fh.run()执行结束，但fh.run()是持续运行的websocket监听逻辑，永远不会返回，因此后续的2秒打印循环永远不会执行；使用同步的time.sleep()会阻塞主线程的asyncio事件循环，会导致回调逻辑调度异常。
• 跨线程读写共享变量self.coin_book未做并发保护，可能出现读到半更新数据的异常。

修复方案 #

需要在子线程中手动创建并绑定事件循环，同时将FeedHandler启动为后台线程不阻塞主线程打印逻辑，修复后的完整代码如下：

import asyncio
import threading
from asyncio import Lock

from cryptofeed import FeedHandler
from cryptofeed.exchanges import FTX
from cryptofeed.defines import L2_BOOK


class TestApp:
    def __init__(self, symbols: list):
        self.coin_book: dict = {}
        self.symbols = symbols
        self.lock = Lock()  # 加锁保证跨线程读写安全

    async def book_update(self, feed, symbol, book, timestamp, receipt_timestamp):
        async with self.lock:
            self.coin_book = book

    def run_feed_handler(self):
        # 子线程手动创建并绑定事件循环
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        fh = FeedHandler()
        fh.add_feed(FTX(symbols=self.symbols, channels=[L2_BOOK], callbacks={L2_BOOK: self.book_update}))
        fh.run()

    async def display(self):
        async with self.lock:
            if not self.coin_book:
                print("尚未获取到订单簿数据")
                return
            print('bid', self.coin_book['bid'].peekitem(-1))
            print('ask', self.coin_book['ask'].peekitem(0))

    async def run(self):
        # 启动FeedHandler为后台守护线程，主线程退出时自动销毁
        feed_thread = threading.Thread(target=self.run_feed_handler, daemon=True)
        feed_thread.start()

        while True:
            await asyncio.sleep(2)  # 改用异步sleep不阻塞事件循环
            await self.display()


if __name__ == '__main__':
    app = TestApp(['RAY-USD'])
    asyncio.run(app.run())

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