多线程编程学习笔记——使用并发集合（三）
四、 使用ConcurrentBag创建一个可扩展的爬虫
本示例在多个独立的即可生产任务又可消费任务的工作者间如何扩展工作量。
1.程序代码如下。
using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using System.Collections.Concurrent; using System.Diagnostics; using System.Threading; namespace ThreadCollectionDemo { class Program { static Dictionary<string, string[]> contextItems = new Dictionary<string, string[]>(); static void Main(string[] args) { Console.WriteLine(string.Format("----- ConcurrentBag 操作----")); CreateLinks(); Task task = RunBag(); task.Wait(); Console.Read(); } static async Task RunBag() { var taskBag = new ConcurrentBag<CrawlingTask>(); string[] urls = new string[] { "http://www.163.com", "http://www.jd.com", "http://www.hexun.com",
"http://www.tmall.com", "http://www.qq.com" }; var crawlers = new Task[5]; for (int i = 1; i <= 5; i++) { string crawlerName = "Crawler " + i.ToString(); taskBag.Add(new CrawlingTask { UrlToCraw = urls[i - 1], ProductName = "root" }); crawlers[i - 1] = Task.Run(() => Craw(taskBag,crawlerName)); } await Task.WhenAll(crawlers); } static async Task Craw(ConcurrentBag<CrawlingTask> bag, string crawlerName) { CrawlingTask task; while (bag.TryTake(out task)) { Console.WriteLine(" {0} url 从ConcurrentBag 取出，上一节点{1},名称{2}", task.UrlToCraw, task.ProductName, crawlerName); IEnumerable<string> urls = await GetLinksFromContent(task); if (urls != null) { foreach (var url in urls) { var t = new CrawlingTask { UrlToCraw = url, ProductName = crawlerName }; bag.Add(t); } } } if (task != null) { Console.WriteLine("第{0} 个url 添加到ConcurrentBag，线程名称{1},爬虫名称{2}", task.UrlToCraw, task.ProductName, crawlerName); } else Console.WriteLine(" TASK IS NULL "); } static async Task<IEnumerable<string>> GetLinksFromContent(CrawlingTask task) { await GetRandomDely(); if (contextItems.ContainsKey(task.UrlToCraw)) return contextItems[task.UrlToCraw]; return null; } static void CreateLinks() { contextItems["http://www.163.com"] = new[]{ "http://www.163.com/a.html","http://www.163.com/b.html" }; contextItems["http://www.jd.com"] = new[]{ "http://www.jd.com/a.html","http://www.jd.com/b.html" }; contextItems["http://www.qq.com"] = new[]{ "http://www.qq.com/1.html","http://www.qq.com/2.html", "http://www.qq.com/3.html","http://www.qq.com/4.html" }; contextItems["http://www.tmall.com"] = new[]{ "http://www.tmall.com/a.html","http://www.tmall.com/b.html" }; contextItems["http://www.hexun.com"] = new[]{ "http://www.hexun.com/a.html","http://www.hexun.com/b.html", "http://www.hexun.com/c.html","http://www.hexun.com/d.html" }; } static Task GetRandomDely() { int dely = new Random(DateTime.Now.Millisecond).Next(150, 600); return Task.Delay(dely); } } class CrawlingTask { public string UrlToCraw { get; set; } public string ProductName { get; set; } } }
2.程序运行结果，如下图。
这个程序模拟了使用多个网络爬虫进行网页索引的场景。刚开始，我们定义了一个包含网页URL的字典。这个字典模拟了包含其他页面链接的网页。这个实现非常简单，并不关心索引已经访问过的页面，但正因为它如此简单，我们才可能关注并行工作负载。
接着创建了一个并发包，其中包含爬虫任务。我们创建了四个爬虫，并且给每个爬虫都提供了一个不同的网站根URL。然后等等所有爬虫完成工作。现在每个爬虫开始检查提供给它的网站URL。我们通过等待一个随机事件来模拟网络IO处理。如果页面包含的URL越多，爬虫向包中放入的任务也越多。然后检查包中是否还有任何需要爬虫处理的任务，如果没有说明爬虫完成了工作 。
如果检查前四个根URL后的第一行输出，我们将看到爬虫N放置的任务通过会被同一个爬虫处理。然而，接下来的行则会不同。这是因为ConcurrentBag内部针对多个线程既可以添加元素又可以删除元素的场景进行了优化。实现方式是每个线程使用自己的本地队列的元素，所以使用这个队列时无需要任何锁。只有当本地队列中没有任何元素时，我们才执行一些锁定操作并尝试从其他线程的本地队列中“偷取”工作。这种行为 有豕于在所有工作 者间分发工作并避免使用锁。
五、 使用BlockingCollention进行异步处理
本示例学习如何使用BlockingCollection来简化实现 异步处理的工作负载。
1.程序代码如下。
using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using System.Collections.Concurrent; using System.Diagnostics; using System.Threading; namespace ThreadCollectionDemo { class Program { static void Main(string[] args) { Console.WriteLine(string.Format("----- BlockingCollection 操作----")); Console.WriteLine(string.Format("----- BlockingCollection 操作 queue----")); Task task = RunBlock(); task.Wait(); Console.WriteLine(string.Format("----- BlockingCollection 操作 Stack----")); task = RunBlock(new ConcurrentStack<CustomTask>()); task.Wait(); Console.Read(); } static async Task RunBlock(IProducerConsumerCollection<CustomTask> collection = null) { string name = "queue "; var taskBlock = new BlockingCollection<CustomTask>(); if (collection != null) { taskBlock = new BlockingCollection<CustomTask>(collection); name = "stack"; } var taskSrc = Task.Run(() => TaskProduct(taskBlock)); Task[] process = new Task[4]; for (int i = 1; i <= 4; i++) { string processId = i.ToString(); process[i - 1] = Task.Run(() => TaskProcess(taskBlock, name + processId)); } await taskSrc; await Task.WhenAll(process); } static async Task TaskProduct(BlockingCollection<CustomTask> block) { for (int i = 0; i < 20; i++) { await Task.Delay(50); var workitem = new CustomTask { Id = i }; block.Add(workitem); Console.WriteLine(string.Format("把{0} 元素添加到BlockingCollection", workitem.Id)); } block.CompleteAdding(); } static async Task TaskProcess(BlockingCollection<CustomTask> collection, string name) { await GetRandomDely(); foreach (var item in collection) { Console.WriteLine(string.Format("--- Task {0} 处理 操作 名称： {1} ---",item.Id,name)); await GetRandomDely(); } } static Task GetRandomDely() { int dely = new Random(DateTime.Now.Millisecond).Next(1, 1000); return Task.Delay(dely); } } }
2.程序运行结果，如下图。
先说第一个场景，这里我们使用了BlockingCollection类，它带来了很多优势。首先，我们能够改变任务存储在阻塞集合中的方式。默认情况下它使用的是ConcurrentQueue容器，但是我们能够使用任何实现 IProducerConsumerConllection泛型接口的集合。
工作者通过对阻塞集合迭代调用GetConsumingEnumerable方法来获取 工作项。如果在这个集合中没有任何元素，迭代器会阻塞工作线程直到有元素被放到集合中。当生产才调用集合的Completedding时迭代周期会结束。这标志着工作完成。
工作量生产者将任务插入到BlockingCollection，然后调用 CompleteAdding方法，这会使用所有工作 者完成工作 。现在在程序输出中我们看到两个结果序列，演示了并发队列和堆栈集合的不同之处。