09 使用自动重置事件进行信号通知

JessQiu

514人浏览 · 2025-08-26 19:44:30

JessQiu · 2025-08-26 19:44:30 发布

使用自动重置事件进行信号通知

知识点

什么是AutoResetEvent

AutoResetEvent是一种同步原语，用于在线程间发送信号通知。它就像一个开关，可以处于信号状态（开）或非信号状态（关），当线程等待时如果事件处于信号状态，线程会继续执行并自动将事件重置为非信号状态。

AutoResetEvent的特点

自动重置：当一个等待的线程被释放后，事件自动重置为非信号状态
单线程唤醒：每次Set()操作只能唤醒一个等待的线程
二进制状态：只有信号和非信号两种状态
线程安全：多个线程可以安全地调用WaitOne()和Set()

核心方法

WaitOne()：等待事件信号
WaitOne(timeout)：带超时的等待
Set()：设置事件为信号状态，唤醒一个等待线程
Reset()：手动将事件重置为非信号状态

使用场景

生产者通知消费者有新数据
主线程等待工作线程完成
实现简单的任务调度
线程间的握手通信

代码案例

案例1：基本的生产者消费者通知

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

class BasicAutoResetEventExample
{
    private static readonly AutoResetEvent dataAvailable = new AutoResetEvent(false);
    private static readonly Queue<string> dataQueue = new Queue<string>();
    private static readonly object queueLock = new object();
    private static volatile bool isProducing = true;
    
    static void Main(string[] args)
    {
        Console.WriteLine("AutoResetEvent基本示例");
        
        // 启动生产者
        Task producer = Task.Run(() => Producer());
        
        // 启动消费者
        Task consumer = Task.Run(() => Consumer());
        
        // 运行5秒后停止
        Thread.Sleep(5000);
        isProducing = false;
        
        // 发送最后一个信号让消费者检查退出条件
        dataAvailable.Set();
        
        Task.WaitAll(producer, consumer);
        Console.WriteLine("程序结束");
        
        dataAvailable.Dispose();
    }
    
    static void Producer()
    {
        int itemCount = 1;
        
        while (isProducing)
        {
            // 生产数据
            string data = $"Data_{itemCount++}_{DateTime.Now:HH:mm:ss.fff}";
            
            lock (queueLock)
            {
                dataQueue.Enqueue(data);
                Console.WriteLine($"生产者: 生产了 {data}，队列长度: {dataQueue.Count}");
            }
            
            // 通知消费者有新数据可用
            dataAvailable.Set();
            
            // 模拟生产时间
            Thread.Sleep(800);
        }
        
        Console.WriteLine("生产者: 停止生产");
    }
    
    static void Consumer()
    {
        while (true)
        {
            Console.WriteLine("消费者: 等待数据...");
            
            // 等待数据可用信号
            if (dataAvailable.WaitOne(2000)) // 2秒超时
            {
                string data = null;
                
                lock (queueLock)
                {
                    if (dataQueue.Count > 0)
                    {
                        data = dataQueue.Dequeue();
                    }
                }
                
                if (data != null)
                {
                    Console.WriteLine($"消费者: 消费了 {data}");
                    
                    // 模拟处理时间
                    Thread.Sleep(500);
                }
                else if (!isProducing)
                {
                    Console.WriteLine("消费者: 生产结束且队列为空，退出");
                    break;
                }
            }
            else
            {
                Console.WriteLine("消费者: 等待数据超时");
                
                if (!isProducing)
                {
                    lock (queueLock)
                    {
                        if (dataQueue.Count == 0)
                        {
                            Console.WriteLine("消费者: 生产结束且队列为空，退出");
                            break;
                        }
                    }
                }
            }
        }
        
        Console.WriteLine("消费者: 已退出");
    }
}

案例2：任务协调器

using System;
using System.Threading;
using System.Threading.Tasks;

class TaskCoordinator
{
    private readonly AutoResetEvent taskReady = new AutoResetEvent(false);
    private readonly AutoResetEvent taskCompleted = new AutoResetEvent(false);
    private volatile string currentTask = null;
    private volatile bool isRunning = true;
    
    public void Start()
    {
        Console.WriteLine("任务协调器启动");
        
        // 启动工作线程
        Task workerTask = Task.Run(() => WorkerThread());
        
        // 主线程分配任务
        Task coordinatorTask = Task.Run(() => CoordinatorThread());
        
        Task.WaitAll(workerTask, coordinatorTask);
        
        taskReady.Dispose();
        taskCompleted.Dispose();
    }
    
    private void CoordinatorThread()
    {
        Console.WriteLine("协调线程: 开始分配任务");
        
        for (int i = 1; i <= 8; i++)
        {
            // 准备新任务
            currentTask = $"任务_{i}_{DateTime.Now:HH:mm:ss.fff}";
            Console.WriteLine($"协调线程: 分配 {currentTask}");
            
            // 通知工作线程有新任务
            taskReady.Set();
            
            // 等待任务完成
            Console.WriteLine($"协调线程: 等待 {currentTask} 完成...");
            if (taskCompleted.WaitOne(5000)) // 5秒超时
            {
                Console.WriteLine($"协调线程: {currentTask} 已完成");
            }
            else
            {
                Console.WriteLine($"协调线程: {currentTask} 超时");
            }
            
            // 模拟准备下一个任务的时间
            Thread.Sleep(1000);
        }
        
        isRunning = false;
        taskReady.Set(); // 通知工作线程退出
        Console.WriteLine("协调线程: 所有任务分配完成");
    }
    
    private void WorkerThread()
    {
        Console.WriteLine("工作线程: 准备接收任务");
        
        while (isRunning)
        {
            Console.WriteLine("工作线程: 等待新任务...");
            
            // 等待新任务信号
            if (taskReady.WaitOne())
            {
                if (!isRunning) break; // 检查是否需要退出
                
                if (currentTask != null)
                {
                    Console.WriteLine($"工作线程: 开始执行 {currentTask}");
                    
                    // 模拟任务执行
                    ExecuteTask(currentTask);
                    
                    Console.WriteLine($"工作线程: 完成 {currentTask}");
                    
                    // 通知协调线程任务完成
                    taskCompleted.Set();
                }
            }
        }
        
        Console.WriteLine("工作线程: 已退出");
    }
    
    private void ExecuteTask(string taskName)
    {
        // 模拟不同的任务执行时间
        int executionTime = new Random().Next(1000, 3000);
        Console.WriteLine($"工作线程: {taskName} 执行中（预计 {executionTime}ms）...");
        Thread.Sleep(executionTime);
    }
}

class TaskCoordinatorExample
{
    static void Main(string[] args)
    {
        Console.WriteLine("任务协调器示例");
        
        var coordinator = new TaskCoordinator();
        coordinator.Start();
        
        Console.WriteLine("任务协调器示例完成");
    }
}

案例3：多阶段处理流水线

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

class ProcessingPipeline
{
    private readonly AutoResetEvent stage1Ready = new AutoResetEvent(false);
    private readonly AutoResetEvent stage2Ready = new AutoResetEvent(false);
    private readonly AutoResetEvent stage3Ready = new AutoResetEvent(false);
    
    private readonly ConcurrentQueue<WorkItem> stage1Queue = new ConcurrentQueue<WorkItem>();
    private readonly ConcurrentQueue<WorkItem> stage2Queue = new ConcurrentQueue<WorkItem>();
    private readonly ConcurrentQueue<WorkItem> stage3Queue = new ConcurrentQueue<WorkItem>();
    
    private volatile bool isRunning = true;
    
    class WorkItem
    {
        public int Id { get; set; }
        public string Data { get; set; }
        public DateTime CreatedTime { get; set; }
        public DateTime Stage1CompletedTime { get; set; }
        public DateTime Stage2CompletedTime { get; set; }
        public DateTime Stage3CompletedTime { get; set; }
    }
    
    public void Start()
    {
        Console.WriteLine("流水线处理示例");
        
        // 启动各个阶段的处理线程
        Task inputTask = Task.Run(() => InputStage());
        Task stage1Task = Task.Run(() => ProcessingStage1());
        Task stage2Task = Task.Run(() => ProcessingStage2());
        Task stage3Task = Task.Run(() => ProcessingStage3());
        
        // 运行8秒后停止输入
        Thread.Sleep(8000);
        isRunning = false;
        
        // 等待输入阶段完成
        inputTask.Wait();
        
        // 发送信号让各个阶段处理完剩余数据
        stage1Ready.Set();
        stage2Ready.Set();
        stage3Ready.Set();
        
        Task.WaitAll(stage1Task, stage2Task, stage3Task);
        
        Console.WriteLine("流水线处理完成");
        
        // 清理资源
        stage1Ready.Dispose();
        stage2Ready.Dispose();
        stage3Ready.Dispose();
    }
    
    private void InputStage()
    {
        int itemId = 1;
        
        while (isRunning)
        {
            var workItem = new WorkItem
            {
                Id = itemId++,
                Data = $"Input_Data_{itemId}",
                CreatedTime = DateTime.Now
            };
            
            stage1Queue.Enqueue(workItem);
            Console.WriteLine($"输入阶段: 创建工作项 {workItem.Id}");
            
            // 通知阶段1有新数据
            stage1Ready.Set();
            
            Thread.Sleep(1000); // 模拟输入间隔
        }
        
        Console.WriteLine("输入阶段: 停止创建新工作项");
    }
    
    private void ProcessingStage1()
    {
        Console.WriteLine("阶段1: 准备处理");
        
        while (true)
        {
            // 等待工作项信号
            stage1Ready.WaitOne();
            
            // 处理所有可用的工作项
            while (stage1Queue.TryDequeue(out WorkItem workItem))
            {
                Console.WriteLine($"阶段1: 处理工作项 {workItem.Id}");
                
                // 模拟阶段1处理
                Thread.Sleep(800);
                
                // 修改数据
                workItem.Data = $"Stage1_Processed_{workItem.Data}";
                workItem.Stage1CompletedTime = DateTime.Now;
                
                // 传递到阶段2
                stage2Queue.Enqueue(workItem);
                stage2Ready.Set();
                
                Console.WriteLine($"阶段1: 完成工作项 {workItem.Id}，传递到阶段2");
            }
            
            // 检查是否应该退出
            if (!isRunning && stage1Queue.IsEmpty)
            {
                // 通知阶段2没有更多数据
                stage2Ready.Set();
                break;
            }
        }
        
        Console.WriteLine("阶段1: 处理完成");
    }
    
    private void ProcessingStage2()
    {
        Console.WriteLine("阶段2: 准备处理");
        
        while (true)
        {
            // 等待工作项信号
            stage2Ready.WaitOne();
            
            // 处理所有可用的工作项
            while (stage2Queue.TryDequeue(out WorkItem workItem))
            {
                Console.WriteLine($"阶段2: 处理工作项 {workItem.Id}");
                
                // 模拟阶段2处理
                Thread.Sleep(600);
                
                // 修改数据
                workItem.Data = $"Stage2_Enhanced_{workItem.Data}";
                workItem.Stage2CompletedTime = DateTime.Now;
                
                // 传递到阶段3
                stage3Queue.Enqueue(workItem);
                stage3Ready.Set();
                
                Console.WriteLine($"阶段2: 完成工作项 {workItem.Id}，传递到阶段3");
            }
            
            // 检查是否应该退出
            if (!isRunning && stage1Queue.IsEmpty && stage2Queue.IsEmpty)
            {
                // 通知阶段3没有更多数据
                stage3Ready.Set();
                break;
            }
        }
        
        Console.WriteLine("阶段2: 处理完成");
    }
    
    private void ProcessingStage3()
    {
        Console.WriteLine("阶段3: 准备处理");
        
        while (true)
        {
            // 等待工作项信号
            stage3Ready.WaitOne();
            
            // 处理所有可用的工作项
            while (stage3Queue.TryDequeue(out WorkItem workItem))
            {
                Console.WriteLine($"阶段3: 处理工作项 {workItem.Id}");
                
                // 模拟阶段3处理
                Thread.Sleep(400);
                
                workItem.Data = $"Final_{workItem.Data}";
                workItem.Stage3CompletedTime = DateTime.Now;
                
                // 输出最终结果
                Console.WriteLine($"阶段3: 完成工作项 {workItem.Id} - {workItem.Data}");
                
                // 显示处理时间统计
                var totalTime = workItem.Stage3CompletedTime - workItem.CreatedTime;
                var stage1Time = workItem.Stage1CompletedTime - workItem.CreatedTime;
                var stage2Time = workItem.Stage2CompletedTime - workItem.Stage1CompletedTime;
                var stage3Time = workItem.Stage3CompletedTime - workItem.Stage2CompletedTime;
                
                Console.WriteLine($"  时间统计 - 总计: {totalTime.TotalMilliseconds:F0}ms, " +
                                $"阶段1: {stage1Time.TotalMilliseconds:F0}ms, " +
                                $"阶段2: {stage2Time.TotalMilliseconds:F0}ms, " +
                                $"阶段3: {stage3Time.TotalMilliseconds:F0}ms");
            }
            
            // 检查是否应该退出
            if (!isRunning && stage1Queue.IsEmpty && stage2Queue.IsEmpty && stage3Queue.IsEmpty)
            {
                break;
            }
        }
        
        Console.WriteLine("阶段3: 处理完成");
    }
}

class ProcessingPipelineExample
{
    static void Main(string[] args)
    {
        var pipeline = new ProcessingPipeline();
        pipeline.Start();
    }
}

案例4：线程池任务调度

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

class CustomThreadPool
{
    private readonly AutoResetEvent workAvailable = new AutoResetEvent(false);
    private readonly ConcurrentQueue<Action> workQueue = new ConcurrentQueue<Action>();
    private readonly Thread[] workers;
    private volatile bool isShutdown = false;
    
    public CustomThreadPool(int workerCount)
    {
        workers = new Thread[workerCount];
        
        for (int i = 0; i < workerCount; i++)
        {
            int workerId = i;
            workers[i] = new Thread(() => WorkerThread(workerId))
            {
                IsBackground = true,
                Name = $"Worker-{workerId}"
            };
            workers[i].Start();
        }
        
        Console.WriteLine($"自定义线程池启动，工作线程数: {workerCount}");
    }
    
    public void QueueWork(Action work)
    {
        if (isShutdown)
        {
            throw new InvalidOperationException("线程池已关闭");
        }
        
        workQueue.Enqueue(work);
        workAvailable.Set(); // 通知有新工作可用
    }
    
    public void Shutdown(bool waitForCompletion = true)
    {
        Console.WriteLine("开始关闭线程池...");
        isShutdown = true;
        
        // 通知所有工作线程检查关闭状态
        for (int i = 0; i < workers.Length; i++)
        {
            workAvailable.Set();
        }
        
        if (waitForCompletion)
        {
            foreach (var worker in workers)
            {
                worker.Join();
            }
        }
        
        workAvailable.Dispose();
        Console.WriteLine("线程池已关闭");
    }
    
    private void WorkerThread(int workerId)
    {
        Console.WriteLine($"工作线程{workerId}: 启动");
        
        while (!isShutdown)
        {
            // 等待工作可用信号
            workAvailable.WaitOne();
            
            // 检查是否需要退出
            if (isShutdown) break;
            
            // 处理所有可用的工作项
            while (workQueue.TryDequeue(out Action work))
            {
                try
                {
                    Console.WriteLine($"工作线程{workerId}: 执行任务");
                    work();
                    Console.WriteLine($"工作线程{workerId}: 任务完成");
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"工作线程{workerId}: 任务异常 - {ex.Message}");
                }
            }
        }
        
        Console.WriteLine($"工作线程{workerId}: 已退出");
    }
    
    public int QueuedWorkCount => workQueue.Count;
}

class CustomThreadPoolExample
{
    static void Main(string[] args)
    {
        Console.WriteLine("自定义线程池示例");
        
        // 创建包含3个工作线程的线程池
        var threadPool = new CustomThreadPool(3);
        
        // 提交多个任务
        for (int i = 1; i <= 10; i++)
        {
            int taskId = i;
            threadPool.QueueWork(() => SimulateWork(taskId));
            Console.WriteLine($"主线程: 提交任务 {taskId}，队列中任务数: {threadPool.QueuedWorkCount}");
            
            Thread.Sleep(200); // 模拟任务提交间隔
        }
        
        Console.WriteLine("主线程: 所有任务已提交");
        
        // 等待一段时间让任务完成
        Thread.Sleep(5000);
        
        // 提交一些快速任务
        Console.WriteLine("主线程: 提交快速任务");
        for (int i = 11; i <= 15; i++)
        {
            int taskId = i;
            threadPool.QueueWork(() => QuickWork(taskId));
        }
        
        // 等待快速任务完成
        Thread.Sleep(2000);
        
        // 关闭线程池
        threadPool.Shutdown();
        
        Console.WriteLine("程序结束");
    }
    
    static void SimulateWork(int taskId)
    {
        var random = new Random();
        int workTime = random.Next(1000, 3000);
        
        Console.WriteLine($"  任务{taskId}: 开始工作（预计 {workTime}ms）");
        Thread.Sleep(workTime);
        Console.WriteLine($"  任务{taskId}: 工作完成");
    }
    
    static void QuickWork(int taskId)
    {
        Console.WriteLine($"  快速任务{taskId}: 执行");
        Thread.Sleep(300);
        Console.WriteLine($"  快速任务{taskId}: 完成");
    }
}

知识点总结

AutoResetEvent的核心特性：
- 自动重置机制：每次唤醒一个线程后自动重置
- 二进制状态：只有信号和非信号两种状态
- 线程安全：多线程环境下安全使用
使用场景：
- 生产者消费者通知
- 任务协调和调度
- 流水线处理
- 线程同步点
最佳实践：
- 总是使用using或显式Dispose释放资源
- 设置合理的超时时间避免无限等待
- 结合其他同步机制实现复杂协调
- 避免在持有锁时调用WaitOne
性能考虑：
- AutoResetEvent比ManualResetEvent稍慢
- 频繁的Set/WaitOne操作有性能开销
- 适用于低频率的线程通信
常见陷阱：
- 忘记处理超时情况
- 在信号发送前就开始等待
- 没有正确处理退出条件
- 资源泄漏问题