C# 实现生产者消费者队列 (转)

按语：按照下面文档，测试成功；

 

https://www.cnblogs.com/samgk/p/4772806.html

开发过程当中常常会碰到这样的场景：须要从一个地方获取一些数据，而后处理数据并将其保存在数据库中。

?

1 2 3 4 5 6 7 8 9 10

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获取一些城市的天气状况，并将其保存到数据库中。

实现方式：

1. 须要一个任务队列，生产者能够向队列中插入任务，消费者能够从任务队列中取出任务来执行。
2. 为保证线程安全，使用一个锁来保护这个队列的访问。
3. 制定一个退出策略，在全部任务完成时释放资源。

下边是实现的完整代码：

class Program { // 任务队列 static Queue<string> _tasks = new Queue<string>(); // 为保证线程安全，使用一个锁来保护_task的访问 readonly static object _locker = new object(); // 经过 _wh 给工做线程发信号 static EventWaitHandle _wh = new AutoResetEvent(false); static Thread _worker; static void Main(string[] args) { // 须要获取天气状况的城市对应代码 var cityIds = new List<int> {101280601, 101010100, 101020100, 101110101, 101040100}; // 任务开始，启动工做线程 _worker = new Thread(Work); _worker.Start(); // 生产者将数据插入队里中，并给工做线程发信号 foreach (var cityId in cityIds) EnqueueTask(FetchData(cityId)); // 任务结束  Dispose(); } /// <summary>执行工做</summary> static void Work() { while (true) { string work = 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> static void EnqueueTask(string task) { lock (_locker) _tasks.Enqueue(task); // 向队列中插入任务  _wh.Set(); // 给工做线程发信号  } /// <summary>结束释放</summary> static void Dispose() { EnqueueTask(null); // 插入一个Null任务，通知工做线程退出 _worker.Join(); // 等待工做线程完成 _wh.Close(); // 释放资源  } /// <summary>获取数据</summary> static string FetchData(int cityId) { var wc = new WebClient { Encoding = Encoding.UTF8 }; var url = string.Format("http://www.weather.com.cn/adat/sk/{0}.html", cityId); return wc.DownloadString(url); } /// <summary>处理保存</summary> static void SaveData(string data) { var weatherInfo = (JsonConvert.DeserializeObject(data, typeof(Dictionary<string, Weatherinfo>)) as Dictionary<string, Weatherinfo>)["weatherinfo"]; Console.WriteLine("[{0}]：{1} 气温（{2}） 风向（{3}） 风力（{4}）", weatherInfo.Time, weatherInfo.City, weatherInfo.Temp, weatherInfo.Wd, weatherInfo.Ws); Thread.Sleep(200); // 模拟数据保存  } } public class Weatherinfo { public string City { get; set; } public string Temp { get; set; } public string Time { get; set; } public string Wd { get; set; } public string Ws { get; set; } } }

 解释：

1. Main方法中，咱们首先启动了一个工做线程，因为此时队列中没有任务，所以工做线程在等待信号。
2. 经过EnqueueTask向队列中插入任务，并经过等待句柄_wh发信号给工做线程，工做线程收到信号后就开始执行处理保存。
3. 当生产者获取完全部数据时，插入null任务，并等待工做线程完成。工做线程最后执行到null任务时退出。