运行 2 个异步订阅答案

【问题标题】：Running 2 asyncio subscriptions运行 2 个异步订阅
【发布时间】：2019-11-21 14:47:23
【问题描述】：

我试图在一个脚本中运行 2 个不同的代码。我可以让它们分开工作，但我很难让它们一起工作。

第一个是：

import asyncio
import websockets
import json

msg = \
{
  "jsonrpc" : "2.0",
  "id" : 9929,
  "method" : "public/auth",
  "params" : {
    "grant_type" : "client_credentials",
    "client_id" : "xxxxxxxxxxxxxxxxxx",
    "client_secret" : "xxxxxxxxxxxxxxxxx"
  }
}

async def call_api(msg):
   async with websockets.connect('wss://testapp.deribit.com/ws/api/v2') as websocket:
       await websocket.send(msg)
       while websocket.open:
           response = await websocket.recv()
           # do something with the response...
           print(response)

asyncio.get_event_loop().run_until_complete(call_api(json.dumps(msg)))

第二个是：

import asyncio
import websockets
import json

# To subscribe to this channel:
msg = \
    {"jsonrpc": "2.0",
     "method": "public/subscribe",
     "id": 42,
     "params": {
        "channels": ["ticker.BTC-PERPETUAL.raw"]}
    }

async def call_api(msg):
   async with websockets.connect('wss://testapp.deribit.com/ws/api/v2') as websocket:
       await websocket.send(msg)
       while websocket.open:
           response = await websocket.recv()
           # do something with the notifications...
           print(response)

asyncio.get_event_loop().run_until_complete(call_api(json.dumps(msg)))

我曾尝试将它们组合在 1 个脚本中，如下所示：

import asyncio
import websockets
import json


# To subscribe to this channel:
msg = \
    {"jsonrpc": "2.0",
     "method": "public/subscribe",
     "id": 42,
     "params": {
        "channels": ["ticker.BTC-PERPETUAL.raw"]}
    }

msg1 = \
{
  "jsonrpc" : "2.0",
  "id" : 9929,
  "method" : "public/auth",
  "params" : {
    "grant_type" : "client_credentials",
    "client_id" : "xxxxxxxxxxxxxxxxx",
    "client_secret" : "xxxxxxxxxxxxxxxxxxxxxxxxx"
  }
}

async def call_api1():
   async with websockets.connect('wss://testapp.deribit.com/ws/api/v2') as websocket:
       await websocket.send(msg)       
       await websocket.send(msg1)
       while websocket.open:
           response = await websocket.recv()
           # do something with the response...
           print(response)
           #access_token=response['access_token']
           #print(access_token)

asyncio.get_event_loop().run_until_complete(call_api1())

我如何得到一个错误说 websockets.exceptions.InvalidStatusCode: server denied WebSocket connection: HTTP 502.

我想我在最后一行 asyncio.get_event_loop().run_until_complete(call_api1()) 上搞砸了。请注意，我已经读到我需要从同一个 websocket 进行两个订阅，因为 API 一次只允许 1 个。

感谢任何帮助，因为我还是新手。

【问题讨论】：

标签： python json websocket python-asyncio

【解决方案1】：

自己解决了这个问题，方法是为每条消息使用单独的 async def，而不是使用以下内容而不是最后一行：

loop = asyncio.get_event_loop()
try:
    asyncio.ensure_future(call_api(json.dumps(msg)))
    asyncio.ensure_future(call_api1(json.dumps(msg1)))      
    loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    print('closing loop')
    loop.close()

【讨论】：

【解决方案2】：

确实，您需要为两者创建单独的任务，然后安排任务（使用ensure_future 或create_task）。

除了loop.run_forever()，您还可以对执行 asyncio.wait([task1, task2]) 的协程执行 loop.run_until_complete

例如：

import asyncio

async def task1():
    await asyncio.sleep(3)
    print("task1: done")

async def task2():
    await asyncio.sleep(1)
    print("task2: done")


async def main():
    t1 = asyncio.ensure_future(task1())
    t2 = asyncio.ensure_future(task2())
    await asyncio.wait([t1, t2])


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

【讨论】：