如何在新WebSocket进程运行时关闭旧进程以实现快速重连并更新URL

核心思路是通过进程间同步事件，让新进程在WebSocket连接成功并准备好接收数据后，通知主线程再终止旧进程，避免出现数据中断的空档期。

具体实现步骤 #

• 用multiprocessing.Manager().Event()创建跨进程的就绪信号，用于新进程告知主线程自身已就绪。
• 给start_stream_currency函数添加就绪信号参数，在WebSocket连接成功的on_open回调中触发该信号。
• 主线程启动新进程后，等待就绪信号触发，确认新进程正常运行后，再终止旧进程并清理无用的交易对数据。

修改后的完整代码 #

from functools import partial
from threading import Thread
from multiprocessing import Process, Manager, Event
import time
import json
import websocket
import requests

def on_message(_wsa, result, prices):
    result = json.loads(result)
    prices[result['s']]['best_bid'] = result['b']
    prices[result['s']]['best_ask'] = result['a']

def on_open(_wsa, ready_event):
    # WebSocket连接成功时触发，标记新进程就绪
    ready_event.set()

def start_stream_currency(curr_list, prices, ready_event):
    url = 'wss://fstream.binance.com/ws'
    for curr in curr_list:
        url += '/' + curr.lower() + '@bookTicker'
    
    # 绑定on_open回调，传递就绪信号
    wsa = websocket.WebSocketApp(
        url,
        on_message=partial(on_message, prices=prices),
        on_open=partial(on_open, ready_event=ready_event)
    )

    # 初始化交易对的价格存储
    for curr in curr_list:
        prices[curr] = Manager().dict()

    wsa.run_forever()

def check_price_symbols(prices): # 检查WebSocket推送的价格
    while True:
        for key, value in prices.copy().items():
            print(f"{key} {value}")
            # 这里放你的计算逻辑
        time.sleep(1)

def main():
    prices = Manager().dict()
    th = Thread(target=check_price_symbols, args=(prices,))
    th.daemon = True  # 设置为守护线程，主进程退出时自动结束
    th.start()

    # 初始启动第一个进程
    curr_list = ['BTCUSDT','ADAUSDT']
    pr = Process(target=start_stream_currency, args=(curr_list, prices, Event()))
    pr.start()

    time.sleep(15)
    print('开始切换交易对流')
    curr_list2 = ['ADAUSDT','XRPUSDT']
    
    # 创建新进程的就绪信号
    new_ready = Manager().Event()
    pr2 = Process(target=start_stream_currency, args=(curr_list2, prices, new_ready))
    pr2.start()

    # 等待新进程就绪（WebSocket连接成功）
    new_ready.wait()
    print('新进程已就绪，终止旧进程')
    
    # 终止旧进程并清理旧交易对
    pr.terminate()
    pr.join()  # 等待旧进程完全退出，避免残留写入操作
    if 'BTCUSDT' in prices:
        del prices['BTCUSDT']

    # 这里可以继续运行，或者等待pr2结束
    pr2.join()

if __name__ =='__main__':
    main()

关键细节说明 #

• 就绪信号的触发时机：选择on_open回调是因为它在WebSocket连接成功建立后立即触发，此时新进程已经可以接收Binance的推送数据，是最准确的就绪标记。
• 进程终止后的清理：调用pr.terminate()后必须调用pr.join()，确保旧进程完全退出，避免它继续写入prices字典导致数据冲突。
• 守护线程设置：给检查价格的线程设置daemon=True，主进程退出时该线程会自动结束，无需手动处理。

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