使用multiprocessing并行获取FTX订单簿失败问题排查

我尝试同时从FTX交易所的WebSocket服务器两个通道提取加密货币对BTC/USD和ZRX/USD的orderbook（订单簿）数据。WebSocket会先发送orderbook初始快照，之后发送更新消息，我需要在收到每条消息时应用更新。我计划用字典存储orderbook，通过multiprocessing开启两个Process分别处理两个WebSocket连接，并使用Manager创建asks和bids两个内存共享字典，以嵌套字典形式存储两个币种的买卖盘数据。但实际运行时，两个进程并未并行执行，extract_data1启动后，extract_data2需手动终止extract_data1才会启动，代码如下：

from multiprocessing import Process, Manager
import websocket, json
import time


def on_open(ws, market):
    print('opened connection')
    subscribe_message = {'op': 'subscribe', 'channel': 'orderbook', 'market': market}
    print(subscribe_message)
    ws.send(json.dumps(subscribe_message))


def init_orderbook(ws, asks, bids, market, asks_snapshot, bids_snapshot):
    proxy_asks, proxy_bids = asks, bids
    for level in asks_snapshot:
        proxy_asks[level[0]] = level[1]
    for level in bids_snapshot:
        proxy_bids[level[0]] = level[1]
    asks, bids = proxy_asks, proxy_bids


def apply_changes(ws, asks, bids, market, asks_snapshot, bids_snapshot):
    proxy_asks, proxy_bids = asks, bids
    for level in asks_snapshot:
        if level[1] == 0:
            del proxy_asks[level[0]]
        else:
            proxy_asks[level[0]] = level[1]
    for level in bids_snapshot:
        if level[1] == 0:
            del proxy_bids[level[0]]
        else:
            proxy_bids[level[0]] = level[1]
    asks, bids = proxy_asks, proxy_bids


def on_message(ws, message, asks, bids):
    js = json.loads(message)
    if js['type'] == 'subscribed':
        print('Subscribed, ', js)
    elif js['type'] == 'update':
        market = js['market']
        update_asks = js['data']['asks']
        update_bids = js['data']['asks']
        apply_changes(ws, asks[market], bids[market], market, update_asks, update_bids)
    elif js['type'] == 'partial':
        print('Storing snapshot, ', js)
        market = js['market']
        update_asks = js['data']['asks']
        update_bids = js['data']['asks']
        init_orderbook(ws, asks[market], bids[market], market, update_asks, update_bids)

def get_orderbook(market, asks, bids):
    socket = "wss://ftx.com/ws/"
    ws = websocket.WebSocketApp(socket)
    ws.on_open = lambda *x: on_open(*x, market)
    ws.on_message = lambda ws, msg: on_message(ws, msg, asks, bids)
    ws.run_forever()

if __name__ == '__main__':
    manager = Manager()
    asks = manager.dict()
    bids = manager.dict()

    asks['ZRX/USD'] = manager.dict()
    bids['ZRX/USD'] = manager.dict()
    asks['BTC/USD'] = manager.dict()
    bids['BTC/USD'] = manager.dict()

    extract_data1 = Process(target=get_orderbook, args=('ZRX/USD', asks, bids))
    extract_data2 = Process(target=get_orderbook, args=('BTC/USD', asks, bids))

    extract_data1.run()
    extract_data2.run()

    extract_data1.join()
    extract_data2.join()

问题原因与修正方案 #

核心错误：进程启动方式错误 #

你使用了Process.run()方法，这个方法是在当前主进程中直接执行目标函数，不会创建新的子进程。因为get_orderbook里的ws.run_forever()是无限循环，extract_data1.run()会一直阻塞主进程，导致extract_data2.run()永远无法执行。

正确的做法是使用Process.start()方法，它会创建独立的子进程并启动目标函数，这样两个进程才能并行运行。

其他次要错误 #

1. 买卖盘数据引用错误：在on_message的update和partial分支中，你把js['data']['bids']错误写成了js['data']['asks']，导致买卖盘都用了asks的数据，订单簿逻辑完全错误。
2. 无用的字典代理赋值：init_orderbook和apply_changes里的proxy_asks, proxy_bids = asks, bids以及最后的asks, bids = proxy_asks, proxy_bids都是多余的，字典是引用类型，直接操作传入的共享字典即可。

修正后的代码 #

from multiprocessing import Process, Manager
import websocket, json
import time


def on_open(ws, market):
    print('opened connection')
    subscribe_message = {'op': 'subscribe', 'channel': 'orderbook', 'market': market}
    print(subscribe_message)
    ws.send(json.dumps(subscribe_message))


def init_orderbook(asks, bids, asks_snapshot, bids_snapshot):
    for level in asks_snapshot:
        asks[level[0]] = level[1]
    for level in bids_snapshot:
        bids[level[0]] = level[1]


def apply_changes(asks, bids, asks_snapshot, bids_snapshot):
    for level in asks_snapshot:
        if level[1] == 0:
            del asks[level[0]]
        else:
            asks[level[0]] = level[1]
    for level in bids_snapshot:
        if level[1] == 0:
            del bids[level[0]]
        else:
            bids[level[0]] = level[1]


def on_message(ws, message, asks, bids):
    js = json.loads(message)
    if js['type'] == 'subscribed':
        print('Subscribed, ', js)
    elif js['type'] == 'update':
        market = js['market']
        update_asks = js['data']['asks']
        update_bids = js['data']['bids']  # 修正为bids
        apply_changes(asks[market], bids[market], update_asks, update_bids)
    elif js['type'] == 'partial':
        print('Storing snapshot, ', js)
        market = js['market']
        update_asks = js['data']['asks']
        update_bids = js['data']['bids']  # 修正为bids
        init_orderbook(asks[market], bids[market], update_asks, update_bids)

def get_orderbook(market, asks, bids):
    socket = "wss://ftx.com/ws/"
    ws = websocket.WebSocketApp(socket)
    ws.on_open = lambda *x: on_open(*x, market)
    ws.on_message = lambda ws, msg: on_message(ws, msg, asks, bids)
    ws.run_forever()

if __name__ == '__main__':
    manager = Manager()
    asks = manager.dict()
    bids = manager.dict()

    asks['ZRX/USD'] = manager.dict()
    bids['ZRX/USD'] = manager.dict()
    asks['BTC/USD'] = manager.dict()
    bids['BTC/USD'] = manager.dict()

    extract_data1 = Process(target=get_orderbook, args=('ZRX/USD', asks, bids))
    extract_data2 = Process(target=get_orderbook, args=('BTC/USD', asks, bids))

    # 改用start()启动子进程
    extract_data1.start()
    extract_data2.start()

    extract_data1.join()
    extract_data2.join()

内容的提问来源于stack exchange，提问作者apt45