Python多线程是并发编程中常用的实现方式，通过threading模块可以快速创建和管理多个线程，实现任务的并行执行。不过由于全局解释器锁的存在，Python多线程在CPU密集型任务上的表现有一定局限性，更适合IO密集型场景。
Python多线程核心原理
全局解释器锁GIL的作用
Python的CPython解释器中有一个全局解释器锁，它的作用是保证同一时刻只有一个线程可以执行Python字节码。这是因为CPython的内存管理不是线程安全的，GIL可以避免多个线程同时操作内存导致的数据错误。不过GIL只会影响CPU密集型任务，在IO等待时线程会释放GIL，其他线程可以获得执行权限。
线程的状态与调度
线程的生命周期包含新建、就绪、运行、阻塞、终止五个状态。Python的线程调度由操作系统负责，开发者可以通过time.sleep()、threading.Event等方式主动让出执行权，或者控制线程的阻塞和唤醒。
threading模块基础用法
threading是Python内置的多线程模块，提供了线程创建、同步、通信等核心功能，下面是常用的基础操作示例。
创建和启动线程
import threading
import time
# 定义线程执行的函数
def task(name):
print(f"线程{name}开始执行")
time.sleep(2) # 模拟IO等待，会释放GIL
print(f"线程{name}执行结束")
if __name__ == "__main__":
# 创建两个线程实例
t1 = threading.Thread(target=task, args=("A",))
t2 = threading.Thread(target=task, args=("B",))
# 启动线程
t1.start()
t2.start()
# 等待所有线程执行完成
t1.join()
t2.join()
print("所有线程执行完毕")
线程同步之锁机制
当多个线程操作共享资源时，可能会出现数据竞争问题，这时候需要使用锁来保证同一时刻只有一个线程操作共享资源。threading模块提供了Lock类实现互斥锁。
import threading
# 共享资源
count = 0
# 创建互斥锁
lock = threading.Lock()
def add_count():
global count
for _ in range(100000):
# 获取锁
lock.acquire()
try:
count += 1
finally:
# 释放锁，避免死锁
lock.release()
if __name__ == "__main__":
t1 = threading.Thread(target=add_count)
t2 = threading.Thread(target=add_count)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"最终count值为：{count}") # 输出200000，无锁时结果会小于该值
实战案例
案例1：多URL并发请求
这个场景属于IO密集型任务，使用多线程可以大幅提升请求效率，减少总等待时间。
import threading
import requests
import time
def fetch_url(url):
response = requests.get(url)
print(f"URL:{url} 状态码:{response.status_code} 内容长度:{len(response.text)}")
if __name__ == "__main__":
urls = [
"http://ipipp.com",
"http://127.0.0.1:8080",
"http://192.168.0.1"
]
start_time = time.time()
thread_list = []
for url in urls:
t = threading.Thread(target=fetch_url, args=(url,))
t.start()
thread_list.append(t)
for t in thread_list:
t.join()
end_time = time.time()
print(f"总耗时:{end_time - start_time:.2f}秒")
案例2：生产者消费者模型
该模型通过队列实现线程间的通信，生产者线程生成数据放入队列，消费者线程从队列中取出数据处理，使用Queue可以自动实现线程同步，不需要额外加锁。
import threading
import queue
import time
import random
# 创建线程安全的队列
q = queue.Queue(maxsize=5)
# 生产者函数
def producer():
for i in range(10):
item = f"产品{i}"
q.put(item)
print(f"生产了{item}，队列当前大小:{q.qsize()}")
time.sleep(random.randint(1, 3))
# 消费者函数
def consumer():
while True:
item = q.get()
if item is None:
# 收到结束信号，退出循环
break
print(f"消费了{item}")
time.sleep(random.randint(2, 4))
q.task_done()
if __name__ == "__main__":
# 创建生产者和消费者线程
p = threading.Thread(target=producer)
c = threading.Thread(target=consumer)
p.start()
c.start()
p.join()
# 发送结束信号给消费者
q.put(None)
c.join()
print("生产者消费者模型执行完毕")
多线程使用注意事项
- CPU密集型任务不建议使用Python多线程，优先选择多进程或者异步编程
- 操作共享资源时必须做好同步，避免数据竞争和死锁问题
- 线程数量不是越多越好，过多的线程会带来额外的调度开销，一般建议根据IO等待时间合理设置线程数
- 不要随意终止正在运行的线程，可能会导致资源无法正确释放