Python Threading Event对象：如何通知特定线程？

我有多个线程使用threading.Event对象进行带超时的等待操作。调用该Event的set()方法会解除所有线程的阻塞状态，请问有什么好方法可以仅解除特定线程的阻塞，让其他线程继续处于等待状态？

我曾考虑过让每个线程通过while循环结合全局变量作为条件来判断何时返回，但不确定如何在不检查时间戳的情况下为每个线程保留所需的超时功能。

原代码示例：

import threading
import time

t1 = threading.Thread(target=startTimeout)
t2 = threading.Thread(target=startTimeout)
timeoutEvent = threading.Event()
time.sleep(0.3)
timeoutEvent.set()

# 如何为特定线程设置独立的timeoutEvents？

def startTimeout():
  check = timeoutEvent.wait(1)
  if check:
    # 调用了set()
  else:
    # 超时

方法一：为每个线程分配独立的Event对象 #

核心思路是给每个线程绑定专属的threading.Event实例，这样就能单独控制每个线程的唤醒操作，同时保留wait()方法的原生超时功能。

修改后的代码示例：

import threading
import time

def start_timeout(event):
    check = event.wait(1)
    if check:
        print(f"线程{threading.get_ident()}被主动唤醒")
    else:
        print(f"线程{threading.get_ident()}等待超时")

# 为每个线程创建独立的Event
event_t1 = threading.Event()
event_t2 = threading.Event()

# 启动线程时传入各自的Event
t1 = threading.Thread(target=start_timeout, args=(event_t1,))
t2 = threading.Thread(target=start_timeout, args=(event_t2,))

t1.start()
t2.start()

time.sleep(0.3)
# 仅唤醒t1对应的线程
event_t1.set()

t1.join()
t2.join()

该方案简单直接，每个线程的等待状态完全独立，想唤醒哪个线程就调用对应Event的set()即可，超时逻辑由wait(timeout)原生实现，无需额外处理时间戳。

方法二：用字典动态管理线程与Event的映射 #

如果线程数量较多或者需要动态创建线程，可以用字典存储线程ID（或线程对象）与对应Event的映射关系，方便后续精准定位目标线程的Event。

代码示例：

import threading
import time

thread_events = {}

def start_timeout():
    thread_id = threading.get_ident()
    event = thread_events[thread_id]
    check = event.wait(1)
    if check:
        print(f"线程{thread_id}被主动唤醒")
    else:
        print(f"线程{thread_id}等待超时")

# 创建线程并绑定Event
t1 = threading.Thread(target=start_timeout)
thread_events[t1.ident] = threading.Event()
t1.start()

t2 = threading.Thread(target=start_timeout)
thread_events[t2.ident] = threading.Event()
t2.start()

time.sleep(0.3)
# 仅唤醒t2对应的线程
thread_events[t2.ident].set()

t1.join()
t2.join()

这种方式适合动态场景，比如批量创建线程时，可以自动给每个线程分配Event并记录到字典中，后续根据线程标识精准操作。

方法三：使用Condition实现更灵活的唤醒控制 #

如果需要更复杂的唤醒逻辑（比如按自定义条件筛选线程），可以用threading.Condition。Condition的wait()方法同样支持超时，结合线程标记可以实现精准唤醒特定线程。

代码示例：

import threading
import time

cond = threading.Condition()
# 存储线程的标记与唤醒状态
thread_flags = {"t1": False, "t2": False}

def start_timeout(thread_tag):
    with cond:
        # 等待超时或对应的标记被设为True
        check = cond.wait_for(lambda: thread_flags[thread_tag], timeout=1)
        if check:
            print(f"线程{thread_tag}被主动唤醒")
        else:
            print(f"线程{thread_tag}等待超时")

t1 = threading.Thread(target=start_timeout, args=("t1",))
t2 = threading.Thread(target=start_timeout, args=("t2",))

t1.start()
t2.start()

time.sleep(0.3)
# 仅标记t1为可唤醒状态，然后唤醒等待的线程
with cond:
    thread_flags["t1"] = True
    cond.notify()

t1.join()
t2.join()

这里利用wait_for()方法结合自定义条件，只有当对应线程的标记为True时才会结束等待，同时保留了超时功能。调用notify()时，Condition会唤醒第一个满足条件的线程，实现精准唤醒。

内容的提问来源于stack exchange，提问作者questions1010