Python多线程状态控制：Event与信号量实战解析

七月

发布时间：2026-10-15| 5人浏览过

原创

本文深入探讨Python中多线程的状态控制与同步机制。通过实例解析如何利用自定义标志位停止线程，并重点讲解threading.Event在阶段同步中的应用及其局限性。同时，对比介绍Semaphore信号量如何精确控制并发线程数量，避免Event导致的资源竞争问题，为高并发场景提供实用的同步方案。

一、停止线程

借助Threading库，我们可以便捷地创建线程以执行特定任务，从而提升程序运行效率。然而，线程一旦创建并提交给操作系统调度，主程序便难以直接干预其状态：既无法强制终止线程，也无法发送控制信号或调整调度策略，更缺乏其他高级操作接口。若需实现此类功能，必须自行开发控制逻辑。

具体应如何开发？

在创建线程时，我们指定了target参数指向目标函数。该函数不仅限于全局函数，也可以是类实例的方法。若采用类方法，便可在函数内部访问对象的其他属性，借此实现手动控制线程停止的功能。

这一概念看似复杂，但通过代码示例即可清晰理解：


import time
from threading import Thread

class TaskWithSwitch:
    def __init__(self):
        self._running = True
        
    def terminate(self):
        self._running = False

    def run(self, n):
        while self._running and n > 0:
            print('Running {}'.format(n))
            n -= 1
            time.sleep(1)

c = TaskWithSwitch()
t = Thread(target=c.run, args=(10, ))
t.start()
c.terminate()
t.join()

运行上述代码，屏幕仅输出10。这是因为我们将_running字段设为False后，下一次循环检查条件时不再满足，线程便自动退出。

若利用多线程处理IO操作，由于IO可能阻塞，线程可能长期无法返回，导致在循环内部“卡死”。此时，仅靠_running标志位已不足够，需在线程内部引入计时器机制，以防止循环因IO阻塞而永久挂起。


class IOTask:
    def __init__(self):
        self._running = True
        
    def terminate(self):
        self._running = False

    def run(self, sock):
        # 在socket中设置计时器
        sock.settimeout(10)
        while self._running:
            try:
                # 由于设置了计时器，所以这里不会永久等待
                data = sock.recv(1024)
                break
            except socket.timeout:
                continue
        return

二、线程信号的传递

我们之所以需要费力控制线程运行，根本原因在于线程状态对外不可见且无法直接操作，因为线程由操作系统管理。主线程与子线程相互独立，无从属关系，因此要控制线程状态，必须借助其他同步手段。

考虑一种场景：某任务需在另一线程完成后才能启动。要实现此逻辑，必须感知线程状态，并通过信号传递通知。Python的threading库提供了Event工具，它如同一个开关：当一个线程执行完毕时触发该开关，进而控制另一段逻辑的执行。

以下是示例代码：


import time
from threading import Thread, Event

def run_in_thread():
    time.sleep(1)
    print('Thread is running')

t = Thread(target=run_in_thread)
t.start()

print('Main thread print')

线程内部仅输出提示信息，无其他逻辑。由于run_in_thread函数中休眠了1秒，因此会先输出“Main thread print”，随后输出“Thread is running”。假设该线程执行关键任务，我们希望主线程能等待其运行至特定阶段后再继续执行，该如何处理？

注意，此处强调的是“运行至特定阶段”，而非“运行结束”。若只需等待结束，使用join()即可；但若需等待特定阶段，使用join()会拖累整体效率。此时，必须引入Event机制。加入Event后的代码如下：


import time
from threading import Thread, Event

def run_in_thread(event):
    time.sleep(1)
    print('Thread is running')
    # set一下event，这样外面wait的部分就会被启动
    event.set()

# 初始化Event
event = Event()
t = Thread(target=run_in_thread, args=(event, ))
t.start()

# event等待set
event.wait()
print('Main thread print')

整体逻辑改动不大，主要增加了关于Event使用的几行代码。

使用Event时，建议仅在代码中使用一次。尽管可通过clear()方法重置Event状态，但无法保证重置操作一定发生在wait()之前。若重置发生在wait之后，将引发难以调试的Bug，表现为问题非必现、时有时无，这通常源于多线程使用的复杂性。

因此，若需多次使用开关和信号机制，不建议使用Event，而应选用信号量。

三、信号量

Event的缺陷在于：若多个线程等待Event触发，一旦set()被调用，所有等待线程将同时执行。但有时我们希望控制线程逐个运行。Event无法满足此需求，需使用信号量。

信号量与Event用法相似，但区别在于：信号量保证每次仅启动一个线程。两者底层逻辑不同：Event类似开关，触发后所有关联逻辑同时执行；信号量则类似许可证机制，仅持有许可证的线程可执行任务，且每次仅发放一张许可证。

无需自行开发，threading库提供了Semaphore工具，其使用代码如下：


# 工作线程
def worker(n, sema):
    # 等待信号量
    sema.acquire()
    print('Working', n)

# 初始化
sema = threading.Semaphore(0)
nworkers = 10
for n in range(nworkers):
    t = threading.Thread(target=worker, args=(n, sema,))
    t.start()

上述代码创建了10个线程，虽已启动，但均不执行逻辑，因为sema.acquire()是阻塞方法，未获取到信号量时会一直挂起等待。