Python Event实现线程同步：生产者与消费者模型详解

清欢

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

原创

本文通过Python中的Event对象，详细演示了如何利用事件机制实现线程间的同步通信。以生产者线程和工作者线程为例，展示了如何在10次交互中通过wait和set方法协调资源状态，确保线程安全与执行顺序。

Python Event线程同步概述

前言

在Python多线程编程中，Event对象是实现线程间同步的常用手段。本文将通过一个典型的生产者与消费者模型，具体演示如何利用Event对象完成10次线程间的同步交互。

核心代码实现分析

import threading
from threading import Event, Thread
import time
import random
from time import sleep
 
pevent = Event() #默认是没有事件的
pevent.clear()
 
cevent = Event() 
cevent.clear()
 
runtimes = 10
mutex_lock = threading.Lock()
 
 
class ProducerThread (threading.Thread):
    def __init__(self, name, runflag):
        threading.Thread.__init__(self)
        self.name = name
        self.runflag = runflag
        self.continueflag = Event()
        self.continueflag.set()
        
    def run(self):
        global runtimes  
        sleep(1)
        print ("开始线程：" + self.name)
        while self.continueflag.isSet():
            print("wait consumer ...")
            if runtimes == 0:
                self.continueflag.clear()
                break
            pevent.wait()
            print("come an consumer ...")
            mutex_lock.acquire()
            runtimes = runtimes - 1
            mutex_lock.release()
            pevent.clear()
            sleep(1)
            cevent.set()
        print ("退出线程：" + self.name)
        self.runflag.set()
 
class ConsumerThread (threading.Thread):
    def __init__(self,name, runflag):
        threading.Thread.__init__(self)
        self.name = name
        self.runflag = runflag
        self.continueflag = Event()
        self.continueflag.set()
        
    def run(self):
        global runtimes 
        print ("开始线程：" + self.name)
        while self.continueflag.isSet():
            if 0 == runtimes:
                self.continueflag.clear() 
                pevent.set()
                break
            print("I want to consum ... ", runtimes)
            pevent.set() #通知生产者要消费
            cevent.wait()
            cevent.clear()
            sleep(1)
        print ("退出线程：" + self.name)
        self.runflag.set()
 
def test_pthread():
    runflag = Event() 
    pt = ProducerThread("producer", runflag)
    ct = ConsumerThread("consumer", runflag)
    pt.start()
    ct.start()
    pt.join()
    ct.join()
    runflag.wait()
 
if __name__ == '__main__':
    print('===============begin=================')
    test_pthread()
    print('===============end=================')