开发过程中经常会碰到这样的场景：需要从一个地方获取一些数据，然后处理数据并将其保存在数据库中。
PRivate void FetchData() {}
private void SaveData() {}
static void Main(string[] args)
{
for (int i = 0; i < 10; i++)
{
FetchData(); // 获取数据
SaveData(); // 处理并保存
}
}
例如上述代码例子这样顺序执行，执行会很慢，原因是获取数据和处理并保存的过程都可能导致阻塞，然而FetchData()每次取数据并不需要等待上一条数据保存完成再进行。
这样的场景非常适合用生产者消费者队列：生产者就是FetchData()，用来生产数据；消费者SaveData()，用来消费数据。
举个实际例子，我们需要通过一个Web Api获取一些城市的天气情况，并将其保存到数据库中。
实现方式：
需要一个任务队列，生产者可以向队列中插入任务，消费者可以从任务队列中取出任务来执行。为保证线程安全，使用一个锁来保护这个队列的访问。制定一个退出策略，在所有任务完成时释放资源。下边是实现的完整代码：
classProgram
{//任务队列staticQueue<string> _tasks =newQueue<string>();//为保证线程安全，使用一个锁来保护_task的访问readonlystaticobject_locker =newobject();//通过 _wh 给工作线程发信号staticEventWaitHandle _wh =newAutoResetEvent(false);staticThread _worker;staticvoidMain(string[] args)
{//需要获取天气情况的城市对应代码varcityIds =newList<int> {101280601,101010100,101020100,101110101,101040100};//任务开始，启动工作线程_worker =newThread(Work);
_worker.Start();//生产者将数据插入队里中，并给工作线程发信号foreach(varcityIdincityIds)
EnqueueTask(FetchData(cityId));//任务结束Dispose();
}///<summary>执行工作</summary>staticvoidWork()
{while(true)
{stringwork =null;lock(_locker)
{if(_tasks.Count >0)
{
work= _tasks.Dequeue();//有任务时，出列任务if(work ==null)//退出机制：当遇见一个null任务时，代表任务结束return;
}
}if(work !=null)
SaveData(work);//任务不为null时，处理并保存数据else_wh.WaitOne();//没有任务了，等待信号}
}///<summary>插入任务</summary>staticvoidEnqueueTask(stringtask)
{lock(_locker)
_tasks.Enqueue(task);//向队列中插入任务_wh.Set();//给工作线程发信号}///<summary>结束释放</summary>staticvoidDispose()
{
EnqueueTask(null);//插入一个Null任务，通知工作线程退出_worker.Join();//等待工作线程完成_wh.Close();//释放资源}///<summary>获取数据</summary>staticstringFetchData(intcityId)
{varwc =newWebClient { Encoding =Encoding.UTF8 };varurl =string.Format("http://www.weather.com.cn/adat/sk/{0}.html", cityId);returnwc.DownloadString(url);
}///<summary>处理保存</summary>staticvoidSaveData(stringdata)
{varweatherInfo = (JsonConvert.DeserializeObject(data,typeof(Dictionary<string, Weatherinfo>))asDictionary<string, Weatherinfo>)["weatherinfo"];
Console.WriteLine("[{0}]：{1} 气温（{2}） 风向（{3}） 风力（{4}）", weatherInfo.Time, weatherInfo.City, weatherInfo.Temp, weatherInfo.Wd, weatherInfo.Ws);
Thread.Sleep(200);//模拟数据保存}
}publicclassWeatherinfo
{publicstringCity {get;set; }publicstringTemp {get;set; }publicstringTime {get;set; }publicstringWd {get;set; }publicstringWs {get;set; }
}
}
解释：
Main方法中，我们首先启动了一个工作线程，由于此时队列中没有任务，因此工作线程在等待信号。通过EnqueueTask向队列中插入任务，并通过等待句柄_wh发信号给工作线程，工作线程收到信号后就开始执行处理保存。当生产者获取完所有数据时，插入null任务，并等待工作线程完成。工作线程最后执行到null任务时退出。
参考：Threading in C# --> 中文翻译