如何使用多处理订阅多个 Websocket 流

Question

我不熟悉在 python 中处理多处理、多线程等。

我正在尝试使用 multiprocessing 从我的加密交换 (API Docs Here) 订阅多个 Websocket 流。 但是，当我运行下面的代码时，我只收到ticker information，而不是order book updates。

如何修复代码以获取这两个信息？
在multiprocessing 上运行时似乎只有一个 websocket 工作的原因是什么？

（当我分别运行ws_orderBookUpdates() 和ws_tickerInfo() 函数时，不使用multiprocessing，它单独运行良好，因此不是交易所的问题。）

import websocket
import json
import pprint
from datetime import datetime
import time

# Function to subscribe to ticker information.
def ws_tickerInfo():
    def on_open(self):
        print("opened")
        subscribe_message = {
            "method": "subscribe",
            "params": {'channel': "lightning_ticker_BTC_JPY"}
        }
        ws.send(json.dumps(subscribe_message))

    def on_message(self, message, prev=None):
        print(f"Ticker Info, Received : {datetime.now()}")

        ###### full json payloads ######
        # pprint.pprint(json.loads(message))

    def on_close(self):
        print("closed connection")

    endpoint = 'wss://ws.lightstream.bitflyer.com/json-rpc'
    ws = websocket.WebSocketApp(endpoint,
                                on_open=on_open,
                                on_message=on_message,
                                on_close=on_close)

    ws.run_forever()


# Function to subscribe to order book updates.
def ws_orderBookUpdates():
    def on_open(self):
        print("opened")
        subscribe_message = {
            "method": "subscribe",
            "params": {'channel': "lightning_board_BTC_JPY"}
        }
        ws.send(json.dumps(subscribe_message))

    def on_message(self, message):
        print(f"Order Book, Received : {datetime.now()}")

        ###### full json payloads ######
        # pprint.pprint(json.loads(message))

    def on_close(self):
        print("closed connection")

    endpoint = 'wss://ws.lightstream.bitflyer.com/json-rpc'
    ws = websocket.WebSocketApp(endpoint,
                                on_open=on_open,
                                on_message=on_message,
                                on_close=on_close)
    ws.run_forever()


# Multiprocessing two functions
if __name__ == '__main__':
    import multiprocessing as mp

    mp.Process(target=ws_tickerInfo(), daemon=True).start()
    mp.Process(target=ws_orderBookUpdates(), daemon=True).start()

Answer 1

更新

您已经创建了两个 daemon 进程。当所有非守护进程进程终止时，它们将终止，在这种情况下，主进程是在创建守护进程后立即终止。您很幸运，即使其中一个过程也有机会产生输出，但为什么要冒险呢？ 不要使用 dameon 进程。而是：

if __name__ == '__main__':
    import multiprocessing as mp

    p1 = mp.Process(target=ws_tickerInfo)
    p2 = mp.Process(target=ws_orderBookUpdates)
    p1.start()
    p2.start()
    p1.join() # wait for completion
    p2.join() # wait for completion

但真正的问题是盯着我们的脸，我们都错过了！你有：

    p1 = mp.Process(target=ws_tickerInfo(), daemon=True)
    p2 = mp.Process(target=ws_orderBookUpdates(), daemon=True)

应该是什么时候：

    p1 = mp.Process(target=ws_tickerInfo)
    p2 = mp.Process(target=ws_orderBookUpdates)

看到区别了吗？实际上，您并没有将函数 ws_tickerInfo 传递给 Process，而是 调用 ws_tickerInfo 并尝试传递返回值，如果函数曾经返回，这将是荒谬的 None（它没有）。因此，您甚至都无法执行第二个流程创建语句。

您可能也为此使用了多线程而不是多处理，尽管 Ctrl-C 中断处理程序可能不起作用（见下文）。还应该有一种终止程序的机制。我添加了一些代码来检测 Ctrl-C 并在输入时终止。此外，您使用self 作为函数参数，就好像该函数实际上是一个类方法，但事实并非如此。这不是好的编程风格。这是更新的来源：

import websocket
import json
import pprint
from datetime import datetime
import time
import sys
import signal

# Function to subscribe to ticker information.
def ws_tickerInfo():
    def on_open(wsapp):
        print("opened")
        subscribe_message = {
            "method": "subscribe",
            "params": {'channel': "lightning_ticker_BTC_JPY"}
        }
        wsapp.send(json.dumps(subscribe_message))

    def on_message(wsapp, message, prev=None):
        print(f"Ticker Info, Received : {datetime.now()}")

        ###### full json payloads ######
        # pprint.pprint(json.loads(message))

    def on_close(wsapp):
        print("closed connection")

    endpoint = 'wss://ws.lightstream.bitflyer.com/json-rpc'
    ws = websocket.WebSocketApp(endpoint,
                                on_open=on_open,
                                on_message=on_message,
                                on_close=on_close)

    ws.run_forever()


# Function to subscribe to order book updates.
def ws_orderBookUpdates():
    def on_open(wsapp):
        print("opened")
        subscribe_message = {
            "method": "subscribe",
            "params": {'channel': "lightning_board_BTC_JPY"}
        }
        wsapp.send(json.dumps(subscribe_message))

    def on_message(wsapp, message):
        print(f"Order Book, Received : {datetime.now()}")

        ###### full json payloads ######
        # pprint.pprint(json.loads(message))

    def on_close(wsapp):
        print("closed connection")

    endpoint = 'wss://ws.lightstream.bitflyer.com/json-rpc'
    ws = websocket.WebSocketApp(endpoint,
                                on_open=on_open,
                                on_message=on_message,
                                on_close=on_close)
    ws.run_forever()

def handle_ctrl_c(signum, stack_frame):
    sys.exit(0)

if __name__ == '__main__':
    import multiprocessing as mp

    signal.signal(signal.SIGINT, handle_ctrl_c) # terminate on ctrl-c
    print('Enter Ctrl-C to terminate.')
    p1 = mp.Process(target=ws_tickerInfo)
    p2 = mp.Process(target=ws_orderBookUpdates)
    p1.start()
    p2.start()
    p1.join() # wait for completion (will never happen)
    p2.join() # wait for completion (will never happen)

使用多线程

if __name__ == '__main__':
    import threading

    t1 = threading.Thread(target=ws_tickerInfo, daemon=True)
    t2 = threading.Thread(target=ws_orderBookUpdates, daemon=True)
    t1.start()
    t2.start()
    input('Hit enter to terminate...\n')