基于Python的Bloomberg //blp/mkdata高性能流式数据采集方案问询

一、核心优化方向：解耦消息接收与业务处理 #

BLPAPI底层已维护网络IO线程池，但如果在事件循环中直接阻塞处理业务逻辑，会导致消息队列积压、程序无响应。核心优化思路是将消息接收（BLPAPI事件循环线程）与数据处理（业务逻辑）分离，用多线程/进程池异步处理业务，避免阻塞事件循环。

二、Demo代码的多线程改造步骤 #

1. 基于生产者-消费者模式的消息处理 #

用queue.Queue缓存推送消息，配合concurrent.futures.ThreadPoolExecutor异步处理业务，确保事件循环不被阻塞：

import blpapi
import time
from concurrent.futures import ThreadPoolExecutor
from queue import Queue

# 缓存待处理消息的队列，限制大小防止内存溢出
msg_queue = Queue(maxsize=1500)
# 线程池大小根据CPU核心数和业务复杂度调整
executor = ThreadPoolExecutor(max_workers=4)

def process_message(msg):
    """业务逻辑处理：解析字段、暂存数据等"""
    sec_data = msg.getElement("securityData")
    security = sec_data.getElement("security").getValueAsString()
    field_data = sec_data.getElement("fieldData")
    
    # 示例：提取字段值
    last_price = field_data.getElementAsFloat("LAST_PRICE") if field_data.hasElement("LAST_PRICE") else None
    # 这里可以将数据暂存到内存列表，等待批量写入

def event_loop(session):
    """BLPAPI事件循环：接收消息并投递到线程池"""
    while True:
        # 设置超时，避免无限阻塞
        event = session.nextEvent(500)
        for msg in event:
            if event.eventType() == blpapi.Event.SUBSCRIPTION_DATA:
                # 提交消息到线程池异步处理，不阻塞事件循环
                executor.submit(process_message, msg)
            # 处理订阅状态、会话状态等事件
            elif event.eventType() in [blpapi.Event.SUBSCRIPTION_STATUS, blpapi.Event.SESSION_STATUS]:
                pass

def main():
    # 初始化会话（复用Demo的配置逻辑）
    session_options = blpapi.SessionOptions()
    session_options.setServerHost("localhost")
    session_options.setServerPort(8194)
    
    session = blpapi.Session(session_options)
    if not session.start():
        print("会话启动失败")
        return

    # 构建100个标的+20个字段的订阅列表
    target_securities = ["T US Equity", "MSFT US Equity"] + [f"{i} US Equity" for i in range(98)]
    target_fields = ["LAST_PRICE", "BID", "ASK", "VOLUME"] + [f"FIELD_{i}" for i in range(16)]
    subscriptions = [blpapi.Subscription(sec, target_fields) for sec in target_securities]
    
    session.subscribe(subscriptions)
    # 启动事件循环
    event_loop(session)

if __name__ == "__main__":
    main()

2. 5分钟定时任务的实现 #

用独立线程处理定时逻辑，避免干扰事件循环：

import threading

def scheduled_task(session):
    """5分钟定时任务：更新订阅+批量存储数据"""
    while True:
        time.sleep(300)  # 5分钟间隔
        # 1. 查询数据库获取需新增的标的
        new_securities = query_database_for_new_secs()  # 替换为你的数据库查询逻辑
        if new_securities:
            new_subs = [blpapi.Subscription(sec, target_fields) for sec in new_securities]
            session.subscribe(new_subs)
        # 2. 批量将暂存的流式数据写入数据库/文件
        batch_save_to_storage()  # 替换为你的批量存储逻辑

# 在main函数中添加定时线程启动逻辑
def main():
    # ... 会话初始化、订阅代码 ...
    # 启动定时任务线程（设置为守护线程，随主进程退出）
    threading.Thread(target=scheduled_task, args=(session,), daemon=True).start()
    event_loop(session)

三、SDK自带的高性能示例参考 #

Bloomberg Windows SDK的\blpapi\examples\Python目录下有两个关键示例：

• SubscriptionExample：基础订阅流程示例，展示标准的事件循环写法
• AdvancedSubscriptionExample：高吞吐量场景优化示例，包含消息批量处理、非阻塞事件循环的最佳实践，完全适配100+标的的订阅需求

四、额外性能优化建议 #

• 批量处理IO：不要单条数据写入数据库，而是将数据暂存到内存列表，累计到一定数量或到达5分钟节点时批量写入，减少IO开销
• 监控队列积压：定时打印msg_queue.qsize()，如果队列持续增长，说明处理速度不足，需要增加线程池大小或优化业务逻辑
• CPU密集型场景用进程池：如果数据处理包含复杂计算，改用ProcessPoolExecutor替代线程池，规避Python GIL限制
• 设置消息丢弃策略：当队列满时，可选择丢弃旧消息或阻塞生产者，根据业务优先级调整

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