如何用更简洁的Python代码实现多线程并行运行？

你可以通过自定义辅助类或函数，实现类似你期望的简洁语法，以下是两种可行方案：

方案一：自定义Threads类（支持链式调用） #

通过封装线程列表，实现start()和join()的链式调用，完全匹配你想要的写法：

import threading

class Threads:
    def __init__(self, *targets):
        # 传入多个目标函数，批量创建线程对象
        self.threads = [threading.Thread(target=func) for func in targets]
    
    def start(self):
        # 启动所有线程，返回self以支持链式调用
        for thread in self.threads:
            thread.start()
        return self
    
    def join(self):
        # 等待所有线程执行完毕
        for thread in self.threads:
            thread.join()

# 测试代码
def test1():
    print("test1")

def test2():
    print('test2')

# 一行完成创建、启动、等待
Threads(test1, test2).start().join()

扩展支持带参数的线程函数 #

如果需要给线程函数传递参数，可以修改类的初始化逻辑：

class Threads:
    def __init__(self, *tasks):
        self.threads = []
        for task in tasks:
            # 支持传入(target,) 或 (target, args) 或 (target, args, kwargs)格式的任务
            if isinstance(task, tuple):
                target = task[0]
                args = task[1] if len(task) >=2 else ()
                kwargs = task[2] if len(task)==3 else {}
                self.threads.append(threading.Thread(target=target, args=args, kwargs=kwargs))
            else:
                self.threads.append(threading.Thread(target=task))
    
    def start(self):
        for thread in self.threads:
            thread.start()
        return self
    
    def join(self):
        for thread in self.threads:
            thread.join()

# 带参数的测试函数
def test3(name):
    print(f"test3: {name}")

# 使用示例
Threads(test1, test2, (test3, ("Alice",))).start().join()

方案二：轻量函数封装 #

如果不需要链式调用，用函数封装会更简洁直接：

import threading

def run_threads(*targets):
    threads = [threading.Thread(target=func) for func in targets]
    # 启动所有线程
    for t in threads:
        t.start()
    # 等待所有线程结束
    for t in threads:
        t.join()

# 使用示例
run_threads(test1, test2)

同样可以扩展支持参数：

def run_threads(*tasks):
    threads = []
    for task in tasks:
        if isinstance(task, tuple):
            target = task[0]
            args = task[1] if len(task)>=2 else ()
            kwargs = task[2] if len(task)==3 else {}
            threads.append(threading.Thread(target=target, args=args, kwargs=kwargs))
        else:
            threads.append(threading.Thread(target=task))
    for t in threads:
        t.start()
    for t in threads:
        t.join()

run_threads(test1, test2, (test3, ("Bob",)))

这两种方案都能大幅简化多线程的创建、启动和等待流程，根据你的需求选择即可。

内容的提问来源于stack exchange，提问作者Super Kai - Kazuya Ito