线程内的Asyncio函数

问题描述

加密源是一个python库，它使用异步库来获取不同Crypto交换的实时价格。 在这个简短的程序中，我们尝试在一个独立的线程中运行加密源FeedHandler。代码示例如下所示：

import functools as fct
from cryptofeed import FeedHandler
from cryptofeed.defines import BID, ASK, L2_BOOK
from cryptofeed.exchanges import Kraken
from datetime import datetime
import threading


async def bookfunc(params , orderbooks, feed, symbol, book, timestamp, receipt_timestamp):

    print(f'Timestamp: {timestamp} Cryptofeed Receipt: {receipt_timestamp} Feed: {feed} Symbol: {symbol}'
          f' Book Bid Size is {len(book[BID])} Ask Size is {len(book[ASK])}')
    orderbooks = filter_orderbook(orderbooks, book, symbol, params['orderbook']['depth'])


def func():

    # Parameters
    params = {'orderbook': {'depth': 2}, 'price_model':{}, 'trade_model': {}}
    config = {'log': {'filename': 'logs/demo.log', 'level': 'INFO'}}

    orderbooks = {}

    f = FeedHandler(config=config)
    f.add_feed(Kraken(checksum_validation=True, subscription={L2_BOOK: ['BTC-USD', 'ETH-USD', 'LINK-USD', 'LTC-USD', 'ADA-USD']},
                      callbacks={L2_BOOK: fct.partial(bookfunc, params, orderbooks)})) # This way passes the orderbooks inside the callback

    f.run()


if __name__ == '__main__':
    thread = threading.Thread(target=func, args=())
    thread.start()

执行代码时，会出现以下错误：

raise RuntimeError('There is no current event loop in thread %r.'
RuntimeError: There is no current event loop in thread 'Thread-1'.

您知道如何解决此问题吗？

编辑： 这是堆栈溢出中不同问题的解决方案。下面是一个示例问题：

Running cryptofeed (asyncio library) in a new thread

---

解决方案

(截至cryptofeed==1.9.2)要使其从线程运行，有两点很重要：

1. FeedHandler.run默认设置信号处理程序，必须 从主线程完成。为了避免这种情况，有 方法上的install_signal_handlers参数。
2. FeedHandlerset's uvloop's policy的初始值设定项，但是 不调用uvloop.install()(假设它是应用程序的 我想是责任)。不使用它asyncio.set_event_loop 不会有效果的。或者，您也可以设置'uvloop': False 在提要处理程序配置中(如下所示)，或仅卸载 uvloop。

import asyncio
import threading

from cryptofeed import FeedHandler
from cryptofeed.defines import BID, ASK, L2_BOOK
from cryptofeed.exchanges import Kraken


async def bookfunc(**kwargs):
    print('bookfunc', kwargs)
    

def run_feed_handler_forever():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    config = {
        'uvloop': False,
        'log': {'filename': 'log.log', 'level': 'DEBUG'},
    }
    l2_book = ['BTC-USD', 'ETH-USD', 'LINK-USD', 'LTC-USD', 'ADA-USD']
    feed = Kraken(
        subscription={L2_BOOK: l2_book}, callbacks={L2_BOOK: bookfunc}
    )
    fh = FeedHandler(config)
    fh.add_feed(feed)
    fh.run(install_signal_handlers=False)


if __name__ == '__main__':
    thread = threading.Thread(target=run_feed_handler_forever)
    thread.start()
    thread.join()