Gérer plus de connexions Websocket

Question

J'ai le code de base suivant, qui se connecte à un serveur websocket et reçoit des données:

import websocket, json, time

def process_message(ws, msg):
    message = json.loads(msg)
    print(message)

def on_error(ws, error):
    print('Error', e)

def on_close(ws):
    print('Closing')

def on_open(ws):
    def run(*args):
        Subs = []
       
        tradeStr=  """{"method": "SUBSCRIBE", "params":%s, "id": 1}"""%(json.dumps(Subs))
        ws.send(tradeStr)

    thread.start_new_thread(run, ())

def Connect():
    websocket.enableTrace(False)
    ws = websocket.WebSocketApp("wss://myurl", on_message = process_message, on_error = on_error, on_close = on_close)
    ws.on_open = on_open
    ws.run_forever()

Connect()

Maintenant, je voudrais créer plus de connexions vers différents serveurs et recevoir des données simultanément dans le même script. J'ai essayé ce qui suit:

def run(url):

    def process_message(ws, msg):
        message = json.loads(msg)
        print(message)

    def on_error(ws, error):
        print('Error', e)

    def on_close(ws):
        print('Closing')

    def on_open(ws):
        def run(*args):
            Subs = []
           
            tradeStr=  """{"method": "SUBSCRIBE", "params":%s, "id": 1}"""%(json.dumps(Subs))
            ws.send(tradeStr)

        thread.start_new_thread(run, ())

    def Connect():
        websocket.enableTrace(False)
        ws = websocket.WebSocketApp(url, on_message = process_message, on_error = on_error, on_close = on_close)
        ws.on_open = on_open
        ws.run_forever()

    Connect()

threading.Thread(target=run, kwargs={'url': 'url1'}).start()
threading.Thread(target=run, kwargs={'url': 'url2'}).start()
threading.Thread(target=run, kwargs={'url': 'url3'}).start()

Maintenant, ce code fonctionne, mais je me connecte à différentes URL et je diffuse les données de chacune d'entre elles, mais cela m'a semblé une solution "hacky". Je ne sais pas non plus si ce que je fais pourrait être une mauvaise pratique ou non. Chaque connexion enverra environ 600/700 petits dictionnaires JSON, et je dois mettre à jour chaque enregistrement dans la base de données.

Ma question est donc la suivante: cette implémentation est-elle correcte? Puisqu'il fonctionne avec des threads, peut-il créer des problèmes à long terme? Dois-je faire une autre bibliothèque telle que Tornado?

Answer 1

Bienvenue dans la communauté de révision de code. Peu de réflexions initiales, lors de l'écriture de code python; suivre le guide de style PEP-8 rend votre code plus maintenable. Ce qui reviendrait à (mais sans s'y limiter):

1. Fonctions et variables nommées dans lower_snake_case.
2. Éviter les espaces dans les paramètres / arguments de fonction.

---

En continuant, vous pouvez placer votre code de gestion de websocket dans son propre sous-module. Pour par exemple.

class WebSocket:
    def __init__(self, url):
        self._url = url

    def close(self, ws):
        # handle closed connection callback
        pass

    def connect(self):
        self.ws = websocket.WebSocketApp(self._url, on_close=self.close)
        self.ws.on_open = self.open  # for eg.
        self.ws.run_forever()
    .
    .
    .

Si vous voulez qu'il prenne en charge le threading, vous pouvez étendre cette classe avec un threading.Threadsous-classement (vous pouvez également choisir de sous-classer multiprocessing.Processplus tard):

class ThreadedWebSocket(WebSocket, threading.Thread):
    def run(self):
        # This simply calls `self.connect()`

de cette façon, si vous souhaitez initialiser les threads:

url1_thread = ThreadedWebSocket(url=url1)
url1_thread.start()
url1_thread.join()

Le sous-classement est principalement une question de préférence.

---

Dans le on_openrappel, vous videz un dict dans json, puis vous l'insérez dans un modèle de chaîne. Ce qui suit permet d'obtenir le même résultat et est plus intuitif (imo):

def on_open(ws):
    def run(*args):
        subs = []
        trade =  {
            "method": "SUBSCRIBE",
            "params": subs,
            "id": 1
        }
        ws.send(json.dumps(trade))

---

Pour en venir à votre question sur les alternatives de bibliothèque et les performances, les E / S réseau sont généralement un appel bloquant. C'était l'une des principales raisons pour lesquelles j'ai suggéré d'aller avec le websocketspaquet. L'autre être; nous utilisons des websockets dans notre organisation depuis plus d'un an en production. Et cela a fonctionné de manière vraiment impressionnante pour nous, gérant un débit de quelques Go de données chaque jour avec une seule machine de bas niveau.

Answer 2

C'est la même approche que vous avez utilisée, juste réécrite comme une approche oop par pacmaninbw avec une classe socketConnhéritant dewebsocket.WebSocketApp

Parce que la on_message, on_erroret les on_closefonctions que vous définissez sont si court / simple , vous pouvez les mettre directement dans la surcharge des fonctions d' initialisation lambda au lieu d'écrire chacun d' eux séparément et de les transmettre sous forme de paramètres

De nouvelles connexions se forment avec ws = SocketConn('url'). Vous devriez également faire une boucle sur une liste ou un ensemble d'URL si vous vous connectez à plusieurs (ce que vous faites probablement dans votre code réel, mais cela vaut la peine d'être noté au cas où)

import websocket, json, threading, _thread

class SocketConn(websocket.WebSocketApp):
    def __init__(self, url):
        #WebSocketApp sets on_open=None, pass socketConn.on_open to init
        super().__init__(url, on_open=self.on_open)

        self.on_message = lambda ws, msg: print(json.loads(msg))
        self.on_error = lambda ws, e: print('Error', e)
        self.on_close = lambda ws: print('Closing')

        self.run_forever()

    def on_open(self, ws):
        def run(*args):
            Subs = []
            
            tradeStr={"method":"SUBSCRIBE", "params":Subs, "id":1}
            ws.send(json.dumps(tradeStr))

        _thread.start_new_thread(run, ())



if __name__ == '__main__':
    websocket.enableTrace(False)

    urls = {'url1', 'url2', 'url3'}


    for url in urls:
        # args=(url,) stops the url from being unpacked as chars
        threading.Thread(target=SocketConn, args=(url,)).start()