Потоки могут быть запущены только один раз в Django каналах

Question

Я создал простой Django Потребитель каналов, который должен подключаться к внешнему источнику, извлекать данные и отправлять их клиенту. Итак, пользователь открывает страницу> потребитель подключается к внешней службе и получает данные> данные отправляются на веб-сокет.

Вот мой код:

import json
from channels.generic.websocket import WebsocketConsumer, AsyncConsumer, AsyncJsonWebsocketConsumer

from binance.client import Client
import json
from binance.websockets import BinanceSocketManager
import time
import asyncio

client = Client('', '')

trades = client.get_recent_trades(symbol='BNBBTC')
bm = BinanceSocketManager(client)
class EchoConsumer(AsyncJsonWebsocketConsumer):


    async def connect(self):
        await self.accept()
        await self.send_json('test')


        bm.start_trade_socket('BNBBTC', self.process_message)
        bm.start()


    def process_message(self, message):
        JSON1 = json.dumps(message)
        JSON2 = json.loads(JSON1)

        #define variables
        Rate = JSON2['p']
        Quantity = JSON2['q']
        Symbol = JSON2['s']
        Order = JSON2['m']

        asyncio.create_task(self.send_json(Rate))
        print(Rate)

Этот код работает, когда я открываю одну страницу; если я попытаюсь открыть новое окно с новой учетной записью, он выдаст следующую ошибку:

File "C:\Users\User\Desktop\Heroku\github\master\myapp\consumers.py", line 54, in connect
    bm.start()
  File "C:\Users\User\lib\threading.py", line 843, in start
    raise RuntimeError("threads can only be started once")
  threads can only be started once

Я новичок в каналах, так что это нубский вопрос, но как я могу это исправить? Эта проблема? Я хотел сделать следующее: пользователь открывает страницу и получает данные, другой пользователь открывает страницу и получает данные; нет ли способа сделать это? Или я просто неправильно понимаю, как работает Django Каналы и веб-сокеты?

Answer 1

Здесь start () Запускает активность потока.

Он должен вызываться не более одного раза для объекта потока. Вы создали глобальный объект BinanceSocketManager как «bm».

Он всегда вызывает RuntimeError, если вызывается более одного раза для одного и того же объекта потока.

Пожалуйста, обратитесь к приведенному ниже коду, это может помочь вам

from channels.generic.websocket import WebsocketConsumer, AsyncConsumer, AsyncJsonWebsocketConsumer

from binance.client import Client
import json
from binance.websockets import BinanceSocketManager
import time
import asyncio


class EchoConsumer(AsyncJsonWebsocketConsumer):
    client = Client('', '')

    trades = client.get_recent_trades(symbol='BNBBTC')
    bm = BinanceSocketManager(client)

    async def connect(self):
        await self.accept()
        await self.send_json('test')


        self.bm.start_trade_socket('BNBBTC', self.process_message)
        self.bm.start()


    def process_message(self, message):
        JSON1 = json.dumps(message)
        JSON2 = json.loads(JSON1)

        #define variables
        Rate = JSON2['p']
        Quantity = JSON2['q']
        Symbol = JSON2['s']
        Order = JSON2['m']

        asyncio.create_task(self.send_json(Rate))
        print(Rate)

Answer 2

Вам действительно нужен вторичный поток?

class EchoConsumer(AsyncJsonWebsocketConsumer):

    symbol = ''

    async def connect(self):
        self.symbol = 'BNBBTC'
        # or, more probably, retrieve the value for "symbol" from query_string
        # so the client can specify which symbol he's interested into:
        #    socket = new WebSocket("ws://.../?symbol=BNBBTC");
        await self.accept()

    def process_message(self, message):
        # PSEUDO-CODE BELOW !
        if self.symbol == message['symbol']:
            await self.send({
                'type': 'websocket.send',
                'text': json.dumps(message),
            })

Для дополнительной гибкости вы можете также принять все списки символов от клиента:

//HTML
socket = new WebSocket("ws://.../?symbols=XXX,YYY,ZZZ");

затем (в потребитель):

class EchoConsumer(AsyncJsonWebsocketConsumer):

    symbols = []

    async def connect(self):
        # here we need to parse "?symbols=XXX,YYY,ZZZ" ...
        # the code below has been stolen from another project of mine and should be suitably adapted
        params = urllib.parse.parse_qs(self.scope.get('query_string', b'').decode('utf-8'))
        try:
            self.symbols = json.loads(params.get('symbols', ['[]'])[0])
        except:
            self.symbols = []

    def process_message(self, message):
        if message['symbol'] in self.symbols:
            ...

Answer 3

Я не Django разработчик, но если я правильно понимаю, функция connect вызывается более одного раза - и bm.start ссылается на один и тот же поток, скорее всего, сделанный в bm.start_trade_socket (или где-то еще в соединении). В заключение, когда вызывается bm.start, запускается поток, а когда это делается снова, вы получаете эту ошибку.