Python多线程互斥锁与异步通信机制详解

Evan

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

原创

本文详解Python中互斥锁（Lock）的创建、锁定与释放操作，对比不同加锁策略对性能的影响，并阐述同步执行顺序与异步通信机制。通过代码示例说明如何避免资源竞争，优化多线程环境下的数据一致性，并介绍基于进程池的异步回调实现方式。

当多个线程需要访问共享数据时，需先对资源进行锁定，此时资源状态标记为“锁定”，其他线程无法修改；直至持有锁的线程释放资源，状态转为“非锁定”，其他线程方可再次获取锁。互斥锁确保同一时刻仅有一个线程执行写入操作，从而保障多线程环境下数据的一致性。

采用f_flag方法的执行效率较低。

创建锁对象。

使用mutex=threading.Lock()语句创建锁实例。

执行锁定操作。

调用mutex.acquire([blocking])，参数blocking可选，指定等待时间；若不传参，线程将无限期阻塞等待。

执行释放操作。

调用mutex.release()释放锁。

import threading 
from threading import Thread 
from threading import Lock 
import time 
 
thnum=0 
#两个线程都在抢着对这个锁进行上锁，如果有一方成功上锁，那么导致另外一方会堵塞（一直等待），到这个锁被解开为之 
class MyThread(threading.Thread): 
  def run(self): 
    mutex.acquire() 
    for i in range(10000): 
      global thnum 
      thnum+=1   
    print(thnum) 
    mutex.release()  
def test(): 
  global thnum 
  mutex.acquire() #等待可以上锁，通知而不是轮训，没有占用CPU 
  for i in range(10000): 
    thnum+=1 
  print(thnum) 
  mutex.release()#解锁 
mutex=Lock() 
if __name__=='__main__': 
  t=MyThread() 
  t.start() 
 
#创建一把互斥锁，默认是没有上锁的 
 
thn=Thread(target=test) 
thn.start() 
 
''''' 
10000 
20000 
'''

一旦上锁，多任务并发执行转为单任务串行执行，相当于当前仅有一个线程在运行。

相较于前文示例，下述代码中加锁的时间开销更短。

import threading 
from threading import Thread 
from threading import Lock 
import time 
 
thnum=0 
#两个线程都在抢着对这个锁进行上锁，如果有一方成功上锁，那么导致另外一方会堵塞（一直等待），到这个锁被解开为之 
class MyThread(threading.Thread): 
  def run(self): 
    for i in range(10000): 
      mutex.acquire() 
      global thnum 
      thnum+=1 
      mutex.release()#释放后，都开始抢，这样上锁的时间变短  
    print(thnum) 
     
def test(): 
  global thnum 
  for i in range(10000): 
    mutex.acquire() 
    thnum+=1 
    mutex.release()#解锁 
  print(thnum) 
mutex=Lock() 
if __name__=='__main__': 
  t=MyThread() 
  t.start() 
 
#创建一把互斥锁，默认是没有上锁的 
 
thn=Thread(target=test) 
thn.start() 
 
''''' 
10000 
20000 
'''

仅在必须保证数据一致性的关键代码段加锁。

同步：指任务按照预定的先后顺序依次执行。

执行流程为：一个任务运行完毕后释放锁，下一个任务获取锁后运行，再释放锁，如此循环，直至第一个任务最终释放锁。

异步：指任务无需等待前一个任务完成即可提交或执行，通常通过回调机制处理结果。

#异步的实现 
from multiprocessing import Pool 
import time 
import os 
 
#getpid()获取当前进程的进程号 
#getppid()获取当前进程的父进程号 
 
def test():#子进程 
  print("----进程池中的进程-----pid=%d,ppid=%d --"%(os.getpid(),os.getppid())) 
  for i in range(3): 
    print("-----%d----"%i) 
    time.sleep(1) 
  return "over" #子进程执行完后返回给操作系统，返回给父进程 
 
def test2(args): 
  print("-----callback func----pid=%d"%os.getpid())#主进程调用test2 
  print("------callback func---args=%s"%args) 
 
def main(): 
  pool=Pool(3) 
  pool.apply_async(func=test,callback=test2)#回调 
  time.sleep(5)#收到func进程结束后的信号后，执行回调函数test2 
 
  print("----主进程-pid = %d"%os.getpid()) 
 
if __name__=="__main__": 
  #main() 
  pool=Pool(3) 
  pool.apply_async(test,callback=test2)#回调 
  time.sleep(5)#收到func进程结束后的信号后，执行回调函数test2 
 
  print("----主进程-pid = %d"%os.getpid()) 
 
'''''显示结果不太正确，应该先运行test呀，再运行test2 
-----callback func----pid=7044 
------callback func---args=over 
----主进程-pid = 7044 
----进程池中的进程-----pid=3772,ppid=7044 -- 
-----0---- 
-----1---- 
-----2---- 
'''