Python多线程案例如何创建线程

本文目录导读：

1. 使用threading.Thread直接创建线程
2. 继承Thread类创建线程
3. 带返回值的线程（使用队列）
4. 线程同步（使用锁）
5. 完整的多线程下载模拟案例
6. 关键知识点总结
7. 注意事项

我来为您详细介绍Python多线程的创建方法及案例。

使用threading.Thread直接创建线程

创建Thread实例（最常用）

import threading
import time
def worker(name, delay):
    """线程要执行的任务"""
    for i in range(3):
        print(f"线程 {name}: 正在执行第 {i+1} 次")
        time.sleep(delay)
    print(f"线程 {name}: 完成")
# 创建并启动线程
thread1 = threading.Thread(target=worker, args=("Thread-1", 1))
thread2 = threading.Thread(target=worker, args=("Thread-2", 1.5))
# 启动线程
thread1.start()
thread2.start()
# 等待线程结束
thread1.join()
thread2.join()
print("所有线程执行完毕")

使用args参数传递数据

import threading
import time
def calculate_square(numbers):
    """计算数字的平方"""
    for num in numbers:
        result = num * num
        print(f"线程 {threading.current_thread().name}: {num}的平方 = {result}")
        time.sleep(0.5)
def calculate_cube(numbers):
    """计算数字的立方"""
    for num in numbers:
        result = num * num * num
        print(f"线程 {threading.current_thread().name}: {num}的立方 = {result}")
        time.sleep(0.5)
# 数据
numbers = [1, 2, 3, 4, 5]
# 创建线程
square_thread = threading.Thread(
    target=calculate_square, 
    args=(numbers,), 
    name="Square-Thread"
)
cube_thread = threading.Thread(
    target=calculate_cube, 
    args=(numbers,), 
    name="Cube-Thread"
)
# 启动线程
square_thread.start()
cube_thread.start()
# 等待线程完成
square_thread.join()
cube_thread.join()

继承Thread类创建线程

import threading
import time
class MyThread(threading.Thread):
    def __init__(self, name, delay, count):
        super().__init__()
        self.name = name
        self.delay = delay
        self.count = count
    def run(self):
        """重写run方法，线程启动时自动执行"""
        for i in range(self.count):
            print(f"{self.name}: 计数 {i+1}")
            time.sleep(self.delay)
        print(f"{self.name}: 完成")
# 创建自定义线程对象
thread1 = MyThread("线程A", 0.5, 5)
thread2 = MyThread("线程B", 1.0, 3)
# 启动线程
thread1.start()
thread2.start()
# 等待线程结束
thread1.join()
thread2.join()
print("所有自定义线程执行完毕")

带返回值的线程（使用队列）

import threading
import queue
import time
def worker_with_result(task_queue, result_queue):
    """处理任务并返回结果"""
    while True:
        try:
            # 从任务队列获取任务
            task = task_queue.get(timeout=1)
        except queue.Empty:
            break
        # 执行任务
        result = task * task
        time.sleep(0.5)
        # 将结果放入结果队列
        result_queue.put((task, result))
        task_queue.task_done()
# 创建任务和结果队列
task_queue = queue.Queue()
result_queue = queue.Queue()
# 添加任务
for i in range(10):
    task_queue.put(i)
# 创建多个工作线程
threads = []
for i in range(3):
    thread = threading.Thread(
        target=worker_with_result, 
        args=(task_queue, result_queue),
        name=f"Worker-{i+1}"
    )
    thread.start()
    threads.append(thread)
# 等待所有任务完成
task_queue.join()
# 等待所有线程结束
for thread in threads:
    thread.join()
# 收集结果
results = []
while not result_queue.empty():
    task, result = result_queue.get()
    results.append((task, result))
    print(f"任务 {task} 的结果: {result}")
print(f"所有结果: {results}")

线程同步（使用锁）

import threading
import time
# 共享资源
counter = 0
counter_lock = threading.Lock()
def increment_with_lock():
    """使用锁保护共享资源"""
    global counter
    for _ in range(100000):
        with counter_lock:
            counter += 1
def increment_without_lock():
    """不使用锁，可能导致数据竞争"""
    global counter
    for _ in range(100000):
        # 注意：这里没有使用锁！
        counter += 1
# 测试带锁的线程
counter = 0
threads = []
for i in range(5):
    thread = threading.Thread(target=increment_with_lock)
    thread.start()
    threads.append(thread)
for thread in threads:
    thread.join()
print(f"使用锁后的计数器值: {counter}")  # 应该等于 500000

完整的多线程下载模拟案例

import threading
import time
import random
class DownloadManager:
    def __init__(self):
        self.downloads = []
        self.lock = threading.Lock()
    def download_file(self, file_name, file_size):
        """模拟文件下载"""
        downloaded = 0
        while downloaded < file_size:
            # 模拟下载速度
            chunk = random.randint(10, 50)
            time.sleep(random.uniform(0.1, 0.3))
            downloaded += chunk
            if downloaded > file_size:
                downloaded = file_size
            # 使用锁更新进度
            progress = (downloaded / file_size) * 100
            with self.lock:
                print(f"{file_name}: {progress:.1f}% [{downloaded}/{file_size} MB]")
        print(f"{file_name}: 下载完成！")
        return file_name
    def start_downloads(self):
        """启动多个下载任务"""
        # 模拟下载任务
        download_tasks = [
            ("Video.mp4", 200),
            ("Document.pdf", 50),
            ("Music.mp3", 20),
            ("Image.jpg", 10)
        ]
        threads = []
        start_time = time.time()
        # 为每个下载任务创建线程
        for file_name, file_size in download_tasks:
            thread = threading.Thread(
                target=self.download_file,
                args=(file_name, file_size)
            )
            threads.append(thread)
            thread.start()
        # 等待所有下载完成
        for thread in threads:
            thread.join()
        end_time = time.time()
        print(f"\n所有文件下载完成！总耗时: {end_time - start_time:.2f} 秒")
# 运行下载管理器
if __name__ == "__main__":
    manager = DownloadManager()
    manager.start_downloads()

关键知识点总结

注意事项

1. GIL限制：Python的全局解释器锁（GIL）限制了CPU密集型任务的多线程效率
2. I/O密集型任务：多线程适合网络请求、文件读写等I/O操作
3. 线程安全：操作共享变量时必须使用锁
4. 避免死锁：小心使用多个锁，避免循环等待

这些示例涵盖了Python多线程的主要使用模式,您可以根据实际需求选择合适的方式。