python - 线程内的异步函数

问题描述

cryptofeed是一个 python 库，它使用 asyncio 库来获取不同加密交易所的实时价格。在这个简短的程序中，我们尝试在一个独立的线程中运行 cryptofeed 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'.

知道如何解决这个问题吗？

编辑：这是stackoverflow中不同问题的解决方案。一个例子是以下问题：

在新线程中运行 cryptofeed（asyncio 库）

标签： pythonpython-asynciopython-multithreadingcryptofeed

（截至cryptofeed==1.9.2）有两件重要的事情让它从一个线程运行：

FeedHandler.run默认情况下设置信号处理程序，这必须从主线程完成。为避免这种情况， install_signal_handlers该方法存在争议。
FeedHandler set 的uvlooppolicy的初始化程序，但不调用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()

python - 线程内的异步函数

问题描述

解决方案

推荐阅读