Python3でasyncを使った非同期処理がうまく並列で動かない

実現したいこと

asyncを使った非同期処理をうまく並列に動かしたい

発生している問題・エラーメッセージ

task1とtask2が同時に出力させたいのにtask1しか結果を出力できていない

現状、task1,task2片方をコメントアウトすれば、それぞれは正常に動く状態です。
python-binanceを使っています

該当のソースコード

Python
1from __future__ import annotations
2
3import asyncio
4
5from binance import AsyncClient, DepthCacheManager, BinanceSocketManager, ThreadedDepthCacheManager
6
7
8# main関数
9async def main(symbol):
10    # tradesを取得するタスクとdepthを取得するタスクを起動
11    t = await trades(symbol)
12    task1 = asyncio.create_task(t)
13    # depthを取得するタスクを起動
14    task2 = asyncio.create_task(depth(symbol))
15    await asyncio.gather(task1, task2)
16
17
18async def trades(symbol):
19    client = await AsyncClient.create()
20    bsm = BinanceSocketManager(client)
21
22    async with bsm.individual_symbol_ticker_futures_socket(symbol) as fs:
23        while True:
24            res = await fs.recv()
25            print(res['data']['c'])
26
27
28async def depth(symbol):
29    dcm = ThreadedDepthCacheManager()
30    # 深さ情報を取得
31    # start is required to initialise its internal loop
32    dcm.start()
33
34    def handle_depth_cache(depth_cache):
35        print(f"symbol {depth_cache.symbol}")
36        print("top 5 bids")
37        print(depth_cache.get_bids()[:5])
38        print("top 5 asks")
39        print(depth_cache.get_asks()[:5])
40        print("last update time {}".format(depth_cache.update_time))
41
42    dcm_name = dcm.start_depth_cache(handle_depth_cache, symbol=symbol)
43
44    dcm.join()
45
46
47if __name__ == "__main__":
48    loop = asyncio.get_event_loop()
49    loop.run_until_complete(main('ETHUSDT'))

回答1件

ベストアンサー

１．depth コルーチンの中で thread ベースのクライアント使ってるようですが、
多分非同期Client がなかった時代の古いバージョンの使い方と混在してると思います。
AsyncClient と DepthCacheManagerを使用して下さい。

DepthCacheManager or OptionsDepthCacheManager Usage

２．そのうえで

py
1    # tradesを取得するタスクとdepthを取得するタスクを起動
2    t = await trades(symbol)
3    task1 = asyncio.create_task(t)

ここが並行動作にならない原因です。ここはシンプルに

py
1    # t = await trades(symbol)  #削除
2    task1 = asyncio.create_task(trades(symbol))

でいいでしょう。

追記

非同期使うときはちゃんと、Gracefully にストップできるように考慮しましょう。大切なリソースなので、close_connection を使いましょう。質問文のままで止めるのにCtrl+C乱発してたらそのうちソケットが枯渇するかもしれませんので。

py
1from __future__ import annotations
2import asyncio
3from binance import AsyncClient, DepthCacheManager, BinanceSocketManager, ThreadedDepthCacheManager
4
5clients = []
6
7async def main(symbol):
8    task1 = asyncio.create_task(trades(symbol))
9    task2 = asyncio.create_task(depth(symbol))
10    await asyncio.gather(task1, task2)
11
12async def trades(symbol):
13    client = await AsyncClient.create()
14    clients.append(client)
15    bsm = BinanceSocketManager(client)
16
17    async with bsm.individual_symbol_ticker_futures_socket(symbol) as fs:
18        while True:
19            res = await fs.recv()
20            print(res['data']['c'])
21
22async def depth(symbol):
23    client = await AsyncClient.create()
24    clients.append(client)
25    dcm = DepthCacheManager(client, symbol)
26    async with dcm as dcm_socket:
27        while True:
28            depth_cache = await dcm_socket.recv()
29            print("symbol {}".format(depth_cache.symbol))
30            print("top 5 bids")
31            print(depth_cache.get_bids()[:5])
32            print("top 5 asks")
33            print(depth_cache.get_asks()[:5])
34            print("last update time {}".format(depth_cache.update_time))
35
36async def astop():
37    [await client.close_connection() for client in clients]
38    print("stop.")
39
40if __name__ == "__main__":
41    try:
42        asyncio.run(main('ETHUSDT'))
43    except KeyboardInterrupt:
44        asyncio.run(astop())