如何在Python子类中实现ThreadPoolExecutor多线程功能？

我有类嵌套结构，希望在子类showit中启用多线程功能，使用@threaded装饰器时出现错误，提示'showit'对象不可迭代，tp.map认为没有传入列表；但在外部使用ThreadPoolExecutor调用test_first方法时没有问题，现在想在类内实现该功能，让程序打印出列表中的每个数字。

复现代码 #

from concurrent.futures import ThreadPoolExecutor
from typing import List

id_str = ['1', '2', '3', '4', '5']
id_int = [1, 2, 3, 4, 5]

def threaded(fn, pools=10):
    tp = ThreadPoolExecutor(pools)
    def wrapper(*args):
        return tp.map(fn, *args)  # returns Future object
    return wrapper

class testit:
    def __init__(self, some_list: List[str]) -> None:
        self._ids = some_list
        print(self._ids)

    def test_first(self, some_id: List[int]) -> None:
        print(some_id)

class showit(testit):
    def __init__(self, *args):
        super(showit, self).__init__(*args)
    
    @threaded
    def again(self):
        global id_int
        for values in self.test_first(id_int):
            print(values)

a = showit(id_str)
print(a.again())

错误信息 #

File "test_1.py", line 32, in <module>
    print(a.again())
  File "test_1.py", line 10, in wrapper
    return tp.map(fn, *args)  # returns Future object
  File "/Users/usr/opt/anaconda3/lib/python3.8/concurrent/futures/_base.py", line 600, in map
    fs = [self.submit(fn, *args) for args in zip(*iterables)]
TypeError: 'showit' object is not iterable

预期输出 #

1
2
3
4
5

问题原因 #

装饰类方法时，wrapper函数接收的第一个参数是类实例self，tp.map会把这个实例当成要迭代的参数，直接触发报错。另外原代码中test_first仅打印整个列表，未返回可迭代对象，again方法的循环逻辑也不符合多线程处理单个元素的需求。

修正后的代码 #

from concurrent.futures import ThreadPoolExecutor
from typing import List

id_str = ['1', '2', '3', '4', '5']
id_int = [1, 2, 3, 4, 5]

def threaded(fn, pools=10):
    tp = ThreadPoolExecutor(pools)
    def wrapper(*args, **kwargs):
        # 分离类实例与待迭代参数
        self = args[0]
        iterable = args[1]
        # 通过lambda将实例传递给目标方法，确保多线程处理单个元素
        return tp.map(lambda x: fn(self, x), iterable)
    return wrapper

class testit:
    def __init__(self, some_list: List[str]) -> None:
        self._ids = some_list
        print(self._ids)

    def test_first(self, some_id: int) -> None:
        # 修改为接收单个元素并打印
        print(some_id)

class showit(testit):
    def __init__(self, *args):
        super().__init__(*args)
    
    @threaded
    def again(self, iterable):
        pass

a = showit(id_str)
# 传入待处理的列表
a.again(id_int)

关键修改说明 #

1. 装饰器内分离了类实例self和待迭代参数，用lambda确保目标方法能拿到实例对象和单个元素；
2. 调整test_first方法，使其接收单个元素而非整个列表，适配多线程逐个处理的逻辑；
3. again方法仅需接收待处理的可迭代对象，装饰器自动完成多线程调度。

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