Python线程启动案例如何开启线程

wen python案例 2026-06-30 27

本文目录导读：

Python线程启动案例如何开启线程

直接使用threading.Thread
继承Thread类
使用类方法作为线程函数
使用ThreadPoolExecutor（推荐）
带参数和返回值的线程
守护线程（Daemon Thread）
重要注意事项

我来为您介绍Python中开启线程的几种常见方式：

直接使用`threading.Thread`

import threading
import time
def worker(name):
    """线程执行的函数"""
    for i in range(3):
        print(f"线程 {name}: 第{i+1}次执行")
        time.sleep(1)
# 创建并启动线程
t1 = threading.Thread(target=worker, args=("Thread-1",))
t2 = threading.Thread(target=worker, args=("Thread-2",))
t1.start()  # 启动线程
t2.start()
# 等待线程结束
t1.join()
t2.join()
print("所有线程执行完毕")

继承`Thread`类

import threading
import time
class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name
    def run(self):
        """重写run方法"""
        for i in range(3):
            print(f"线程 {self.name}: 第{i+1}次执行")
            time.sleep(1)
# 创建并启动线程
t1 = MyThread("Thread-1")
t2 = MyThread("Thread-2")
t1.start()
t2.start()
t1.join()
t2.join()

使用类方法作为线程函数

import threading
class TaskManager:
    def worker(self, task_id):
        print(f"执行任务 {task_id}")
    @classmethod
    def class_worker(cls, task_name):
        print(f"执行类任务 {task_name}")
    @staticmethod
    def static_worker(data):
        print(f"处理数据: {data}")
# 实例方法
mgr = TaskManager()
t1 = threading.Thread(target=mgr.worker, args=(1,))
# 类方法
t2 = threading.Thread(target=TaskManager.class_worker, args=("Task-A",))
# 静态方法
t3 = threading.Thread(target=TaskManager.static_worker, args=("data",))
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()

使用`ThreadPoolExecutor`（推荐）

from concurrent.futures import ThreadPoolExecutor
import time
def task(n):
    print(f"执行任务 {n}")
    time.sleep(1)
    return f"任务 {n} 完成"
# 创建线程池
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务
    futures = [executor.submit(task, i) for i in range(5)]
    # 获取结果
    for future in futures:
        print(future.result())

带参数和返回值的线程

import threading
import time
def calculate_square(numbers, results, index):
    """计算平方并存储结果"""
    result = []
    for num in numbers:
        time.sleep(0.5)
        result.append(num ** 2)
    results[index] = result
# 准备数据
data_sets = [
    [1, 2, 3],
    [4, 5, 6],
    [7, 8, 9]
]
results = [None] * len(data_sets)
threads = []
# 创建并启动多个线程
for i, data in enumerate(data_sets):
    t = threading.Thread(target=calculate_square, args=(data, results, i))
    threads.append(t)
    t.start()
# 等待所有线程完成
for t in threads:
    t.join()
print("计算结果:", results)

守护线程（Daemon Thread）

import threading
import time
def daemon_task():
    """守护线程任务"""
    while True:
        print("守护线程正在运行...")
        time.sleep(1)
def normal_task():
    """普通线程任务"""
    for i in range(3):
        print(f"普通线程执行中... ({i+1}/3)")
        time.sleep(1)
# 创建守护线程
daemon = threading.Thread(target=daemon_task, daemon=True)
daemon.start()
# 创建普通线程
normal = threading.Thread(target=normal_task)
normal.start()
normal.join()  # 等待普通线程结束
print("主程序结束，守护线程自动退出")

重要注意事项

线程安全：多线程访问共享资源时需要加锁

import threading
lock = threading.Lock()
shared_counter = 0
def safe_increment():
    global shared_counter
    with lock:  # 使用上下文管理器加锁
        shared_counter += 1

线程数量：不要创建过多线程，通常使用ThreadPoolExecutor管理
GIL限制：Python的GIL（全局解释器锁）会限制CPU密集型任务的并行性能

避免竞争条件：

# 不安全的写法
counter = 0
def unsafe():
    global counter
    counter += 1  # 可能发生竞态条件
# 安全的写法
counter = 0
lock = threading.Lock()
def safe():
    global counter
    with lock:
        counter += 1

选择哪种方式取决于您的具体需求：