Python线程同步机制详解:Lock、RLock、Condition与Semaphore
本文深入解析Python中threading模块提供的线程同步原语,涵盖Lock、RLock、Condition和Semaphore。通过代码示例说明join与setDaemon的区别,演示如何避免竞态条件,并解析Condition的唤醒机制及Semaphore控制并发数的原理与源码实现,帮助开发者掌握多线程编程的核心技巧。
本文主要介绍Python语言中用于实现线程同步的相关对象,核心内容涉及thread和threading这两个基础模块的使用与区别。
在threading模块中,提供了多种线程同步原语,具体包括Lock(互斥锁)、RLock(可重入锁)、Condition(条件变量)、Event(事件)以及Semaphore(信号量)等关键对象。
线程执行基础
join方法与setDaemon属性的应用
默认情况下,子线程在主线程运行结束后,仍会继续执行直至完成。若通过setDaemon(True)将子线程设置为守护线程,则当主线程结束时,该子线程也会立即终止,不再执行剩余代码。
若在主线程中调用join()方法,主线程将会阻塞并等待所指定的子线程执行完毕后,再继续执行后续代码。
import threading
import time
def get_thread_a():
print("get thread A started")
time.sleep(3)
print("get thread A end")
def get_thread_b():
print("get thread B started")
time.sleep(5)
print("get thread B end")
if __name__ == "__main__":
thread_a = threading.Thread(target=get_thread_a)
thread_b = threading.Thread(target=get_thread_b)
start_time = time.time()
thread_b.setDaemon(True)
thread_a.start()
thread_b.start()
thread_a.join()
end_time = time.time()
print("execution time: {}".format(end_time - start_time))
在示例中,thread_a设置了join(),因此主线程会等待thread_a执行;而thread_b被设置为守护线程,当主线程执行完毕后,thread_b将不再继续执行。具体执行结果如下所示:
get thread A started get thread B started get thread A end execution time: 3.003199815750122
线程同步机制
当多个线程共享全局变量并对该变量执行不同的操作时,变量最终的值可能是不确定的(即每次执行结果可能不同)。例如,对count变量进行加减操作时,其最终值具有不确定性。为了确保count的值确定,必须对线程执行的代码段加锁,以保护临界区。
Python中实现线程加锁主要依赖Lock和RLock这两个模块。
Lock(互斥锁):
from threading import Lock lock = Lock() lock.acquire() lock.release()
Lock对象包含acquire()和release()两个方法,这两个方法必须成对出现。在调用acquire()获取锁后,必须在后续调用release()释放锁,否则可能导致死锁或资源无法释放。
RLock(可重入锁):
考虑到Lock可能引发死锁问题,RLock(Reentrant Lock,可重入锁)对Lock进行了改进。RLock允许同一个线程在内部连续多次调用acquire(),但必须执行相同次数的release()来释放锁,从而支持递归调用场景。
from threading import RLock lock = RLock() lock.acquire() lock.acquire() lock.release() lock.release()
Condition(条件变量):线程在执行过程中,只有当满足特定条件后,才能访问相关数据或继续执行。
import threading
def get_thread_a(condition):
with condition:
condition.wait()
print("A : Hello B,that's ok")
condition.notify()
condition.wait()
print("A : I'm fine,and you?")
condition.notify()
condition.wait()
print("A : Nice to meet you")
condition.notify()
condition.wait()
print("A : That's all for today")
condition.notify()
def get_thread_b(condition):
with condition:
print("B : Hi A, Let's start the conversation")
condition.notify()
condition.wait()
print("B : How are you")
condition.notify()
condition.wait()
print("B : I'm fine too")
condition.notify()
condition.wait()
print("B : Nice to meet you,too")
condition.notify()
condition.wait()
print("B : Oh,goodbye")
if __name__ == "__main__":
condition = threading.Condition()
thread_a = threading.Thread(target=get_thread_a, args=(condition,))
thread_b = threading.Thread(target=get_thread_b, args=(condition,))
thread_a.start()
thread_b.start()
Condition内部持有一把锁,默认为RLock。在调用wait()和notify()之前,必须先调用acquire()获取这把锁才能继续执行。当wait()和notify()执行完毕后,需调用release()释放锁。在使用with condition语句时,进入上下文会自动执行acquire(),退出上下文时自动执行release()。因此,Condition包含两层锁机制:最底层的锁在调用wait()时会释放,同时线程会被加入等待队列,直到被notify()唤醒并重新获取锁。
wait():允许线程等待某个条件变量的通知,直到其他线程调用notify()或notify_all()。
notify():用于唤醒等待队列中正在等待该条件变量的线程。
Condition示例的执行结果如下:
B : Hi A, Let's start the conversation A : Hello B,that's ok B : How are you A : I'm fine,and you? B : I'm fine too A : Nice to meet you B : Nice to meet you,too A : That's all for today B : Oh,goodbye
Semaphore(信号量)
Semaphore主要用于控制线程的并发数量。例如,在爬虫场景中,若请求过于频繁可能导致IP被禁止,通过控制同时爬取网页的线程数量,可以在一定程度上防止IP被封禁。在文件读写场景中,可限制写线程每次只有一个,而允许多个读线程并发执行。
import time
import threading
def get_thread_a(semaphore,i):
time.sleep(1)
print("get thread : {}".format(i))
semaphore.release()
def get_thread_b(semaphore):
for i in range(10):
semaphore.acquire()
thread_a = threading.Thread(target=get_thread_a, args=(semaphore,i))
thread_a.start()
if __name__ == "__main__":
semaphore = threading.Semaphore(2)
thread_b = threading.Thread(target=get_thread_b, args=(semaphore,))
thread_b.start()
上述示例展示了每隔1秒并发执行两个线程的情况。当调用一次semaphore.acquire()时,Semaphore的内部计数值减1,直至计数值为0时被阻塞(锁上);调用release()后,计数值加1。从本质上看,Semaphore是基于Condition实现的:当Semaphore值为0时,semaphore.acquire()会调用Condition.wait()进行等待;否则将值减1。而semaphore.release()会将Semaphore的值加1,并调用Condition.notify()唤醒等待线程。
Semaphore相关源码片段
def acquire(self, blocking=True, timeout=None):
if not blocking and timeout is not None:
raise ValueError("can't specify timeout for non-blocking acquire")
rc = False
endtime = None
with self._cond:
while self._value == 0:
if not blocking:
break
if timeout is not None:
if endtime is None:
endtime = _time() + timeout
else:
timeout = endtime - _time()
if timeout <= 0:
break
self._cond.wait(timeout)
else:
self._value -= 1
rc = True
return rc
def release(self):
with self._cond:
self._value += 1
self._cond.notify()
以上就是本文关于Python线程同步机制的全部内容,希望这些内容对大家的学习或工作具有参考价值。如有任何疑问,欢迎留言交流,感谢大家对本站的支持。
本站声明
本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。













