Python多线程编程实战：从基础概念到线程通信机制详解

Chloe

发布时间：2026-10-13| 5人浏览过

原创

本文深入解析Python多线程编程，探讨GIL对并行的限制及异步与同步的选择。详细讲解threading模块中Lock、Rlock、Condition和Event的核心用法，通过代码示例演示线程同步、数据竞争处理及线程间通信的具体实现，帮助开发者掌握高效的多线程开发技巧。

Python多线程基础与GIL限制

前言：受限于全局解释器锁（GIL），Python中的线程无法实现真正的并行计算，其性能与异步编程相比存在显著差距。然而，鉴于异步编程逻辑复杂、学习与维护成本较高，大多数开发者仍倾向于使用同步编码模式。除非涉及高性能处理场景，否则多线程仍是许多应用的首选方案。

首先普及下进程和线程的概念：

进程：进程是操作系统资源分配的基本单位。

线程：线程是任务调度和执行的基本单位。

一个应用程序至少一个进程，一个进程至少一个线程。

两者区别：同一进程内的线程共享本进程的资源如内存、I/O、cpu等，但是进程之间的资源是独立的。

多线程基本用法：start与join

一、多线程

python 可以通过 thread 或 threading 模块实现多线程，threading 相比 thread 提供了更高阶、更全面的线程管理。我们下文主要以 threading 模块介绍多线程的基本用法。


import threading
import time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)

 def run(self):
 print('%s:Now timestamp is %s'%(self.name,time.time()))

threads = []
for a in range(int(5)): # 线程个数
 threads.append(thread(str(a)))
for t in threads: # 开启线程
 t.start()
for t in threads: # 阻塞线程
 t.join()
print('END')

输出：线程3:Now timestamp is 1557386184.7574518 线程2:Now timestamp is 1557386184.7574518 线程0:Now timestamp is 1557386184.7574518 线程1:Now timestamp is 1557386184.7574518 线程4:Now timestamp is 1557386184.7582724 END

start() 方法开启子线程。运行多次 start() 方法代表开启多个子线程。

join() 方法用来阻塞主线程，等待子线程执行完成。举个例子，主线程A创建了子线程B，并使用了 join() 方法，主线程A在 join() 处就被阻塞了，等待子线程B完成后，主线程A才能执行 print('END')。如果没有使用 join() 方法，主线程A创建子线程B后，不会等待子线程B，直接执行 print('END')，如下：


import threading
import time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)

 def run(self):
 time.sleep(1)
 print('%s:Now timestamp is %s'%(self.name,time.time()))

threads = []
for a in range(int(5)): # 线程个数
 threads.append(thread(str(a)))
for t in threads: # 开启线程
 t.start()
# for t in threads: # 阻塞线程
# t.join()
print('END')

输出： END 线程0:Now timestamp is 1557386321.376941 线程3:Now timestamp is 1557386321.377937 线程1:Now timestamp is 1557386321.377937 线程2:Now timestamp is 1557386321.377937 线程4:Now timestamp is 1557386321.377937

线程间通信与同步机制

二、线程之间的通信

使用Lock解决数据竞争

1.threading.Lock()

如果多个线程对某一资源同时进行修改，可能会存在不可预知的情况。为了修改数据的正确性，需要把这个资源锁住，只允许线程依次排队进去获取这个资源。当线程A操作完后，释放锁，线程B才能进入。如下脚本是开启多个线程修改变量的值，但输出结果每次都不一样。


import threading

money = 0
def Order(n):
 global money
 money = money + n
 money = money - n

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 for i in range(1000000):
  Order(self.threadname)

t1 = thread('1')
t2 = thread('5')
t1.start()
t2.start()
t1.join()
t2.join()
print(money)

接下来我们用 threading.Lock() 锁住这个变量，等操作完再释放这个锁。lock.acquire() 给资源加一把锁，对资源处理完成之后，lock.release() 再释放锁。以下脚本执行结果都是一样的，但速度会变慢，因为线程只能一个个的通过。


import threading

money = 0
def Order(n):
 global money
 money = money + n
 money = money - n

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 for i in range(1000000):
  lock.acquire()
  Order(self.threadname)
  lock.release()
# print('%s:Now timestamp is %s'%(self.name,time.time()))

lock = threading.Lock()
t1 = thread('1')
t2 = thread('5')
t1.start()
t2.start()
t1.join()
t2.join()
print(money)

ReentrantLock与Condition高级同步

2.threading.Rlock()

用法和 threading Lock() 一致，区别是 threading.Rlock() 允许多次锁资源，acquire() 和 release() 必须成对出现，也就是说加了几把锁就得释放几把锁。


lock = threading.Lock()
# 死锁
lock.acquire()
lock.acquire()
print('...')
lock.release()
lock.release()

rlock = threading.RLock()
# 同一线程内不会阻塞线程
rlock.acquire()
rlock.acquire()
print('...')
rlock.release()
rlock.release()

3.threading.Condition()

threading.Condition() 可以理解为更加高级的锁，比 Lock 和 Rlock 的用法更高级，能处理一些复杂的线程同步问题。threading.Condition() 创建一把资源锁（默认是Rlock），提供 acquire() 和 release() 方法，用法和 Rlock 一致。此外 Condition 还提供 wait()、Notify() 和 NotifyAll() 方法。

wait()：线程挂起，直到收到一个 Notify() 通知或者超时（可选参数），wait() 必须在线程得到 Rlock 后才能使用。

Notify() ：在线程挂起的时候，发送一个通知，让 wait() 等待线程继续运行，Notify() 也必须在线程得到 Rlock 后才能使用。 Notify(n=1)，最多唤醒 n 个线程。

NotifyAll() ：在线程挂起的时候，发送通知，让所有 wait() 阻塞的线程都继续运行。

举例说明下 Condition() 使用


import threading,time

def TestA():
 cond.acquire()
 print('李白：看见一个敌人，请求支援')
 cond.wait()
 print('李白：好的')
 cond.notify()
 cond.release()

def TestB():
 time.sleep(2)
 cond.acquire()
 print('亚瑟：等我...')
 cond.notify()
 cond.wait()
 print('亚瑟：我到了，发起冲锋...')

if __name__=='__main__':
 cond = threading.Condition()
 testA = threading.Thread(target=TestA)
 testB = threading.Thread(target=TestB)
 testA.start()
 testB.start()
 testA.join()
 testB.join()

输出李白：看见一个敌人，请求支援亚瑟：等我... 李白：好的亚瑟：我到了，发起冲锋...

利用Event实现线程通知

4.threading.Event()

threading.Event() 原理是在线程中立了一个 Flag ，默认值是 False ，当一个或多个线程遇到 event.wait() 方法时阻塞，直到 Flag 值变为 True 。threading.Event() 通常用来实现线程之间的通信，使一个线程等待其他线程的通知，把 Event 传递到线程对象中。

event.wait() ：阻塞线程，直到 Flag 值变为 True

event.set() ：设置 Flag 值为 True

event.clear() ：修改 Flag 值为 False

event.isSet() : 仅当 Flag 值为 True 时返回

下面这个例子，主线程启动子线程后 sleap 2秒，子线程因为 event.wait() 被阻塞。当主线程醒来后执行 event.set() ，子线程才继续运行，两者输出时间差 2s。


import threading
import datetime,time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 event.wait()
 print('子线程运行时间：%s'%datetime.datetime.now())

if __name__ == '__main__':
 event = threading.Event()
 t1 = thread('0')
 #启动子线程
 t1.start()
 print('主线程运行时间：%s'%datetime.datetime.now())
 time.sleep(2)
 # Flag设置成True
 event.set()
 t1.join()