使用线程装饰器时报错'TEST_THREAD'对象不可调用的问题

错误信息 #

TypeError: 'TEST_THREAD' object is not callable

出问题的代码 #

尝试实现线程装饰器为类方法添加并发功能，代码如下：

import threading

def start_workload(NUM_THREADS):
    def wrapper(fn, *args):
        thread = []
        for i in range(*args):
            t = threading.Thread(target=fn, args=(i,))
            t.start()
            thread.append(t)
        for i in range(*args):
            thread[i].join()
    return wrapper

class TEST_THREAD(object):
    def __init__(self, *args):
        super().__init__()
        self._args = args
    @start_workload
    def print(self, threads):
        print(threads, self._args)

if __name__ == '__main__':
    test = TEST_THREAD(*list([1, 2, 3, 4, 5]))
    test.print(5)

期望实现的功能 #

希望装饰器等价于以下手动实现并发的代码：

import threading

class TEST_THREAD:
    def __init__(self, *args):
        super().__init__()
        self._args = args
    
    def print(self, threads):
        print(threads, self._args)

def start_workload(fn, num_thread):
    thread = []
    for i in range(num_thread):
        t = threading.Thread(target=fn, args=(i,))
        t.start()
        thread.append(t)
    for i in range(num_thread):
        thread[i].join()

if __name__ == '__main__':
    test = TEST_THREAD(*list([1, 2, 3, 4, 5]))
    start_workload(test.print, 5)

预期输出 #

0 (1, 2, 3, 4, 5)
1 (1, 2, 3, 4, 5)
2 (1, 2, 3, 4, 5)
3 (1, 2, 3, 4, 5)
4 (1, 2, 3, 4, 5)

问题分析与修复 #

问题根源 #

1. 带参数装饰器结构错误：start_workload被设计为带参数的装饰器，但使用@start_workload时未传参，导致装饰器逻辑混乱，最终返回的wrapper被错误绑定，调用时触发类型错误。
2. 类方法self参数未处理：装饰类方法时，未正确传递实例的self参数给线程的目标函数，导致方法无法正确绑定实例。
3. 参数解析逻辑混乱：range(*args)的用法不符合调用逻辑，装饰器整体参数传递链路错误。

修正后的代码 #

调整装饰器为三层嵌套结构，正确处理类方法的self参数，并修正线程创建逻辑：

import threading

def start_workload(num_threads=None):
    # 三层嵌套：处理装饰器参数、被装饰函数、函数调用参数
    def decorator(fn):
        def wrapper(self, *args):
            # 优先使用调用时传入的线程数，否则用装饰器指定的值
            thread_count = args[0] if args else num_threads
            threads = []
            for i in range(thread_count):
                # 类方法需要绑定self，将self作为目标方法的第一个参数
                t = threading.Thread(target=fn, args=(self, i))
                t.start()
                threads.append(t)
            # 遍历线程列表完成join，更简洁可靠
            for t in threads:
                t.join()
        return wrapper
    # 兼容两种使用方式：@start_workload 或 @start_workload(5)
    if callable(num_threads):
        return decorator(num_threads)
    return decorator

class TEST_THREAD(object):
    def __init__(self, *args):
        super().__init__()
        self._args = args
    
    @start_workload
    def print(self, thread_num):
        print(thread_num, self._args)

if __name__ == '__main__':
    test = TEST_THREAD(1, 2, 3, 4, 5)
    test.print(5)

说明 #

• 装饰器改为三层嵌套，同时支持两种使用方式：调用时传入线程数（@start_workload）或装饰时指定固定线程数（@start_workload(5)）。
• 创建线程时将self作为目标方法的第一个参数，确保类方法能正确访问实例属性。
• 线程join逻辑改为直接遍历线程列表，避免索引越界风险。

运行修正后的代码，即可得到预期输出。

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