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

当前位置:

首页 > Python基础教程 >

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

			else:

			print(threading.current_thread().name, 'after barrier',

			worker_id)

			NUM_THREADS = 3

			barrier = 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()调用会产生一个异常。

1.10 限制资源的并发访问

有时可能需要允许多个工作线程同时访问一个资源，但要限制总数。例如，连接池支持同时连接，但数目可能是固定的，或者一个网络应用可能支持固定数目的并发下载。这些连接就可以使用Semaphore来管理。

import logging
import threading
import time
class ActivePool:
def __init__(self):
super(ActivePool, self).__init__()
self.active = []
self.lock = threading.Lock()
def makeActive(self, name):
with self.lock:
self.active.append(name)
logging.debug('Running: %s', self.active)
def makeInactive(self, name):
with self.lock:
self.active.remove(name)
logging.debug('Running: %s', self.active)
def worker(s, pool):
logging.debug('Waiting to join the pool')
with s:
name = threading.current_thread().getName()
pool.makeActive(name)
time.sleep(0.1)
pool.makeInactive(name)
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s (%(threadName)-2s) %(message)s',
)
pool = ActivePool()
s = threading.Semaphore(2)
for i in range(4):
t = threading.Thread(
target=worker,
name=str(i),
args=(s, pool),
)
t.start()

在这个例子中，ActivePool类只作为一种便利方法，用来跟踪某个给定时刻哪些线程能够运行。真正的资源池会为新的活动线程分配一个连接或另外某个值，并且当这个线程工作完成时再回收这个值。在这里，资源池只是用来保存活动线程的名，以显示至少有两个线程在并发运行。

1.11 线程特定的数据

有些资源需要锁定以便多个线程使用，另外一些资源则需要保护，以使它们对并非是这些资源的“所有者”的线程隐藏。local()函数会创建一个对象，它能够隐藏值，使其在不同线程中无法被看到。

import random
import threading
import logging
def show_value(data):
try:
val = data.value
except AttributeError:
logging.debug('No value yet')
else:
logging.debug('value=%s', val)
def worker(data):
show_value(data)
data.value = random.randint(1, 100)
show_value(data)
logging.basicConfig(
level=logging.DEBUG,
format='(%(threadName)-10s) %(message)s',
)
local_data = threading.local()
show_value(local_data)
local_data.value

栏目列表