Python：win32event 库高级用法举例和应用详解

import win32event  # 导入win32event库
import threading    # 导入线程模块
import time         # 导入时间模块

# 创建一个事件对象
event = win32event.CreateEvent(None, 0, 0, None)  # None: 默认安全性; 0: 非手动重置事件; 0: 初始状态为未发信号; None: 不关联任何名字

def worker():
    print("Worker waiting for event...")  # 输出等待提示
    win32event.WaitForSingleObject(event, win32event.INFINITE)  # 等待事件信号
    print("Worker received the signal!")  # 输出信号接收提示

t = threading.Thread(target=worker)  # 创建线程
t.start()  # 启动线程

time.sleep(2)  # 主线程睡眠2秒
print("Main thread setting the event.")  # 输出设置事件的提示
win32event.SetEvent(event)  # 发出事件信号
t.join()  # 等待线程结束

import win32event  # 导入win32event库
import threading    # 导入线程模块
import time         # 导入时间模块

mutex = win32event.CreateMutex(None, 0, None)  # 创建一个互斥量

shared_resource = 0  # 共享资源

def increment():
    global shared_resource  # 使用全局变量
    for _ in range(100):
        win32event.WaitForSingleObject(mutex, win32event.INFINITE)  # 请求互斥量
        shared_resource += 1  # 对共享资源进行操作
        win32event.ReleaseMutex(mutex)  # 释放互斥量

threads = [threading.Thread(target=increment) for _ in range(10)]  # 创建10个线程
for t in threads:
    t.start()  # 启动线程

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

print(f"Final value of shared resource: {shared_resource}")  # 输出共享资源最终值

import win32event  # 导入win32event库
import threading    # 导入线程模块
import time         # 导入时间模块

max_threads = 3  # 并发最大线程数
semaphore = win32event.CreateSemaphore(None, max_threads, max_threads, None)  # 创建信号量

def worker(thread_id):
    win32event.WaitForSingleObject(semaphore, win32event.INFINITE)  # 请求信号量
    print(f"Thread {thread_id} is running.")  # 输出运行提示
    time.sleep(2)  # 模拟工作
    print(f"Thread {thread_id} is finished.")  # 输出结束提示
    win32event.ReleaseSemaphore(semaphore, 1, None)  # 释放信号量

threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]  # 创建5个线程
for t in threads:
    t.start()  # 启动线程

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