网站qq客服制作/关键字搜索引擎
10.3.9 同步线程
除了使用Event,还可以通过使用一个Condition对象来同步线程。由于Condition使用了一个Lock，所以它可以绑定到一个共享资源，允许多个线程等待资源更新。在下一个例子中，consumer()线程要等待设置了Condition才能继续。producer()线程负责设置条件，以及通知其他线程继续。
import logging
import threading
import timedef consumer(cond):"""wait for the condition and use the resource"""logging.debug('Starting consumer thread')with cond:cond.wait()logging.debug('Resource is available ti consumer')def producer(cond):"""set up the resource to be used bu the consumer"""logging.debug('Strating produce thread')with cond:logging.debug('Makeing resource available')cond.notifyAll()logging.basicConfig(level=logging.DEBUG,format='%(asctime)s (%(threadName)-2s) %(message)s',)condition = threading.Condition()
c1 = threading.Thread(name='c1',target=consumer,args=(condition,))
c2 = threading.Thread(name='c2',target=consumer,args=(condition,))
p = threading.Thread(name='p',target=producer,args=(condition,))
c1.start()
time.sleep(0.2)
c2.start()
time.sleep(0.2)
p.start()
这些线程使用with来获得与Condition关联的锁。也可以显式地使用acquire()和release()方法。
运行结果：
屏障(barrier)是另一种新出同步机制。Barrier会建立一个控制点，所有参与线程会在这里阻塞，直到所有这些参与“方”都到达这一点。采用这种方法，线程可以单独启动然后暂停，直到所有线程都准备好才可以继续。
import threading
import timedef worker(barrier):print(threading.current_thread().name,'waiting for barrier with {} others'.format(barrier.n_waiting))worker_id = barrier.wait()print(threading.current_thread().name,'after barrier',worker_id)NUM_THREADS = 3barrier = threading.Barrier(NUM_THREADS)threads = [threading.Thread(name='worker-%s' % i,target=worker,args=(barrier,),)for i in range(NUM_THREADS)]
for t in threads:print(t.name,'starting')t.start()time.sleep(0.1)for t in threads:t.join()
在这个例子中，Barrier被配置为会阻塞线程，直到3个线程都在等待。满足这个条件时，所有线程被同时释放从而越过这个控制点。wait()的返回值指示了释放的参与线程数，可以用来限制一些线程做清理资源等动作。
运行结果：
Barrier的abort()方法会使所有等待线程接收一个BrokenBarrierError。如果线程在wait()上被阻塞而停止处理，这就允许线程完成清理工作。
import threading
import timedef worker(barrier):print(threading.current_thread().name,'waiting for barrier with {} others'.format(barrier.n_waiting))try:worker_id = barrier.wait()except threading.BrokenBarrierError:print(threading.current_thread().name,'aborting')else:print(threading.current_thread().name,'after barrier',worker_id)
NUM_THREADS = 3barrier = threading.Barrier(NUM_THREADS + 1)threads = [threading.Thread(name='worker-%s' % i,target=worker,args=(barrier,),)for i in range(NUM_THREADS)]for t in threads:print(t.name,'starting')t.start()time.sleep(0.1)barrier.abort()for t in threads:t.join()
这个例子将Barrier配置为多加一个线程，即需要比实际启动的线程再多一个参与线程，所以所有线程中的处理都会阻塞。在被阻塞的各个线程中，abort()调用会产生一个异常。
运行结果：
