Python3标准库：threading进程中管理并发操作(6)

当前位置:

首页 > Python基础教程 >

Python3标准库：threading进程中管理并发操作(6)

'(%(threadName)-10s) %(message)s',

)

lock = threading.Lock()

w = threading.Thread(target=worker_with, args=(lock,))

nw = threading.Thread(target=worker_no_with, args=(lock,))

w.start()

nw.start()

函数worker_with()和worker_no_with()用等价的方式管理锁。

1.9 同步线程

除了使用Event，还可以通过使用一个Condition对象来同步线程。由于Condition使用了一个Lock，所以它可以绑定到一个共享资源，允许多个线程等待资源更新。在下一个例子中，consumer()线程要等待设置了Condition才能继续。producer()线程负责设置条件，以及通知其他线程继续。

import logging
import threading
import time
def consumer(cond):
"""wait for the condition and use the resource"""
logging.debug('Starting consumer thread')
with cond:
cond.wait()
logging.debug('Resource is available to consumer')
def producer(cond):
"""set up the resource to be used by the consumer"""
logging.debug('Starting producer thread')
with cond:
logging.debug('Making 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 time
def 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 = 3
barrier = 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 time
def 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')

栏目列表