线程化的、非阻塞的websocket客户端

24

我希望能够在Python中运行一个程序,通过WebSockets每秒发送一条消息到Tornado服务器。我一直在使用websocket-client上的例子;

这个例子不起作用,因为ws.run_forever()会停止while循环的执行。

有人可以给我提供一个示例,说明如何正确实现这个作为线程类,我既可以调用发送方法,也可以接收消息吗?

import websocket
import thread
import time

def on_message(ws, message):
    print message

def on_error(ws, error):
    print error

def on_close(ws):
    print "### closed ###"

def on_open(ws):
    pass

if __name__ == "__main__":
    websocket.enableTrace(True)
    ws = websocket.WebSocketApp("ws://echo.websocket.org/", on_message = on_message, on_error = on_error, on_close = on_close)
    ws.on_open = on_open
    ws.run_forever()

    while True:
        #do other actions here... collect data etc.
        for i in range(100):
            time.sleep(1)
            ws.send("Hello %d" % i)
        time.sleep(1)
2个回答

32

他们的Github页面中有一个例子,正好做到了这一点。似乎你是以那个例子为起点,从on_open函数中拿出每秒发送消息的代码,并将其粘贴在run_forever调用之后,这个调用会一直运行,直到套接字断开连接。

也许你对基本概念存在一些问题。总会有一个线程专门用于监听套接字(在这种情况下,是进入run_forever内部循环等待消息的主线程)。如果你想要进行其他操作,就需要另一个线程。

下面是他们示例代码的另一版本,其中创建了另一个线程作为“套接字监听器”,run_forever在那里运行。我认为它稍微复杂了一些,因为你必须编写代码来确保套接字已连接,而你本可以使用on_open回调函数,但也许它会帮助你理解。

import websocket
import threading
from time import sleep

def on_message(ws, message):
    print message

def on_close(ws):
    print "### closed ###"

if __name__ == "__main__":
    websocket.enableTrace(True)
    ws = websocket.WebSocketApp("ws://echo.websocket.org/", on_message = on_message, on_close = on_close)
    wst = threading.Thread(target=ws.run_forever)
    wst.daemon = True
    wst.start()

    conn_timeout = 5
    while not ws.sock.connected and conn_timeout:
        sleep(1)
        conn_timeout -= 1

    msg_counter = 0
    while ws.sock.connected:
        ws.send('Hello world %d'%msg_counter)
        sleep(1)
        msg_counter += 1

整个服务器只在一个线程中运行?如果这个线程变得饱和了怎么办? - Paul
@Paul:你在这里所说的“saturated”是什么意思? - igauravsehrawat
2
@Paul - 在没有等待的情况下,线程不会提高效率;归根结底,GIL确保您仍在运行一个线程。 - Michael Ward
2
Python显然不支持多线程。如果你想处理大量的连接,你应该使用asyncio和multiprocessing将消息交给其他进程来处理。当你想向客户端返回结果时,你可以将它放入一个回复队列中,由主要的asyncio进程接收并返回。 - Erik Aronesty
谢谢你提醒我将线程指定为守护线程。 - Vaidøtas I.
显示剩余2条评论

0
在2023年,他们将使用类似于rel的异步调度程序更新dispatching multiple WebSocketApps的示例。
import websocket, rel

addr = "wss://api.gemini.com/v1/marketdata/%s"
for symbol in ["BTCUSD", "ETHUSD", "ETHBTC"]:
    ws = websocket.WebSocketApp(addr % (symbol,), on_message=lambda w, m : print(m))
    ws.run_forever(dispatcher=rel, reconnect=3)  
rel.signal(2, rel.abort)  # Keyboard Interrupt  
rel.dispatch()

希望能对你有所帮助!


1
不行,它仍然阻塞! - Sunding Wei

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接