首页 > > 网络编程 > Mysql >

多线程编程学习笔记——使用并发集合（三）

2018-06-17 22:31:57来源：未知阅读 ()

接上文多线程编程学习笔记——使用并发集合（一）

接上文多线程编程学习笔记——使用并发集合（二）

四、使用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方法，这会使用所有工作者完成工作。现在在程序输出中我们看到两个结果序列，演示了并发队列和堆栈集合的不同之处。

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有

上一篇：Unity主线程和子线程跳转调用（1）

下一篇：测试

IDC资讯：主机资讯注册资讯托管资讯 vps资讯网站建设

网站运营：建站经验策划盈利搜索优化网站推广免费资源

网站联盟：联盟新闻联盟介绍联盟点评网赚技巧

行业资讯：搜索引擎网络游戏电子商务广告传媒

网络编程： Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它

服务器技术： Web服务器 Ftp服务器 Mail服务器 Dns服务器安全防护

软件技巧：其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷

网页制作： FrontPages Dreamweaver Javascript css photoshop fireworks Flash

程序设计： Java技术 C/C++ VB delphi

网络知识：网络协议网络安全网络管理组网方案 Cisco技术

操作系统： Win2000 WinXP Win2003 Mac OS Linux FreeBSD

热门词条

最新资讯

热门关注

热门标签