1. 协议解析的并发模型：从单线程到多线程

做量化交易这些年，我见过太多团队在协议解析上栽跟头。说白了，交易所的行情数据就像高压水枪，你接不住，后面所有策略都是白搭。

今天咱们聊聊最基础的问题：为什么交易所协议解析需要并发？

1.1 单线程的困境：你扛得住每秒10万笔订单吗？

先看一个真实场景。假设你只用一个线程去解析行情数据：

def parse_tick(data):
    # 解析一条行情数据
    symbol = data[0:8]
    price = struct.unpack('>d', data[8:16])[0]
    volume = struct.unpack('>I', data[16:20])[0]
    return Tick(symbol, price, volume)

while True:
    raw_data = socket.recv(4096)
    tick = parse_tick(raw_data)
    process_tick(tick)  # 交给策略处理

这段代码看起来没问题，对吧？但我在项目中遇到过，当行情爆发时，单线程的瓶颈会暴露无遗。

举个例子。上期所的CTP协议，行情快照每秒推送2-4次，每次包含5000+只合约的深度数据。单次解析耗时约0.5ms，算下来每秒要花2ms在解析上。听起来不多？

但别忘了，还有网络I/O、内存拷贝、策略计算……这些加起来，单线程的CPU很快就跑满了。更可怕的是，如果某次解析卡住了（比如数据包不完整），整个系统都会阻塞。

1.2 为什么需要并发？三个核心原因

你想想看，交易所的数据流有几个特点：

• 高吞吐：深交所的行情峰值可达每秒50万笔
• 低延迟：很多策略要求在100微秒内完成解析
• 多路复用：你可能同时连接多个交易所

单线程根本应付不来。我总结了一下，并发带来的好处主要有三点：

维度	单线程	多线程
CPU利用率	一个核跑满，其他核闲置	多核并行，利用率高
延迟抖动	一个慢操作拖累全局	独立线程隔离，互不影响
扩展性	加机器才能提升	单机内可扩展

1.3 并发模型怎么选？三种主流方案

搞清楚了为什么需要并发，接下来就是怎么实现。我见过三种主流方案：

方案一：生产者-消费者模型

这是最经典的做法。一个线程收数据，多个线程解析：

from queue import Queue
from threading import Thread

data_queue = Queue(maxsize=1000)

def receiver():
    while True:
        raw = socket.recv(4096)
        data_queue.put(raw)

def parser(worker_id):
    while True:
        raw = data_queue.get()
        tick = parse_tick(raw)
        process_tick(tick)

# 启动1个接收线程 + 4个解析线程
Thread(target=receiver).start()
for i in range(4):
    Thread(target=parser, args=(i,)).start()

这个模型的好处是简单，但要注意队列的容量。我曾经遇到过队列满了导致接收线程阻塞，结果丢包了……

方案二：无锁环形缓冲区

追求极致性能时，可以用无锁队列。比如用collections.deque或者第三方库pyringbuffer：

from ringbuffer import RingBuffer

# 创建容量为1024的环形缓冲区
buf = RingBuffer(1024, dtype='uint8')

def receiver():
    while True:
        raw = socket.recv(4096)
        buf.push(raw)

def parser():
    while True:
        if buf.available > 0:
            raw = buf.pop()
            tick = parse_tick(raw)
            process_tick(tick)

无锁的好处是延迟更低，但实现起来要小心ABA问题。我个人建议，除非你对性能有极致要求（比如纳秒级），否则用标准库的Queue就够了。

方案三：异步I/O + 线程池

Python的asyncio配合线程池，也是个不错的选择：

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def handle_data(raw):
    loop = asyncio.get_event_loop()
    # 把CPU密集的解析任务丢到线程池
    tick = await loop.run_in_executor(executor, parse_tick, raw)
    process_tick(tick)

async def receiver():
    reader = await asyncio.open_connection(host, port)
    while True:
        raw = await reader.read(4096)
        asyncio.create_task(handle_data(raw))

这个方案适合I/O密集型的场景。但要注意，Python的GIL会限制线程池的并行度。说白了，如果你有8个核，线程池设4个就差不多了。

1.4 并发模型的核心逻辑图

下面这张图，是我自己总结的并发模型选择逻辑。你可以根据实际情况对号入座：

这张图的核心逻辑很简单：延迟敏感用无锁，高吞吐用生产者-消费者，一般场景用异步I/O。我在实际项目中，90%的情况都用生产者-消费者模型，因为它够简单、够稳定。

1.5 线程安全：并发编程的命门

聊并发，绕不开线程安全。我见过太多人把共享变量直接丢给多线程，结果数据全乱了。

举个典型的反例：

# 错误的做法：共享变量不加锁
shared_tick = None

def parser():
    global shared_tick
    while True:
        raw = data_queue.get()
        shared_tick = parse_tick(raw)  # 多个线程同时写

def strategy():
    global shared_tick
    while True:
        if shared_tick:
            # 这里读到的数据可能已经被其他线程修改了
            process_tick(shared_tick)

正确的做法是：每个线程有自己的独立数据副本，或者用线程安全的队列传递数据。

1.6 实战建议：从单线程到多线程的迁移路径

如果你现在还是单线程，想改成多线程，我建议分三步走：

1. 先做性能分析：用cProfile跑一下，看看瓶颈到底在哪。有时候是网络I/O，有时候是解析逻辑本身。
2. 从生产者-消费者开始：这是最稳妥的方案。把接收和解析拆成两个线程，看看效果。
3. 逐步增加并发度：先加到2个解析线程，测一下延迟和吞吐。没问题再加到4个、8个。

我记得有个团队，一上来就搞了16个解析线程，结果锁竞争导致性能还不如单线程。嗯，步子迈大了容易扯着蛋。

好了，这一章的内容就到这里。并发模型的选择没有银弹，关键是根据你的业务场景做权衡。下一章咱们聊聊具体的协议解析细节，比如怎么处理粘包、怎么校验数据完整性——这些都是实战中绕不开的坑。

---