Python多线程返回值混淆问题求助及代码示例

你的多线程代码出现数据串混，核心原因是彭博blp.bdp接口不是线程安全的。多线程同时调用时，API内部的共享会话状态被两个线程交叉修改，导致返回的数据错误混入另一个线程的结果集里。另外你的my_func里把pd.concat放在循环内属于无效操作，会额外消耗性能。

修复方案 #

方案1：每个线程使用独立的彭博会话 #

给每个线程创建专属的API会话，避免共享状态：

from threading import Thread
import pandas as pd
import blpapi

class ThreadWithReturnValue(Thread):
    def __init__(self, group=None, target=None, name=None,
                args=(), kwargs=None, Verbose=None):
        # 修复可变默认参数的陷阱
        kwargs = kwargs or {}
        Thread.__init__(self, group, target, name, args, kwargs)
        self._return = None
    def run(self):
        if self._target is not None:
            self._return = self._target(*self._args, **self._kwargs)
    def join(self, *args):
         Thread.join(self, *args)
         return self._return

def my_func(str_lst, num, freq):
    # 每个线程初始化独立的彭博会话
    session = blpapi.Session()
    session.start()
    df_list = []
    for i, y, f in zip(str_lst, num, freq):
        # 使用当前线程的会话调用bdp（需确认bdp支持传入session参数，参考彭博API文档）
        data = blp.bdp(i, flds=['xxx'], y, f, session=session)
        df_list.append(data)
    # 循环结束后再拼接DataFrame，提升性能
    final_df = pd.concat(df_list)
    session.stop()
    return final_df

# 线程调用逻辑不变
Thread1 = ThreadWithReturnValue(target=my_func, args=(str_a, num_lst_a,freq_lst_a))
Thread2 = ThreadWithReturnValue(target=my_func, args=(str_b, num_lst_b,freq_lst_b))

Thread1.start()
Thread2.start()

output1 = Thread1.join()
output2 = Thread2.join()

方案2：用线程锁串行化API调用 #

如果彭博API不支持自定义会话，就用锁确保同一时间只有一个线程调用bdp：

from threading import Thread, Lock
import pandas as pd

# 全局锁，控制bdp调用的串行化
blp_api_lock = Lock()

class ThreadWithReturnValue(Thread):
    def __init__(self, group=None, target=None, name=None,
                args=(), kwargs=None, Verbose=None):
        kwargs = kwargs or {}
        Thread.__init__(self, group, target, name, args, kwargs)
        self._return = None
    def run(self):
        if self._target is not None:
            self._return = self._target(*self._args, **self._kwargs)
    def join(self, *args):
         Thread.join(self, *args)
         return self._return

def my_func(str_lst, num, freq):
    df_list = []
    for i, y, f in zip(str_lst, num, freq):
        # 调用bdp前加锁，避免线程冲突
        with blp_api_lock:
            data = blp.bdp(i, flds=['xxx'], y, f)
        df_list.append(data)
    final_df = pd.concat(df_list)
    return final_df

# 线程调用逻辑不变
Thread1 = ThreadWithReturnValue(target=my_func, args=(str_a, num_lst_a,freq_lst_a))
Thread2 = ThreadWithReturnValue(target=my_func, args=(str_b, num_lst_b,freq_lst_b))

Thread1.start()
Thread2.start()

output1 = Thread1.join()
output2 = Thread2.join()

额外说明 #

• 方案1能保留多线程的并行效率，是最优解；方案2会让API调用变成串行，效率略低，但实现简单。
• 修复了ThreadWithReturnValue中kwargs={}的可变默认参数问题，避免潜在的共享字典bug。

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