Python多线程获取websocket实时数据流运行卡顿无响应如何解决？

核心卡顿原因 #

• 共享资源无锁竞争：定义的threading.Lock仅包裹了线程启动和join逻辑，完全未作用于三个全局DataFrame的读写操作。websocket线程写数据、策略线程读数据时出现竞态条件，极端情况下会触发索引越界、数据异常直接阻塞程序。
• 策略线程死循环空转：run_strategy的while True循环未添加任何休眠逻辑，会占满单个CPU核心，导致系统调度优先级降低、程序响应卡顿。
• DataFrame append性能损耗：pandas的append方法为非原地操作，每次新增数据都会全量复制整个DataFrame，数据量积累后操作耗时会指数级上升，直接阻塞websocket消息接收。
• 启动时索引越界：程序启动后websocket还未收到第一条数据时，就直接执行iloc[-1]取最新值，会触发索引错误导致线程静默崩溃，最终程序无响应。

修复步骤 #

• 调整锁的使用逻辑：所有读写全局共享变量（三个DataFrame/替代的队列）的位置都加锁，避免竞态条件。
• 策略循环添加休眠：在run_strategy的while循环中添加至少0.05~0.1秒的休眠，大幅降低CPU占用。
• 替换DataFrame为高性能队列：用collections.deque存储实时数据，双端队列的追加和弹出操作均为O(1)时间复杂度，性能远高于pandas append，需要统计分析时再转成DataFrame即可。
• 增加非空判断：计算价差前先判断现货、期货队列都至少有一条数据，避免启动时报错。
• 修正asyncio写法：移除废弃的@asyncio.coroutine装饰器，用asyncio.gather替代asyncio.wait适配新版本websockets库。

核心修改后代码示例 #

import time 
import threading
import pandas as pd 
import json
import asyncio
import websockets
import datetime
from collections import deque

lock = threading.Lock()

# Define websocket address
SOCKET_SPOT = 'wss://stream.binance.com:9443/ws/btcusdt@trade'
SOCKET_FUTURES =  'wss://fstream.binance.com/ws/btcusdt_210625@trade'

connections = set([SOCKET_SPOT, SOCKET_FUTURES])

# 用deque替代DataFrame存储实时数据，性能提升百倍
max_len = 2000
q_spot = deque(maxlen=max_len)
q_futures = deque(maxlen=max_len)
q_results = deque(maxlen=500)

async def handle_socket(uri):
    global q_spot, q_futures, q_results
    async with websockets.connect(uri) as websocket:
        async for message in websocket:
            data = json.loads(message)
            ts = datetime.utcfromtimestamp(data['T']/1000).strftime('%Y-%m-%d %H:%M:%S.%f')
            price = float(data['p'])
            
            # 写共享变量前加锁
            with lock:
                if uri == SOCKET_SPOT:
                    q_spot.append({'time': ts, 'price_s': price})
                else:
                    q_futures.append({'time': ts, 'price_f': price})
                
                # 两个队列都有数据再计算价差
                if len(q_spot) == 0 or len(q_futures) == 0:
                    continue
                
                pct_diff = ((q_futures[-1]['price_f'] - q_spot[-1]['price_s']) / q_spot[-1]['price_s']) * 100
                q_results.append({'time': q_futures[-1]['time'], 'pct_diff': pct_diff})
 
async def handler():
    await asyncio.gather(*[handle_socket(uri) for uri in connections])

def loop_in_thread(loop):
    asyncio.set_event_loop(loop)
    loop.run_until_complete(handler())

loop = asyncio.new_event_loop() # 避免主线程event loop冲突
t_1 = threading.Thread(target=loop_in_thread, args=(loop,), daemon=True)

# Generator for producing signal
def signal(ut, lt):
    current_state = "Outside market"
    while True:
        pct_change = yield current_state
        if (
            current_state in ("Outside market", "Market exit")
            and pct_change >= ut
        ):
            current_state = "Entered market"
        elif current_state == "Entered market" and pct_change > lt:
            current_state = "Inside market"
        elif current_state == "Market exit" and pct_change < ut:
            current_state = "Outside market"
        elif (
            current_state in ("Entered market", "Inside market")
            and pct_change <= lt
        ):
            current_state = "Market exit"

UT = 0.85
LT = 0.70

s = signal(UT, LT)
next(s)

# Run function 
def run_strategy():
    global q_results
    last_process = None
    while True:
        time.sleep(0.1) # 降CPU占用
        # 读共享变量前加锁
        with lock:
            if len(q_results) == 0:
                continue
            # 避免重复处理同一条数据
            latest = q_results[-1]
            if latest['time'] == last_process:
                continue
            last_process = latest['time']
            pct_diff = latest['pct_diff']
        sig = s.send(pct_diff)
        print(sig)
   
t_2 = threading.Thread(target=run_strategy, args=(), daemon=True)

t_1.start()
t_2.start()
t_1.join()
t_2.join()

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