如何在 Python Tornado websocket 中进行非阻塞 while 循环以流式传输数据答案

【问题标题】：How to make non-blocking while loop in Python Tornado websocket to stream data如何在 Python Tornado websocket 中进行非阻塞 while 循环以流式传输数据
【发布时间】：2021-01-07 13:38:35
【问题描述】：

我正在尝试从 Deribit 的 API 中提取数据，将其输入队列，然后在 Python Tornado Websocket 服务器中的 while 循环中将其发送出去。我面临的问题是 on_message 方法中的 while 循环阻塞了执行流程。我不确定如何创建一种方法，使 websocket 保持打开状态并流式传输数据而不阻塞队列生成功能。这可能吗？每个任务的代码如下。

Websocket 代码：

class web_socket_handler(ws.WebSocketHandler):
    @classmethod
    def route_urls(cls):
        return [(r'/',cls, {}),]
    
    def simple_init(self):
        self.last = time.time()
        self.stop = False
    
    def open(self):
        self.simple_init()
        print("New client connected")
        
    async def on_message(self, message):
        if message == 'BTC':
            while True:
                try:
                    reply = wsQueue.get(block=False)
                    self.write_message(reply)
                except:
                    pass
        else:
            self.write_message('bad request - closing connection!')
            self.close()
        self.last = time.time()
    
    def on_close(self):
        print("connection is closed")
        try:
            self.loop.stop()
        except:
            pass
    def close(self):
        self.ws_connection.close()
    def check_origin(self, origin):
        return True

API 代码：

class Deribit:
    def __init__(self, contract, currency, q):
        self.baseUrl = "wss://www.deribit.com/ws/api/v2"
        self.contract = contract
        self.currency = currency
        self.q = q

    async def getIndex(self):
        msg = {
            "jsonrpc": "2.0",
            "method": "public/get_index",
            "params": {f"currency": "{self.currency}"},
        }
        async with websockets.connect(self.baseUrl) as websocket:
            await websocket.send(json.dumps(msg))
            while websocket.open:
                if flag == 'stop':
                    await websocket.close()
                    return
                elif not websocket.open:
                    await self.getIndex()
                else:        
                    response = await websocket.recv()
                    response = json.loads(response)
                    response['type'] = 'getIndex'
                    self.q.put(response)

    async def flowSubscribe(self):
        msg = {
            "jsonrpc": "2.0",
            "method": "public/subscribe",
            "params": {"channels": [f"trades.option.{self.currency}.raw"]},
        }
        async with websockets.connect(self.baseUrl) as websocket:
            await websocket.send(json.dumps(msg))
            while websocket.open:
                if flag == 'stop':
                    await websocket.close()
                    return
                else:        
                    response = await websocket.recv()
                    response = json.loads(response)
                    response['type'] = 'flowSubscribe'
                    self.q.put(response)
            '''if not websocket.open:
                print('flowSubscribe closed')
                await self.flowSubscribe()'''

    async def tradeSubscribe(self):
        msg = {
            "jsonrpc": "2.0",
            "method": "public/subscribe",
            "params": {"channels": [f"trades.{self.contract}.raw"]},
        }
        async with websockets.connect(self.baseUrl) as websocket:
            await websocket.send(json.dumps(msg))
            while websocket.open:
                if flag == 'stop':
                    await websocket.close()
                    return
                elif not websocket.open:
                    await self.tradeSubscribe()
                else:        
                    response = await websocket.recv()
                    response = json.loads(response)
                    response['type'] = 'tradeSubscribe'
                    self.q.put(response)

    async def orderbookSubscribe(self):
        msg = {
            "jsonrpc": "2.0",
            "method": "public/subscribe",
            "params": {"channels": [f"book.{self.contract}.100ms"]},
        }
        async with websockets.connect(self.baseUrl) as websocket:
            await websocket.send(json.dumps(msg))
            while websocket.open:
                if flag == 'stop':
                    await websocket.close()
                    return
                elif not websocket.open:
                    await self.orderbookSubscribe()
                else:        
                    response = await websocket.recv()
                    response = json.loads(response)
                    response['type'] = 'orderbookSubscribe'
                    self.q.put(response)

    async def orderbookUpdateSubscribe(self):
        msg = {
            "jsonrpc": "2.0",
            "method": "public/subscribe",
            "params": {"channels": [f"book.{self.contract}.1.raw"]},
        }
        async with websockets.connect(self.baseUrl) as websocket:
            await websocket.send(json.dumps(msg))
            while websocket.open:
                if flag == 'stop':
                    await websocket.close()
                    return 
                else:        
                    response = await websocket.recv()
                    response = json.loads(response)
                    response['type'] = 'orderbookUpdateSubscribe'
                    self.q.put(response)
            if not websocket.open:
                print('orderbookUpdateSubscribe Closed')
                await self.orderbookUpdateSubscribe()

    async def tickerSubscribe(self, i):
        msg = {
            "jsonrpc": "2.0",
            "method": "public/subscribe",
            "params": {"channels": i},
        }
        async with websockets.connect(self.baseUrl) as websocket:
            await websocket.send(json.dumps(msg))
            while websocket.open:
                if flag == 'stop':
                    await websocket.close()
                    return
                elif not websocket.open:
                    await self.tickerSubscribe(i)
                else:
                    response = await websocket.recv()
                    response = json.loads(response)
                    response['type'] = 'tickerSubscribe'
                    self.q.put(response)

    def getInstruments(self, expired):
        msg = f"https://www.deribit.com/api/v2/public/get_instruments?currency={self.currency}&expired={expired}&kind=option"
        response = requests.get(msg)
        return response.json()
    
    def getLastTradesByCurrency(self, count, expired):
        msg = f"https://www.deribit.com/api/v2/public/get_last_trades_by_currency?count={count}&currency={self.currency}"
        response = requests.get(msg)
        return response.json()

队列生产者函数：

def threadLoop():
    global flag
    q = queue.Queue()
    flag = 'go'
    count = 0
    loop = asyncio.get_event_loop()
    time = datetime.datetime.now().time().strftime("%H:%M:%S")
    time = datetime.datetime.strptime(time, '%H:%M:%S').time()
    startTime = datetime.time(11, 58, 0)
    endTime = datetime.time(11, 58, 2)
    while True:
        time = datetime.datetime.now().time().strftime("%H:%M:%S")
        time = datetime.datetime.strptime(time, '%H:%M:%S').time()
        if time <= startTime or time >= endTime:
            t1 = threading.Thread(target=loop.run_until_complete, args=(main(q),), daemon=True)
            t1.start()
            break
    while True:
        try:
            if flag == 'stop':
                for i in range(len(tasks)):
                    tasks[i].cancel()
                t1.join()
                if t1.is_alive() is False:
                    print('threads stopped')
                    threadLoop()
                else:
                    break
            elif flag != 'stop':
                message = q.get(block = False)
                if message != None:
                    if message['type'] == 'tickerSubscribe':
                        instrument_name = message['params']['data']['instrument_name']
                        timestamp = message['params']['data']['timestamp']
                        bid_price = message['params']['data']['best_bid_price']
                        bid_iv = message['params']['data']['bid_iv']
                        ask_price = message['params']['data']['best_ask_price']
                        ask_iv = message['params']['data']['ask_iv']
                        open_interest = message['params']['data']['open_interest']
                        try:
                            old_open_interest = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['open_interest']
                            orderbooks[instrument_name].append({'timestamp': timestamp, 'bid_price': bid_price, 'bid_iv': bid_iv, 'ask_price': ask_price, 'ask_iv': ask_iv, 'open_interest': open_interest, 'old_open_interest': old_open_interest})
                        except:
                            orderbooks[instrument_name].append({'timestamp': timestamp, 'bid_price': bid_price, 'bid_iv': bid_iv, 'ask_price': ask_price, 'ask_iv': ask_iv, 'open_interest': open_interest})
                    elif message['type'] == 'flowSubscribe':
                        instrument_name = message['params']['data'][0]['instrument_name']
                        trade_price = message['params']['data'][0]['price']
                        bid_price = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['bid_price']
                        ask_price = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['ask_price']
                        bid_trade_difference = str(bid_price - trade_price)
                        ask_trade_difference = str(ask_price - trade_price)
                        if bid_trade_difference[0] == '-':
                            bid_trade_difference = bid_trade_difference[1:]
                        if ask_trade_difference[0] == '-':
                            ask_trade_difference = ask_trade_difference[1:]
                        bid_trade_difference = float(bid_trade_difference)
                        ask_trade_difference = float(ask_trade_difference)
                        if bid_price or ask_price != message['params']['data'][0]['price']:
                            bid_price = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['bid_price']
                            ask_price = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['ask_price']
                            if bid_price or ask_price != message['params']['data'][0]['price']:
                                bid_price = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['bid_price']
                                ask_price = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['ask_price']
                                if bid_price == message['params']['data'][0]['price']:
                                    message['params']['data'][0]['direction'] == 'sell'
                                    message['params']['data'][0]['bid_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['bid_price']
                                    message['params']['data'][0]['ask_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['ask_price']
                                    message['params']['data'][0]['bid_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['bid_iv']
                                    message['params']['data'][0]['ask_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['ask_iv']
                                    message['params']['data'][0]['open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['open_interest']
                                    message['params']['data'][0]['old_open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['old_open_interest']
                                elif ask_price == message['params']['data'][0]['price']:
                                    message['params']['data'][0]['direction'] == 'buy'
                                    message['params']['data'][0]['bid_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['bid_price']
                                    message['params']['data'][0]['ask_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['ask_price']
                                    message['params']['data'][0]['bid_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['bid_iv']
                                    message['params']['data'][0]['ask_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['ask_iv']
                                    message['params']['data'][0]['open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['open_interest']
                                    message['params']['data'][0]['old_open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 3]['old_open_interest']
                            elif bid_price or ask_price == message['params']['data'][0]['price']:
                                if bid_price == message['params']['data'][0]['price']:                      
                                    message['params']['data'][0]['direction'] == 'sell'
                                    message['params']['data'][0]['bid_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['bid_price']
                                    message['params']['data'][0]['ask_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['ask_price']
                                    message['params']['data'][0]['bid_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['bid_iv']
                                    message['params']['data'][0]['ask_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['ask_iv']
                                    message['params']['data'][0]['open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['open_interest']
                                    message['params']['data'][0]['old_open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['old_open_interest']
                                elif ask_price == message['params']['data'][0]['price']:
                                    message['params']['data'][0]['direction'] == 'buy'
                                    message['params']['data'][0]['bid_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['bid_price']
                                    message['params']['data'][0]['ask_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['ask_price']
                                    message['params']['data'][0]['bid_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['bid_iv']
                                    message['params']['data'][0]['ask_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['ask_iv']
                                    message['params']['data'][0]['open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['open_interest']
                                    message['params']['data'][0]['old_open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 1]['old_open_interest']
                        elif bid_price or ask_price == message['params']['data'][0]['price']:
                            if bid_price == message['params']['data'][0]['price']:                       
                                message['params']['data'][0]['direction'] == 'sell'
                                message['params']['data'][0]['bid_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['bid_price']
                                message['params']['data'][0]['ask_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['ask_price']
                                message['params']['data'][0]['bid_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['bid_iv']
                                message['params']['data'][0]['ask_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['ask_iv']
                                message['params']['data'][0]['open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['open_interest']
                                message['params']['data'][0]['old_open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['old_open_interest']
                            elif ask_price == message['params']['data'][0]['price']:
                                message['params']['data'][0]['direction'] == 'buy'
                                message['params']['data'][0]['bid_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['bid_price']
                                message['params']['data'][0]['ask_price'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['ask_price']
                                message['params']['data'][0]['bid_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['bid_iv']
                                message['params']['data'][0]['ask_iv'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['ask_iv']
                                message['params']['data'][0]['open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['open_interest']
                                message['params']['data'][0]['old_open_interest'] = orderbooks[instrument_name][len(orderbooks[instrument_name]) - 2]['old_open_interest']
                        print(message)
                        wsQueue.put(message)
        except:
            pass

【问题讨论】：

标签： python websocket python-asyncio tornado

【解决方案1】：

while-loop 继续运行，因此 ioloop 没有机会运行任何其他代码。

由于self.write_message 返回一个等待对象，您可以使用await 语句暂停循环，直到它被解决：

...
while True:
    await self.write_message(reply)
...

await 语句将控制权提交回 ioloop。

【讨论】：