多线程实践

到目前为止，我们已经见到的这些简单的示例片段都无法代表你要在实践中写出的代码。
除了演示多线程和创建线程的不同方式外，之前的代码实际上什么有用的事情都没有做。我
们启动这些线程以及等待它们结束的方式都是一样的，它们也全都睡眠。

4.3.1 节曾提到，由于Python 虚拟机是单线程（GIL）的原因，只有线程在执行I/O 密集
型的应用时才能更好地发挥Python 的并发性（对比计算密集型应用，它只需要做轮询），因
此让我们看一个I/O 密集型的例子，然后作为进一步的练习，尝试将其移植到Python 3 中，
以让你对向Python 3 移植的处理有一定认识。

图书排名示例

示例4-9 的bookrank.py 脚本非常直接。它将前往我最喜欢的在线零售商之一Amazon，
然后请求你希望查询的图书的当前排名。在这个示例代码中，你可以看到函数getRanking()
使用正则表达式来拉取和返回当前的排名，而函数_showRanking()用于向用户显示结果。

请记住，根据Amazon 的使用条件，“Amazon 对您在本网站的访问和个人使用授予有限
许可，未经Amazon 明确的书面同意，不允许对全部或部分内容进行下载（页面缓存除外）
或修改。”在该程序中，我们所做的只是查询指定书籍的当前排名，没有任何其他操作，甚至
都不会对页面进行缓存。

示例4-9 是我们对于bookrank.py 的第一次（不过与最终版本也很接近了）尝试，这是一
个没有使用线程的版本。示例4-9 图书排名“Screenscraper”（bookrank.py）

from urllib.request import urlopen  # 修正1：导入正确的urlopen
from re import compile
from threading import Thread
from time import ctime
from atexit import register

REGEX = compile(r'#([\d,]+) in Books')  # 修正2：添加正则表达式r前缀
AMZN = 'https://www.amazon.com/dp/'
ISBNs = {
    '0132269937': 'Core Python Programming',
    '0132356139': 'Python Web Development with Django',
    '0137143419': 'Python Fundamentals',
}


def getRanking(isbn):  # 修正3：修复函数名拼写错误
    with urlopen(f'{AMZN}{isbn}') as page:  # 使用上下文管理器自动关闭资源
        data = page.read().decode('utf-8')
    match = REGEX.search(data)
    return match.group(1) if match else 'Ranking not found'


def _showRanking(isbn):
    title = ISBNs[isbn]
    ranking = getRanking(isbn)
    print(f'- {title} ranked {ranking}')  # 修正4：使用f-string格式化输出


def _main():
    print(f'At {ctime()} on Amazon...')
    threads = []
    for isbn in ISBNs:
        # 为每个ISBN创建独立线程
        t = Thread(target=_showRanking, args=(isbn,))
        threads.append(t)
        t.start()

        # 等待所有线程完成
    for t in threads:
        t.join()


@register
def _atexit():
    print('All DONE at:', ctime())


if __name__ == '__main__':
    _main()

你可能会感到疑惑，为什么我们会把数据的获取（getRanking()）和显示（_showRanking()
和_main()）过程分开呢？这样做是为了防止你产生除了通过终端向用户显示结果以外的想
法。在实践中，你可能会有将数据通过Web 模板返回、存储在数据库中或者发送结果文本到
手机上等需求。如果把所有代码都放在一个函数里，会难以复用和/或重新调整。
此外，如果Amazon 修改了商品页的布局，你可能需要修改正则表达式“screenscraper”
以继续从商品页提取数据。还需要说明的是，在这个简单的例子中使用正则表达式（或者只
是简学的旧式字符串处理）是没有问题的，不过你可能需要一个更强大的标记解析器，比如
标准库中的HTMLParser，第三方工具BeautifulSoup、html5lib 或者lxml（第9 章会演示其中
部分工具）。

引入线程

不需要你告诉我这仍然是一个愚蠢的单线程程序，我们接下来就要使用多线程来修改这
个应用。由于这是一个I/O 密集型应用，因此这个程序使用多线程是一个好的选择。简单起
见，我们不会在这里使用任何类和面向对象编程，而是使用threading 模块。我们将直接使用
Thread 类，所以你可以认为这更像是mtsleepC.py 的衍生品，而不是它之后的例子。我们将
只是派生线程，然后立即启动这些线程。

将应用中的_showRanking(isbn)进行如下修改。

Th read(target=_showRanking, args=(isbn,)).start().

就是这样！现在，你得到了bookrank.py 的最终版本，可以看出由于增加了并发，这个
应用（一般）会运行得更快。不过，程序能够运行多快还取决于最慢的那个响应。

$ python bookrank.py
At Thu Mar 31 10:11:32 2011 on Amazon...
- 'Python Fundamentals' ranked 869,010
- 'Core Python Programming' ranked 36,481
- 'Python Web Development with Django' ranked 219,228
al l DONE at: Thu Mar 31 10:11:35 2011

正如你在输出中所看到的，相比于单线程版本的6 秒，多线程版本只需要运行3 秒。而
另外一个需要注意的是，多线程版本按照完成的顺序输出，而单线程版本按照变量的顺序。
在单线程版本中，顺序是由字典的键决定的，而现在查询是并发产生的，输出的先后则会由
每个线程完成任务的顺序来决定。

在之前mtsleepX.py 的例子中，对所有线程使用了Thread.join()用于阻塞执行，直到每个
线程都已退出。这可以有效阻止主线程在所有子线程都完成之前继续执行，所以输出语句“all
DONE at”可以在正确的时间调用。

在这些例子中，对所有线程调用join()并不是必需的，因为它们不是守护线程。无论
如何主线程都不会在派生线程完成之前退出脚本。由于这个原因，我们将在mtsleepF.py
中删除所有的join()操作。不过，要意识到如果我们在同一个地方显示“all done”这是不
正确的。

主线程会在其他线程完成之前显示“all done”，所以我们不能再把print 调用放在
_main()里了。有两个地方可以放置print 语句：第37 行_main()返回之后（脚本最后一行），
或者使用atexit.register()来注册一个退出函数。因为之前没有讨论过后面这种方法，而且
它可能对你以后更有帮助，所以我们认为这是一个介绍它的好位置。此外，这还是一个
在Python 2 和3 中保持一致的接口，接下来我们就要挑战如何将这个程序移植到Python 3
中了。

移植到Python 3
下面我们希望这个脚本能够在Python 3 中运行。对于项目和应用而言，都需要继续进行
迁移，这是你必须要熟悉的事情。幸运的是，有一些工具可以帮助我们，其中之一是2to3 这
个工具。它的一般用法如下。

$ 2to3 foo.py # only output diff
$ 2to3 -w foo.py # overwrites w/3.x code

在第一条命令中，2to3 工具只是显示原始脚本的2.x 版本与其生成的等价的3.x 版本的
区别。而-w 标志则让2to3 工具使用新生成的3.x 版本的代码重写原始脚本，并将2.x 版本重
命名为foo.py.bak。

让我们对bookrank.py 运行2to3 工具，在已有的文件上进行改写。除了给出区别外，它
还会像之前描述的那样保存新版本的脚本。

$ 2to3 -w bookrank.py
RefactoringTool: Skipping implicit fixer: buffer
RefactoringTool: Skipping implicit fixer: idioms
RefactoringTool: Skipping implicit fixer: set_literal
RefactoringTool: Skipping implicit fixer: ws_comma
--- bookrank.py (original)
+++ bookrank.py (refactored)
@@ -4,7 +4,7 @@
from re import compile
from threading import Thread
from time import ctime
-from urllib2 import urlopen as uopen
+from urllib.request import urlopen as uopen
REGEX = compile('#([\d,]+) in Books ')
AMZN = 'http://amazon.com/dp/'
@@ -21,17 +21,17 @@
return REGEX.findall(data)[0]
def _showRanking(isbn):
- print '- %r ranked %s' % (
- ISBNs[isbn], getRanking(isbn))
+ print('- %r ranked %s' % (
+ ISBNs[isbn], getRanking(isbn)))
def _main():
- print 'At', ctime(), 'on Amazon...'
+ print('At', ctime(), 'on Amazon...')
for isbn in ISBNs:
Thread(target=_showRanking,
args=(isbn,)).start()#_showRanking(isbn)
@register
def _atexit():
- print 'all DONE at:', ctime()
+ print('all DONE at:', ctime())
if __name__ == '__main__':
_main()
RefactoringTool: Files that were modified:
Re factoringTool: bookrank.py

接下来的步骤对于读者而言是可选的，我们使用POSIX 命令行将文件重命名为
bookrank.py 和bookrank3.py（Windows 用户应当使用ren 命令）。

$ mv bookrank.py bookrank3.py
$ mv bookrank.py.bak bookrank.py

如果你尝试运行新生成的代码，就会发现假定它是一个完美翻译，不需要你再做任何操
作的想法只是你的一厢情愿。糟糕的事情发生了，你会在每个线程执行时得到如下异常信息
（下面的输出只针对一个线程，因为每个线程的输出都一样）。

$ python3 bookrank3.py
Exception in thread Thread-1:
Traceback (most recent call last):
File "/Library/Frameworks/Python.framework/Versions/
3.2/lib/python3.2/threading.py", line 736, in
_bootstrap_inner
self.run()
File "/Library/Frameworks/Python.framework/Versions/
3.2/lib/python3.2/threading.py", line 689, in run
self._target(*self._args, **self._kwargs)
File "bookrank3.py", line 25, in _showRanking
ISBNs[isbn], getRanking(isbn)))
File "bookrank3.py", line 21, in getRanking
return REGEX.findall(data)[0]
TypeError: can't use a string pattern on a bytes-like object
:

问题看起来是：正则表达式是一个Unicode 字符串，而urlopen()返回来的类似文件对象
的结果经过read()方法得到的是一个ASCII/bytes 字符串。这里的修复方案是将其编译为一个
bytes 对象，而不是文本字符串。因此，修改第9 行，让re.compile()编译一个bytes 字符串（通
过添加bytes 字符串）。为了做到这个，可以在左侧的引号前添加一个bytes 字符串的标记b，
如下所示。

RE GEX = compile(b'#([\d,]+) in Books ')

现在，让我们再试一次。

$ python3 bookrank3.py
At Sun Apr 3 00:45:46 2011 on Amazon...
- 'Core Python Programming' ranked b'108,796'
- 'Python Web Development with Django' ranked b'268,660'
- 'Python Fundamentals' ranked b'969,149'
al l DONE at: Sun Apr 3 00:45:49 2011

现在又是什么问题呢？虽然这个输出结果比之前要好一些（没有错误），但是它看起来还
是有些奇怪。当传给str()时，正则表达式抓取的排名值显示了b 和引号。你的第一直觉可能
是尝试丑陋的字符串切片。

>>> x = b'xxx'
>>> repr(x)
"b'xxx'"
>>> str(x)
"b'xxx'"
>>> str(x)[2:-1]
'x xx'

不过，更合适的方法是将其转换为一个真正的（Unicode）字符串，可能会用到UTF-8。

>>> str(x, 'utf-8')
'x xx'

为了实现这一点，在脚本里，对第53 行进行一个类似的修改，如下所示。

re turn str(REGEX.findall(data)[0], 'utf-8')

现在，Python 3 版本的脚本输出就和Python 2 的脚本一致了。

$ python3 bookrank3.py
At Sun Apr 3 00:47:31 2011 on Amazon...
- 'Python Fundamentals' ranked 969,149
- 'Python Web Development with Django' ranked 268,660
- 'Core Python Programming' ranked 108,796
al l DONE at: Sun Apr 3 00:47:34 2011

一般来说，你会发现从2.x 版本移植到3.x 版本会遵循类似下面的模式：你需要确保所有
的单元测试和集成测试都已经通过，使用2to3（或其他工具）进行所有的基础修改，然后进
行一些善后工作，让代码运行起来并通过相同的测试。我们将在下一个例子中再次尝试这个
练习，这个例子将演示线程同步的使用。

同步原语

在本章的主要部分，我们了解了线程的基本概念，以及如何在Python 应用中利用线程。然
而，我们遗漏了多线程编程中一个非常重要的方面：同步。一般在多线程代码中，总会有一些特
定的函数或代码块不希望（或不应该）被多个线程同时执行，通常包括修改数据库、更新文件或
其他会产生竞态条件的类似情况。回顾本章前面的部分，如果两个线程运行的顺序发生变化，就
有可能造成代码的执行轨迹或行为不相同，或者产生不一致的数据（可以在Wikipedia 页面上阅
读有关竞态条件的更多信息：http://en.wikipedia.org/wiki/Race_condition）。

这就是需要使用同步的情况。当任意数量的线程可以访问临界区的代码
（http://en.wikipedia.org/wiki/Critical_section）但在给定的时刻只有一个线程可以通过时，就是使
用同步的时候了。程序员选择适合的同步原语，或者线程控制机制来执行同步。进程同步有不
同的类型（参见http://en.wikipedia.org/wiki/Synchronization_(computer_ science)），Python 支持
多种同步类型，可以给你足够多的选择，以便选出最适合完成任务的那种类型。

本章前面对同步进行过一些介绍，所以这里就使用其中两种类型的同步原语演示几个示例程
序：锁/互斥，以及信号量。锁是所有机制中最简单、最低级的机制，而信号量用于多线程竞争
有限资源的情况。锁比较容易理解，因此先从锁开始，然后再讨论信号量。

锁示例

锁有两种状态：锁定和未锁定。而且它也只支持两个函数：获得锁和释放锁。它的行为
和你想象的完全一样。
当多线程争夺锁时，允许第一个获得锁的线程进入临界区，并执行代码。所有之后到达
的线程将被阻塞，直到第一个线程执行结束，退出临界区，并释放锁。此时，其他等待的线
程可以获得锁并进入临界区。不过请记住，那些被阻塞的线程是没有顺序的（即不是先到先
执行），胜出线程的选择是不确定的，而且还会根据Python 实现的不同而有所区别。

让我们来看看为什么锁是必需的。mtsleepF.py 应用派生了随机数量的线程，当每个线程
执行结束时它会进行输出。下面是其核心部分的源码（Python 2）。

from atexit import register
from random import randrange
from threading import Thread, currentThread, Lock
from time import sleep, ctime


# 线程安全的集合
class CleanOutputSet(set):
    def __init__(self):
        super(CleanOutputSet, self).__init__()
        self.lock = Lock()

    def add(self, item):
        with self.lock:
            super(CleanOutputSet, self).add(item)

    def remove(self, item):
        with self.lock:
            super(CleanOutputSet, self).remove(item)

    def __str__(self):
        return ', '.join(x for x in self)


remaining = CleanOutputSet()
lock = Lock()


def loop(nsec):
    myname = currentThread().name
    remaining.add(myname)
    with lock:
        print('[%s] Started %s (%d secs)' % (ctime(), myname, nsec))
    sleep(nsec)
    with lock:
        print('[%s] Completed %s (%d secs)' % (ctime(), myname, nsec))
        remaining.remove(myname)
        print(' (remaining: %s)' % (remaining or "NONE"))


def _main():
    threads = []
    for pause in (randrange(2, 5) for _ in range(randrange(3, 7))):
        t = Thread(target=loop, args=(pause,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()


@register
def _atexit():
    print('all DONE at:', ctime())


if __name__ == '__main__':
    _main()

运行结果

[Mon Jul 28 13:27:01 2025] Started Thread-2 (loop) (3 secs)
[Mon Jul 28 13:27:01 2025] Started Thread-3 (loop) (4 secs)
[Mon Jul 28 13:27:01 2025] Started Thread-1 (loop) (3 secs)
[Mon Jul 28 13:27:04 2025] Completed Thread-1 (loop) (3 secs)
 (remaining: Thread-2 (loop), Thread-3 (loop))
[Mon Jul 28 13:27:04 2025] Completed Thread-2 (loop) (3 secs)
 (remaining: Thread-3 (loop))
[Mon Jul 28 13:27:05 2025] Completed Thread-3 (loop) (4 secs)
 (remaining: NONE)
all DONE at: Mon Jul 28 13:27:05 2025

当我们完成这个使用锁的代码后，会有一个比较详细的逐行解释，不过mtsleepF.py
所做的基本上就是之前例子的扩展。和bookrank.py 一样，为了简化代码，没有使用面向
对象编程，删除了线程对象列表和线程的join()，重用了atexit.register()（和bookrank.py
相同的原因）。

另一个和之前的那些mtsleepX.py 例子不同的地方是，这里不再把循环/线程对硬编码
成睡眠4 秒和2 秒，而是将它们随机地混合在一起，创建3～6 个线程，每个线程睡眠
2～4 秒。

这里还有一个新功能，使用集合来记录剩下的还在运行的线程。我们对集合进行了子类
化而不是直接使用，这是因为我们想要演示另一个用例：变更集合的默认可印字符串。

当显示一个集合时，你会得到类似set([X, Y, Z,…])这样的输出。而应用的用户并不需要
（也不应该）知道关于集合的信息，或者我们使用了这些集合。我们只需要显示成类似X, Y,
Z, …这样即可。这也就是派生了set 类并实现了它的__str__()方法的原因。
如果幸运，进行了这些改变之后，输出将会按照适当的顺序给出。

$ python mtsleepF.py
[Sat Apr 2 11:37:26 2011] Started Thread-1
[Sat Apr 2 11:37:26 2011] Started Thread-2
[Sat Apr 2 11:37:26 2011] Started Thread-3
[Sat Apr 2 11:37:29 2011] Completed Thread-2 (3 secs)
(remaining: Thread-3, Thread-1)
[Sat Apr 2 11:37:30 2011] Completed Thread-1 (4 secs)

(remaining: Thread-3)
[Sat Apr 2 11:37:30 2011] Completed Thread-3 (4 secs)
(remaining: NONE)
al l DONE at: Sat Apr 2 11:37:30 2011

不过，如果不幸，你将会得到像下面几对执行示例这样奇怪的输出结果。

$ python mtsleepF.py
[Sat Apr 2 11:37:09 2011] Started Thread-1
[Sat Apr 2 11:37:09 2011] Started Thread-2
[Sat Apr 2 11:37:09 2011] Started Thread-3
[Sat Apr 2 11:37:12 2011] Completed Thread-1 (3 secs)
[Sat Apr 2 11:37:12 2011] Completed Thread-2 (3 secs)
(remaining: Thread-3)
(remaining: Thread-3)
[Sat Apr 2 11:37:12 2011] Completed Thread-3 (3 secs)
(remaining: NONE)
all DONE at: Sat Apr 2 11:37:12 2011
$ python mtsleepF.py
[Sat Apr 2 11:37:56 2011] Started Thread-1
[Sat Apr 2 11:37:56 2011] Started Thread-2
[Sat Apr 2 11:37:56 2011] Started Thread-3
[Sat Apr 2 11:37:56 2011] Started Thread-4
[Sat Apr 2 11:37:58 2011] Completed Thread-2 (2 secs)
[Sat Apr 2 11:37:58 2011] Completed Thread-4 (2 secs)
(remaining: Thread-3, Thread-1)
(remaining: Thread-3, Thread-1)
[Sat Apr 2 11:38:00 2011] Completed Thread-1 (4 secs)
(remaining: Thread-3)
[Sat Apr 2 11:38:00 2011] Completed Thread-3 (4 secs)
(remaining: NONE)
al l DONE at: Sat Apr 2 11:38:00 2011

那么出现什么问题了呢？一个问题是，输出可能部分混乱（因为多个线程可能并行执行
I/O）。同样地，之前的几个示例代码也都有交错输出的问题存在。而另一问题则出现在两个
线程修改同一个变量（剩余线程名集合）时。

I/O 和访问相同的数据结构都属于临界区，因此需要用锁来防止多个线程同时进入临界
区。为了加锁，需要添加一行代码来引入Lock（或RLock），然后创建一个锁对象，因此需
要添加/修改代码以便在合适的位置上包含这些行。

from threading import Thread, Lock, currentThread
lo ck = Lock()

现在应该使用刚创建的这个锁了。下面代码中突出显示的acquire()和release()调用就是
应当在loop()函数中添加的语句。

def loop(nsec):
myname = currentThread().name
lock.acquire()
remaining.add(myname)
print '[%s] Started %s' % (ctime(), myname)
lock.release()
sleep(nsec)
lock.acquire()
remaining.remove(myname)
print '[%s] Completed %s (%d secs)' % (
ctime(), myname, nsec)
print ' (remaining: %s)' % (remaining or 'NONE')
lock.release()

一旦做了这个改变，就不会再产生那种奇怪的输出了。

$ python mtsleepF.py
[Sun Apr 3 23:16:59 2011] Started Thread-1
[Sun Apr 3 23:16:59 2011] Started Thread-2
[Sun Apr 3 23:16:59 2011] Started Thread-3
[Sun Apr 3 23:16:59 2011] Started Thread-4
[Sun Apr 3 23:17:01 2011] Completed Thread-3 (2 secs)
(remaining: Thread-4, Thread-2, Thread-1)
[Sun Apr 3 23:17:01 2011] Completed Thread-4 (2 secs)
(remaining: Thread-2, Thread-1)
[Sun Apr 3 23:17:02 2011] Completed Thread-1 (3 secs)
(remaining: Thread-2)
[Sun Apr 3 23:17:03 2011] Completed Thread-2 (4 secs)
(remaining: NONE)
al l DONE at: Sun Apr 3 23:17:03 2011

修改后的最终版mtsleepF.py 如示例4-10 所示。
示例4-10 锁和更多的随机性（mtsleepF.py）

from atexit import register
from random import randrange
from threading import Thread, Lock, current_thread  # 修改：currentThread->current_thread 
from time import sleep, ctime
 
class CleanOutputSet(set):
    def __str__(self):
        return ', '.join(x for x in self)
 
lock = Lock()
loops = (randrange(2, 5) for x in range(randrange(3, 7)))  # 修改：xrange->range 
remaining = CleanOutputSet()
 
def loop(nsec):
    myname = current_thread().name  # 修改：currentThread()->current_thread()
    with lock:  # 使用with简化锁管理 
        remaining.add(myname) 
        print(f'[{ctime()}] Started {myname}')
    
    sleep(nsec)  # 模拟任务执行
    
    with lock:
        remaining.remove(myname) 
        # 修正拼写和括号问题 ↓ 
        print(f'[{ctime()}] Completed {myname}({nsec} secs)\n(remaining: {remaining or "NONE"})')
 
def _main():
    for pause in loops:
        Thread(target=loop, args=(pause,)).start()
 
@register
def _atexit():
    print(f'all DONE at: {ctime()}')
 
if __name__ == '__main__':
    _main()

运行结果

[Tue Jul 29 13:21:58 2025] Started Thread-1 (loop)
[Tue Jul 29 13:21:58 2025] Started Thread-2 (loop)
[Tue Jul 29 13:21:58 2025] Started Thread-3 (loop)
[Tue Jul 29 13:21:58 2025] Started Thread-4 (loop)
[Tue Jul 29 13:22:00 2025] Completed Thread-4 (loop)(2 secs)
(remaining: Thread-1 (loop), Thread-2 (loop), Thread-3 (loop))
[Tue Jul 29 13:22:01 2025] Completed Thread-2 (loop)(3 secs)
(remaining: Thread-1 (loop), Thread-3 (loop))
[Tue Jul 29 13:22:01 2025] Completed Thread-1 (loop)(3 secs)
(remaining: Thread-3 (loop))
[Tue Jul 29 13:22:02 2025] Completed Thread-3 (loop)(4 secs)
(remaining: NONE)
all DONE at: Tue Jul 29 13:22:02 2025

作为维护你自己的当前运行线程集合的一种替代方案，可以考虑使用threading.
enumerate()，该方法会返回仍在运行的线程列表（包括守护线程，但不包括没有启动的线
程）。在本例中并没有使用这个方案，因为它会显示两个额外的线程，所以我们需要删除这
两个线程以保持输出的简洁。这两个线程是当前线程（因为它还没结束），以及主线程（没
有必要去显示）。

此外，如果你使用的是Python 2.6 或更新的版本（包括3.x 版本），别忘了还可以使用
str.format()方法来代替字符串格式化操作符。换句话说，print 语句

print '[%s] Started %s' % (ctime(), myname)

可以在2.6+版本中被替换成

print '[{0}] Started {1}'.format(ctime(), myname)

或者在3.x 版本中调用print()函数：
pr int(‘[{0}] Started {1}’.format(ctime(), myname))
如果只需要对当前运行的线程进行计数，那么可以使用threading.activeCount()（2.6 版本
开始重命名为active_count()）来代替。

使用上下文管理
如果你使用Python 2.5 或更新版本，还有一种方案可以不再调用锁的acquire()和release()
方法，从而更进一步简化代码。这就是使用with 语句，此时每个对象的上下文管理器负责在
进入该套件之前调用acquire()并在完成执行之后调用release()。

threading 模块的对象Lock、RLock、Condition、Semaphore 和BoundedSemaphore 都包含
上下文管理器，也就是说,它们都可以使用with 语句。当使用with 时，可以进一步简化loop()
循环，如下面的代码所示。

import threading 
import time 
from time import sleep, ctime 
# 全局共享资源 
lock = threading.Lock()  # 线程锁 
remaining = set()        # 活跃线程集合 
def loop(nsec):
    """线程执行函数：模拟耗时任务"""
    myname = threading.current_thread().name  
    
    # 第一阶段：线程启动 
    with lock:
        remaining.add(myname) 
        print(f'[{ctime()}] Started {myname}')
    
    # 模拟耗时操作 
    sleep(nsec)
    
    # 第二阶段：线程完成 
    with lock:
        remaining.remove(myname) 
        print(f'[{ctime()}] Completed {myname} ({nsec} secs)')
        status = 'NONE' if not remaining else ', '.join(remaining)
        print(f'(remaining: {status})')
# 测试执行 
if __name__ == '__main__':
    tasks = [2, 4, 6]  # 各线程休眠时长 
    
    print(f'[{ctime()}] MAIN THREAD START')
    threads = []
    for i, duration in enumerate(tasks):
        t = threading.Thread(target=loop, args=(duration,), name=f'Thread-{i+1}')
        threads.append(t) 
        t.start() 
    
    for t in threads:
        t.join() 
    
    print(f'[{ctime()}] ALL THREADS COMPLETED')

运行结果

[Wed Jul 30 13:14:23 2025] MAIN THREAD START
[Wed Jul 30 13:14:23 2025] Started Thread-1
[Wed Jul 30 13:14:23 2025] Started Thread-2
[Wed Jul 30 13:14:23 2025] Started Thread-3
[Wed Jul 30 13:14:25 2025] Completed Thread-1 (2 secs)
(remaining: Thread-2, Thread-3)
[Wed Jul 30 13:14:27 2025] Completed Thread-2 (4 secs)
(remaining: Thread-3)
[Wed Jul 30 13:14:29 2025] Completed Thread-3 (6 secs)
(remaining: NONE)
[Wed Jul 30 13:14:29 2025] ALL THREADS COMPLETED

信号量示例

如前所述，锁非常易于理解和实现，也很容易决定何时需要它们。然而，如果情况更加
复杂，你可能需要一个更强大的同步原语来代替锁。对于拥有有限资源的应用来说，使用信
号量可能是个不错的决定。

信号量是最古老的同步原语之一。它是一个计数器，当资源消耗时递减，当资源释放
时递增。你可以认为信号量代表它们的资源可用或不可用。消耗资源使计数器递减的操作
习惯上称为P() （来源于荷兰单词probeer/proberen），也称为wait、try、acquire、pend 或procure。
相对地，当一个线程对一个资源完成操作时，该资源需要返回资源池中。这个操作一般称
为V()（来源于荷兰单词verhogen/verhoog），也称为signal、increment、release、post、vacate。
Python 简化了所有的命名，使用和锁的函数/方法一样的名字：acquire 和release。信号量比
锁更加灵活，因为可以有多个线程，每个线程拥有有限资源的一个实例。

在下面的例子中，我们将模拟一个简化的糖果机。这个特制的机器只有5 个可用的槽来
保持库存（糖果）。如果所有的槽都满了，糖果就不能再加到这个机器中了；相似地，如果每
个槽都空了，想要购买的消费者就无法买到糖果了。我们可以使用信号量来跟踪这些有限的
资源（糖果槽）。
示例4-11 为其源代码（candy.py）。

import threading
import time
import random
from atexit import register
from threading import BoundedSemaphore, Lock, Thread
from time import sleep, ctime

# 全局配置
MAX_CANDY = 5
lock = Lock()  # 共享资源锁
candytray = BoundedSemaphore(MAX_CANDY)  # 有限缓冲区信号量


def refill():
    """生产者：补充糖果"""
    with lock:  # 自动获取/释放锁
        print(f'[{ctime()}] Refilling candy...', end=' ')
        try:
            # 尝试释放信号量（增加糖果）
            candytray.release()
            print('SUCCESS (1 candy added)')
        except ValueError:  # 超过最大容量
            print('FAILED (tray full)')


def buy():
    """消费者：购买糖果"""
    with lock:
        print(f'[{ctime()}] Buying candy...', end=' ')
        # 非阻塞方式获取信号量
        if candytray.acquire(blocking=False):
            print('SUCCESS (1 candy taken)')
        else:  # 无可用资源
            print('FAILED (tray empty)')


def producer(loops):
    """生产者线程函数"""
    print(f'Producer STARTED, will refill {loops} times')
    for _ in range(loops):
        refill()
        sleep(random.uniform(0.5, 1.5))  # 随机延迟


def consumer(loops):
    """消费者线程函数"""
    print(f'Consumer STARTED, will buy {loops} times')
    for _ in range(loops):
        buy()
        sleep(random.uniform(0.8, 2.0))  # 随机延迟


@register
def _atexit():
    """程序退出处理"""
    print(f'[{ctime()}] Program terminated')


def main():
    print(f'\n{" CANDY MACHINE SIMULATOR ":=^60}')
    print(f'[{ctime()}] Starting (Capacity: {MAX_CANDY} candies)')

    # 随机生成操作次数 (2-5次)
    nloops = random.randint(2, 5)

    # 创建线程
    threads = [
        Thread(target=consumer, args=(random.randint(nloops, nloops + MAX_CANDY + 2),), name="CONSUMER"),
        Thread(target=producer, args=(nloops,), name="PRODUCER")
    ]

    # 启动线程
    for t in threads:
        t.daemon = True
        t.start()

        # 等待线程结束
    for t in threads:
        t.join(timeout=15)


if __name__ == '__main__':
    main()

运行结果

================= CANDY MACHINE SIMULATOR ==================
[Wed Jul 30 13:28:13 2025] Starting (Capacity: 5 candies)
Consumer STARTED, will buy 8 times
[Wed Jul 30 13:28:13 2025] Buying candy... SUCCESS (1 candy taken)
Producer STARTED, will refill 5 times
[Wed Jul 30 13:28:13 2025] Refilling candy... SUCCESS (1 candy added)
[Wed Jul 30 13:28:14 2025] Refilling candy... FAILED (tray full)
[Wed Jul 30 13:28:15 2025] Buying candy... SUCCESS (1 candy taken)
[Wed Jul 30 13:28:15 2025] Refilling candy... SUCCESS (1 candy added)
[Wed Jul 30 13:28:16 2025] Buying candy... SUCCESS (1 candy taken)
[Wed Jul 30 13:28:17 2025] Refilling candy... SUCCESS (1 candy added)
[Wed Jul 30 13:28:17 2025] Buying candy... SUCCESS (1 candy taken)
[Wed Jul 30 13:28:18 2025] Refilling candy... SUCCESS (1 candy added)
[Wed Jul 30 13:28:18 2025] Buying candy... SUCCESS (1 candy taken)
[Wed Jul 30 13:28:19 2025] Buying candy... SUCCESS (1 candy taken)
[Wed Jul 30 13:28:20 2025] Buying candy... SUCCESS (1 candy taken)
[Wed Jul 30 13:28:22 2025] Buying candy... SUCCESS (1 candy taken)
[Wed Jul 30 13:28:23 2025] Program terminated