4、行情处理流水线并行化设计：Python多进程实战

行情数据来得快，去得也快。单线程处理，迟早要出事。

我之前在搭建高频回测系统时，就吃过这个亏。行情一秒钟来几百笔，单进程里又是解析、又是计算、又是存储，CPU 忙得团团转，结果就是延迟越来越高，最后队列爆了。嗯，从那以后，我彻底转向了多进程架构。

说白了，行情处理流水线，就是把一个大任务拆成几个小步骤，每个步骤交给一个独立进程去跑。这样就能充分利用多核 CPU，让行情数据像流水一样顺畅通过。

4.1 multiprocessing 模块核心 API

Python 的 multiprocessing 模块，是并行化的基石。它跟 threading 不同，它能绕过 GIL 锁，真正利用多核。

我个人习惯，最常用的几个 API 是：

• Process：创建子进程，指定 target 函数
• Pool：进程池，批量管理子进程
• Queue：进程安全队列，用于传递数据
• Pipe：管道，双向通信
• Value / Array：共享内存，存简单数据

举个例子，启动一个子进程处理行情：

from multiprocessing import Process

def handle_tick(tick_data):
    # 处理单笔行情
    print(f"处理行情: {tick_data}")

if __name__ == '__main__':
    p = Process(target=handle_tick, args=('BTC-USD,100.5',))
    p.start()
    p.join()

你想想看，如果每来一笔行情就创建一个进程，那开销太大了。所以实际项目中，我们很少直接这么用。

4.2 进程池 (ProcessPoolExecutor) 的使用

进程池，就是提前创建好一批进程，等着任务来。来了就分配，干完就回收。这样避免了频繁创建销毁的开销。

我推荐用 concurrent.futures.ProcessPoolExecutor，它比 multiprocessing.Pool 的接口更现代，用起来更顺手。

from concurrent.futures import ProcessPoolExecutor

def parse_market_data(raw):
    # 解析原始行情
    return f"解析完成: {raw}"

if __name__ == '__main__':
    market_data = ["tick1", "tick2", "tick3"]
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(parse_market_data, market_data))
    print(results)

4.3 进程间通信 (Queue, Pipe)

进程之间不共享内存，所以通信得靠特殊机制。最常用的就是 Queue 和 Pipe。

Queue：生产者-消费者模式

行情处理流水线，天然就是生产者-消费者模型。数据采集进程往 Queue 里放，处理进程从 Queue 里取。

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(f"行情数据-{i}")
        print(f"生产: 行情数据-{i}")

def consumer(q):
    while True:
        data = q.get()
        if data == "STOP":
            break
        print(f"消费: {data}")

if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    q.put("STOP")
    p2.join()

Pipe：双向通道

Pipe 适合两个进程之间直接对话。比如行情网关和策略引擎之间，用 Pipe 传指令。

from multiprocessing import Process, Pipe

def gateway(conn):
    conn.send(["BTC-USD", 50000.0])
    reply = conn.recv()
    print(f"网关收到回复: {reply}")

def strategy(conn):
    data = conn.recv()
    print(f"策略收到行情: {data}")
    conn.send("已处理")

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p1 = Process(target=gateway, args=(parent_conn,))
    p2 = Process(target=strategy, args=(child_conn,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

4.4 共享内存与同步原语

有些场景下，通信太慢，不如直接共享内存。比如多个进程都要读取最新的行情快照。

multiprocessing.Value 和 Array 就是干这个的。它们存在共享内存里，所有进程都能直接读写。

from multiprocessing import Process, Value, Lock

def update_price(shared_price, lock):
    for _ in range(10):
        with lock:
            shared_price.value += 1.0

if __name__ == '__main__':
    price = Value('d', 100.0)  # 'd' 表示 double
    lock = Lock()
    processes = [Process(target=update_price, args=(price, lock)) for _ in range(4)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(f"最终价格: {price.value}")

同步原语除了 Lock，还有 Semaphore、Event、Condition 等。但在行情处理中，Lock 和 Event 最常用。Event 可以用来通知所有进程「行情来了，准备干活」。

from multiprocessing import Process, Event

def worker(event, name):
    print(f"{name} 等待行情...")
    event.wait()
    print(f"{name} 开始处理行情")

if __name__ == '__main__':
    start_event = Event()
    workers = [Process(target=worker, args=(start_event, f"进程-{i}")) for i in range(3)]
    for w in workers:
        w.start()
    print("行情到达，启动所有进程")
    start_event.set()
    for w in workers:
        w.join()

4.5 流水线并行化实战框架

好了，把上面的知识点串起来，就是一个完整的行情处理流水线。

这个架构图，说白了就是四个阶段串成一条流水线。每个阶段独立进程，通过 Queue 传递数据。如果某个阶段慢了，其他阶段不受影响——这就是并行化的威力。

嗯，到这里，行情处理流水线并行化的核心内容就讲完了。多进程不是银弹，但用对了地方，性能提升是立竿见影的。你想想看，从单进程到四进程，理论上能快四倍——当然，实际还要考虑通信开销，但至少方向是对的。