【python 内置模块】threading

传入目标函数创建多线程

不使用多线程示例，会顺序执行，总共运行时间 5.014s

import time

def task(name, delay):
    print(f"{name}开始执行，延迟{delay}秒")
    time.sleep(delay)  # 模拟I/O操作
    print(f"{name}执行完成")

if __name__ == "__main__":
    task("A", 3)
    task("B", 2)

# 输出:
# A开始执行，延迟2秒
# A执行完成
# B开始执行，延迟1秒
# B执行完成

使用多线程示例，会并发执行，总共运行时间 3.0013s

import threading
import time

def task(name, delay):
    print(f"线程{name}开始执行，延迟{delay}秒")
    time.sleep(delay)  # 模拟I/O操作
    print(f"线程{name}执行完成")

if __name__ == "__main__":
    # 创建线程
    t1 = threading.Thread(target=task, args=("A", 3))
    t2 = threading.Thread(target=task, args=("B", 2))
    
    # 启动线程
    t1.start()
    t2.start()
    
    # 等待线程完成
    t1.join()
    t2.join()
    print("所有线程执行完毕")

上述示例中用到了两个线程方法：

• start() 启用线程，仅能调用一次；
• join(timeout=None) 等待线程结束，可设置超时时间；

import threading
import time

class MyThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__()  # 必须调用父类构造方法
        self.name = name
        self.delay = delay
    
    def run(self):
        print(f"自定义线程{self.name}启动，延迟{self.delay}秒")
        time.sleep(self.delay)
        print(f"自定义线程{self.name}结束")

if __name__ == "__main__":
    t1 = MyThread("C", 3)
    t2 = MyThread("D", 1)
    
    t1.start()
    t2.start()

    t1.join()
    t2.join()
    
    print("自定义线程全部执行完毕")