Python Event事件详解 - 线程间通信与同步控制

# 线程同步Event相关语法示例
import threading

# 创建事件对象
event = threading.Event()

# 设置事件（标志位设为True）
event.set()

# 清除事件（标志位设为False）
event.clear()

# 等待事件（阻塞直到标志位为True）
event.wait()           # 无限等待
event.wait(timeout=5)  # 带超时等待

# 检查事件状态
event.is_set()  # 返回True或False

# 基本使用模式
def worker(event):
    print("等待事件...")
    event.wait()  # 阻塞直到event被set
    print("事件已触发，继续执行")

# 创建线程
t = threading.Thread(target=worker, args=(event,))
t.start()

# 主线程设置事件
event.set()  # 唤醒等待的线程

# 基本用法示例
import threading
import time

# 创建事件对象
start_event = threading.Event()
stop_event = threading.Event()

def worker(name):
    """工作线程"""
    print(f"[{name}] 启动，等待开始信号...")
    start_event.wait()  # 等待开始信号
    
    print(f"[{name}] 开始工作")
    while not stop_event.is_set():
        print(f"[{name}] 工作中...")
        time.sleep(1)
    
    print(f"[{name}] 收到停止信号，退出")

# 创建多个工作线程
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(f"Worker-{i}",))
    threads.append(t)
    t.start()

# 等待2秒后发送开始信号
time.sleep(2)
print("发送开始信号")
start_event.set()

# 工作5秒后发送停止信号
time.sleep(5)
print("发送停止信号")
stop_event.set()

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

print("所有线程已退出")

# 详细代码示例：生产者消费者模式使用Event
import threading
import time
import random

class DataBuffer:
    """数据缓冲区"""
    def __init__(self, max_size=10):
        self.buffer = []
        self.max_size = max_size
        self.lock = threading.Lock()
        self.not_empty = threading.Event()  # 缓冲区非空事件
        self.not_full = threading.Event()   # 缓冲区未满事件
        self.not_full.set()  # 初始时缓冲区未满

class Producer:
    """生产者"""
    def __init__(self, buffer, name):
        self.buffer = buffer
        self.name = name
        self.running = True
    
    def produce(self):
        while self.running:
            # 等待缓冲区有空间
            self.buffer.not_full.wait()
            
            with self.buffer.lock:
                if len(self.buffer.buffer) < self.buffer.max_size:
                    data = f"{self.name}-data-{random.randint(1, 100)}"
                    self.buffer.buffer.append(data)
                    print(f"[{self.name}] 生产: {data}, 缓冲区大小: {len(self.buffer.buffer)}")
                    
                    # 设置非空事件，唤醒消费者
                    self.buffer.not_empty.set()
                    
                    # 如果缓冲区满了，清除未满事件
                    if len(self.buffer.buffer) >= self.buffer.max_size:
                        self.buffer.not_full.clear()
            
            time.sleep(random.uniform(0.5, 1.5))

class Consumer:
    """消费者"""
    def __init__(self, buffer, name):
        self.buffer = buffer
        self.name = name
        self.running = True
    
    def consume(self):
        while self.running:
            # 等待缓冲区非空
            self.buffer.not_empty.wait()
            
            with self.buffer.lock:
                if self.buffer.buffer:
                    data = self.buffer.buffer.pop(0)
                    print(f"[{self.name}] 消费: {data}, 缓冲区大小: {len(self.buffer.buffer)}")
                    
                    # 设置未满事件，唤醒生产者
                    self.buffer.not_full.set()
                    
                    # 如果缓冲区空了，清除非空事件
                    if not self.buffer.buffer:
                        self.buffer.not_empty.clear()
            
            time.sleep(random.uniform(0.8, 2.0))

# 创建缓冲区
buffer = DataBuffer(max_size=5)

# 创建生产者和消费者
producers = [Producer(buffer, f"Producer-{i}") for i in range(2)]
consumers = [Consumer(buffer, f"Consumer-{i}") for i in range(3)]

# 启动线程
producer_threads = []
consumer_threads = []

for p in producers:
    t = threading.Thread(target=p.produce)
    t.start()
    producer_threads.append(t)

for c in consumers:
    t = threading.Thread(target=c.consume)
    t.start()
    consumer_threads.append(t)

# 运行10秒
time.sleep(10)

# 停止所有线程
for p in producers:
    p.running = False
for c in consumers:
    c.running = False

# 等待线程结束
for t in producer_threads + consumer_threads:
    t.join()

print("生产消费完成")