《C# 爬虫破境之道》：第二境爬虫应用 — 第七节：并发控制与计谋

小白学 Python 数据分析（6）：Pandas （五）基础操作（2）数据选择

我们在第五节中提到一个问题，使命行列增长速率太快，与之对应的收集、剖析、处置惩罚速率远远跟不上，形成内存快速增长，带宽占用太高，CPU运用率太高，如许是极端有害系统康健的。

我们在开发收集程序的时刻，老是愿望能够尽快将数据爬取下来，假如总使命数目很小（2~3K请求数以内），总斲丧时长很短（1~2分钟以内），那末，对系统的平常运转不会形成太严峻的影响，我们尽能够毫无所惧。但，当总使命数目更多，总斲丧时长更长，那末，无休止的使命聚集，就会给系统带来难以预料以至是很严峻的效果。

为此，我们不能不斟酌几个问题：

我们的使命总量也许在什么量级，全速收集也许须要斲丧若干时候、若干资本，将来的生长是不是是可控？
收集系统自身依托的环境资本是不是足够，是不是能够满足随之而来的庞大的资本斲丧？
收集的目标资本系统是不是具有某些反爬战略限定？
收集的目标资本系统是不是能够蒙受得住云云数目级的并发收集请求（不管单点或分布式收集系统，都要斟酌这点）？
跟着收集效果返回，带来的后续剖析、处置惩罚、存储才是不是能够满足大批数据的瞬时到来？

由以上问题也能够看出，一个爬虫系统战略的制订，须要斟酌的问题也是全方位的，而不仅仅是收集自身，差别的环境、范围、目标，采纳的战略也不尽相同。本节，我们将讨论一下，假如我们的才不能满足上述前提的状况下，怎样来制订一个并发战略以及怎样完成它。

并发战略，从范围上能够分为全局并发战略和单点并发战略，全局并发战略包括单点并发战略，不过它也须要同时斟酌负载平衡战略对制订并发战略的影响。现在，我们还没有将爬虫框架扩展到分布式框架，临时先不斟酌全局并发战略的制订。重要讨论一下单点并发战略制订与完成。

单点并发战略的制订：

平常，我们在制订单点并发战略时，须要从哪些角度斟酌，运用什么要领，以及怎样决议计划？下面我们就来细致聊聊：）

1、我们先来梳理一下收集系统自身所依托的环境资本：

除了CPU、内存、存储器、带宽这些耳熟能详的资本外，另有就是比较轻易被疏忽的操作系统可用端口。关于种种资本的占用状况，下面给出一些发起（均值）：

CPU：收集系统的占用总量发起不凌驾30%，CPU总运用量发起不凌驾50%。（虽然我这个疯子常常贪欲过渡T_T）。关于多核CPU，线程建立数目发起不凌驾CPU核数的两倍。
内存：收集系统的占用总量发起不凌驾50%，内存总运用量发起不凌驾70%。
存储器：关于贸易或许大范围的爬虫系统，发起将存储星散，运用外部存储装备，比方NAS、分布式缓存、数据仓库等；固然，其他爬虫系统也这么发起，但假如前提不许可的话，只能存储在当地磁盘的话，就须要斟酌磁盘的IOPS了，即使是运用缓存、数据库系统来作为中心存储序言，实质上也是与磁盘IO打交道，不过平常的缓存、数据库系统都会对IO做优化，而且能干涉干与的力度比较小，却是能够稍微“费心”。这个，本人也没法给出一个合理的通用的发起值，磁盘的机能光怪陆离，只能是按实际环境来拿捏了。
带宽：分为上行、下行两个带宽目标，收集系统在这两个目标中的占用总量都不发起凌驾80%。除了斟酌ISP分派的带宽，还要斟酌会影响其效能的周边装备，比方猫、交换机、路由器以至是网线的吞吐才。说来为难，我常常在家里做试验，爬虫系统和目标资本系统都还OK，联通的光猫跪了……重启回生……又跪了……重启回生……又跪了……重启回生……
可用端口：这个是一个隐性前提，也是常常被疏忽的限定。拿Windows系统来讲，可用的端口最大数目为UInt16.MaxValue(65535)个，而伴跟着系统启动，就会有一系列的效劳占用了部份端口，比方IIS中的网站、数据库、QQ，而系统自身也会保存一部份端口，比方443、3389等。而是不是能够运用端口重用手艺来减缓痛苦悲伤，对详细完成以及NAS端口映射划定规矩的请求更高，不好或不可控。所以爬虫自身能够运用的端口数就有一个极限限定，这个也没有发起值，详细状况各不相同。

总之，资本老是有限的，大致准绳就是：做人留一线，往后好相见：）

2、关于目标资本系统的资本环境：

平常，我们没法探知详细的资本状况，再加上对方大概运用反爬战略，就是晓得详细的资本状况，也不见得就有效。关于制订并发战略，我们更体贴的是对方能够吃的下多大的鸭梨，以及探究其反爬战略许可的极限。为此，我们须要运用下述的要领，来辅佐我们制订战略。

3、平常运用的要领：

3.1、须要找到目标资本系统中的一个URI，准绳是轻量、成功率高（最好是100%），比方，一张小小的图片、一个简朴短小的ajax接口、一个静态html以至是一个xxx.min.css，但要注重，我们拔取的URI可万万不如果经由CDN加快的，不然T_T；

3.2、接下来，我们就针对拔取的URI举行周期收集，关于平常的资本站点，初始频次设置为1秒1次，就能够了；

3.3、然后就是运转一段时候视察效果，背面我们再说运转多长时候视察为适宜；

3.4、假如视察效果OK，成功率能够到达95%+，那末，我们就能够恰当减少收集周期，反之，就要恰当延伸收集周期；

3.5、反复3.3~3.4，末了获得一个合理的极限周期；

3.6、至于一次视察多长时候，差别的反爬战略，有着差别的限定，这个须要警惕。我曾的一个项目，当时就比较心急，观察了5分钟，没什么问题，就丢出去了，效果厥后实际告诉我，他们的战略是1分钟累计限定、10分钟累计限定、20分钟累计限定、30分钟累计限定、1小时累计限定……而且累计限定逐级递减，也就是说，你满足了1分钟的累计限定，x10，就不愿定满足10分钟的累计限定，x60就有大概远远超出了1小时累计限定。这里给出一个发起，最少30分钟。由于目标系统去统计每个泉源IP的接见周期，也是一个不小的价值，所以也不大概做到无穷期的监测，平常半小时到一小时已经是极限了。这里也给出一个最保险的观察周期，那就是依据请求总量及当前频次，预估斲丧总时长，作为观察周期，如许是最稳妥的，但，这也多是不切实际的：（

4、怎样制订并发战略：

经由过程上述3步，连系自身的资本状况、目标的反爬战略及蒙受才、以及观察效果，我们就能够制订一个也许的并发量了，制订决议计划也就不那末困难了；

我们的使命都是存储在行列中，并发的限定，不过就是掌握入队的频次，所以，只须要把前面的统计效果转化为最小请求距离，就是我们终究的并发战略了；

为何是掌握入队，而不是出队呢？由于假如不掌握入队，那末行列照样会无穷暴增，直至“殒命”，而限定入队，一方面防止行列暴增，另一方面，壅塞新使命的生成，下降CPU及内存运用量；

单点并发战略的完成：

有了理论基本，在手艺完成上，就不是什么难事儿了。

 1 namespace MikeWare.Core.Components.CrawlerFramework.Policies
 2 {
 3     using System;
 4 
 5     public abstract class AConcurrentPolicy
 6     {
 7         public virtual bool WaitOne(TimeSpan timeout) => throw new NotImplementedException();
 8 
 9         public virtual void ReleaseOne() => throw new NotImplementedException();
10     }
11 }

并发战略 —— AConcurrentPolicy

这是一个笼统类，具有两个笼统要领，作为并发战略的基本完成；

我写了两种并发战略的详细完成，PeriodConcurrentPolicy和SemaphoreConcurrentPolicy，他们的目标都是用来掌握入队的频次，目标一致，要领差别，您也能够完成本身的并发战略；

本节，我们重要说道说道System.Threading.Semaphore的运用及SemaphoreConcurrentPolicy的完成道理；

 1 namespace MikeWare.Core.Components.CrawlerFramework.Policies
 2 {
 3     using System;
 4     using System.Threading;
 5 
 6     public class SemaphoreConcurrentPolicy : AConcurrentPolicy
 7     {
 8         private Semaphore semaphore = null;
 9 
10         public SemaphoreConcurrentPolicy(int init, int max)
11         {
12             semaphore = new Semaphore(init, max);
13         }
14 
15         public override bool WaitOne(TimeSpan timeout)
16         {
17             return semaphore.WaitOne(timeout);
18         }
19 
20         public override void ReleaseOne()
21         {
22             semaphore.Release(1);
23         }
24     }
25 }

并发战略完成 —— SemaphoreConcurrentPolicy

SemaphoreConcurrentPolicy继续自AConcurrentPolicy，定义了一个私有变量Semaphore semaphore，以及重写了基类的两个笼统要领；

namespace System.Threading
{
    //
    // Summary:
    //     Limits the number of threads that can access a resource or pool of resources
    //     concurrently.
    public sealed class Semaphore : WaitHandle
    {
        //
        // Summary:
        //     Initializes a new instance of the System.Threading.Semaphore class, specifying
        //     the initial number of entries and the maximum number of concurrent entries.
        //
        // Parameters:
        //   initialCount:
        //     The initial number of requests for the semaphore that can be granted concurrently.
        //
        //   maximumCount:
        //     The maximum number of requests for the semaphore that can be granted concurrently.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     initialCount is greater than maximumCount.
        //
        //   T:System.ArgumentOutOfRangeException:
        //     maximumCount is less than 1. -or- initialCount is less than 0.
        public Semaphore(int initialCount, int maximumCount);
        //
        // Summary:
        //     Initializes a new instance of the System.Threading.Semaphore class, specifying
        //     the initial number of entries and the maximum number of concurrent entries, and
        //     optionally specifying the name of a system semaphore object.
        //
        // Parameters:
        //   initialCount:
        //     The initial number of requests for the semaphore that can be granted concurrently.
        //
        //   maximumCount:
        //     The maximum number of requests for the semaphore that can be granted concurrently.
        //
        //   name:
        //     The name of a named system semaphore object.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     initialCount is greater than maximumCount. -or- name is longer than 260 characters.
        //
        //   T:System.ArgumentOutOfRangeException:
        //     maximumCount is less than 1. -or- initialCount is less than 0.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred.
        //
        //   T:System.UnauthorizedAccessException:
        //     The named semaphore exists and has access control security, and the user does
        //     not have System.Security.AccessControl.SemaphoreRights.FullControl.
        //
        //   T:System.Threading.WaitHandleCannotBeOpenedException:
        //     The named semaphore cannot be created, perhaps because a wait handle of a different
        //     type has the same name.
        public Semaphore(int initialCount, int maximumCount, string name);
        //
        // Summary:
        //     Initializes a new instance of the System.Threading.Semaphore class, specifying
        //     the initial number of entries and the maximum number of concurrent entries, optionally
        //     specifying the name of a system semaphore object, and specifying a variable that
        //     receives a value indicating whether a new system semaphore was created.
        //
        // Parameters:
        //   initialCount:
        //     The initial number of requests for the semaphore that can be satisfied concurrently.
        //
        //   maximumCount:
        //     The maximum number of requests for the semaphore that can be satisfied concurrently.
        //
        //   name:
        //     The name of a named system semaphore object.
        //
        //   createdNew:
        //     When this method returns, contains true if a local semaphore was created (that
        //     is, if name is null or an empty string) or if the specified named system semaphore
        //     was created; false if the specified named system semaphore already existed. This
        //     parameter is passed uninitialized.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     initialCount is greater than maximumCount. -or- name is longer than 260 characters.
        //
        //   T:System.ArgumentOutOfRangeException:
        //     maximumCount is less than 1. -or- initialCount is less than 0.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred.
        //
        //   T:System.UnauthorizedAccessException:
        //     The named semaphore exists and has access control security, and the user does
        //     not have System.Security.AccessControl.SemaphoreRights.FullControl.
        //
        //   T:System.Threading.WaitHandleCannotBeOpenedException:
        //     The named semaphore cannot be created, perhaps because a wait handle of a different
        //     type has the same name.
        public Semaphore(int initialCount, int maximumCount, string name, out bool createdNew);

        //
        // Summary:
        //     Opens the specified named semaphore, if it already exists.
        //
        // Parameters:
        //   name:
        //     The name of the system semaphore to open.
        //
        // Returns:
        //     An object that represents the named system semaphore.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     name is an empty string. -or- name is longer than 260 characters.
        //
        //   T:System.ArgumentNullException:
        //     name is null.
        //
        //   T:System.Threading.WaitHandleCannotBeOpenedException:
        //     The named semaphore does not exist.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred.
        //
        //   T:System.UnauthorizedAccessException:
        //     The named semaphore exists, but the user does not have the security access required
        //     to use it.
        public static Semaphore OpenExisting(string name);
        //
        // Summary:
        //     Opens the specified named semaphore, if it already exists, and returns a value
        //     that indicates whether the operation succeeded.
        //
        // Parameters:
        //   name:
        //     The name of the system semaphore to open.
        //
        //   result:
        //     When this method returns, contains a System.Threading.Semaphore object that represents
        //     the named semaphore if the call succeeded, or null if the call failed. This parameter
        //     is treated as uninitialized.
        //
        // Returns:
        //     true if the named semaphore was opened successfully; otherwise, false.
        //
        // Exceptions:
        //   T:System.ArgumentException:
        //     name is an empty string. -or- name is longer than 260 characters.
        //
        //   T:System.ArgumentNullException:
        //     name is null.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred.
        //
        //   T:System.UnauthorizedAccessException:
        //     The named semaphore exists, but the user does not have the security access required
        //     to use it.
        public static bool TryOpenExisting(string name, out Semaphore result);
        //
        // Summary:
        //     Exits the semaphore and returns the previous count.
        //
        // Returns:
        //     The count on the semaphore before the System.Threading.Semaphore.Release* method
        //     was called.
        //
        // Exceptions:
        //   T:System.Threading.SemaphoreFullException:
        //     The semaphore count is already at the maximum value.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred with a named semaphore.
        //
        //   T:System.UnauthorizedAccessException:
        //     The current semaphore represents a named system semaphore, but the user does
        //     not have System.Security.AccessControl.SemaphoreRights.Modify. -or- The current
        //     semaphore represents a named system semaphore, but it was not opened with System.Security.AccessControl.SemaphoreRights.Modify.
        public int Release();
        //
        // Summary:
        //     Exits the semaphore a specified number of times and returns the previous count.
        //
        // Parameters:
        //   releaseCount:
        //     The number of times to exit the semaphore.
        //
        // Returns:
        //     The count on the semaphore before the System.Threading.Semaphore.Release* method
        //     was called.
        //
        // Exceptions:
        //   T:System.ArgumentOutOfRangeException:
        //     releaseCount is less than 1.
        //
        //   T:System.Threading.SemaphoreFullException:
        //     The semaphore count is already at the maximum value.
        //
        //   T:System.IO.IOException:
        //     A Win32 error occurred with a named semaphore.
        //
        //   T:System.UnauthorizedAccessException:
        //     The current semaphore represents a named system semaphore, but the user does
        //     not have System.Security.AccessControl.SemaphoreRights.Modify rights. -or- The
        //     current semaphore represents a named system semaphore, but it was not opened
        //     with System.Security.AccessControl.SemaphoreRights.Modify rights.
        public int Release(int releaseCount);
    }
}

System.Threading.Semaphore

看它的summary，我们大致相识这个类就是特地用来做并发限定的，它具有三个组织函数，我们最体贴的，就是个中两个参数int initialCount, int maximumCount及其涵义；

initialCount：能够被Semaphore 授与的数目标初始值；

maximumCount：能够被Semaphore 授与的最大值；

字面意义大概不太好明白，我们来把官宣翻译成普通话：）

举个栗子，我们把Semaphore看成是一个用来装钥匙的盒子，每个想要进入行列这道“门”的使命，都须要先从盒子里取一把钥匙，才进入；initialCount，就是说，这个盒子，一开始的时刻，放几把钥匙，然则进入行列的使命，常常不愿出来，不送还钥匙，无钥匙可用，这时候管理员就决议再多配一些钥匙，以备用，因而，一些新钥匙又被放入盒子里，但盒子的容积有限，一共能包容若干把钥匙，就是maximumCount了。

固然，我们罕见的状况是组织盒子的时刻，initialCount == maximumCount，特别场景下，会设置不相同，这个视详细营业而定。但是，maximumCount不能小于initialCount，initialCount不能小于0，这个是硬性的。

如许是不是是initialCount 和 maximumCount就很轻易明白了。

同时，Semaphore 另有非常重要的要领（Release）要领，再把上面的栗子举起来讲话，Release就是送还钥匙，使命完毕了，那末就出门还钥匙，然后其它在门口守候的使命就能够领到钥匙进门了：）

再者，Semaphore 继续自System.Threading.WaitHandle，因而乎，它就具有了一系列Wait要领，当有新使命来领钥匙，一看，盒子空了，那怎么办呢，等吧，然则等多久呢，是一向等下去照样等一个超常常候，这就看营业逻辑了。

在我的SemaphoreConcurrentPolicy完成里，会供应一个超常常候，爬虫蚂蚁小队长会推断，假如没拿到钥匙，就会再次返来尝试取钥匙。

OK，接下来，就是对我们的蚂蚁小队长举行革新了：

 1 namespace MikeWare.Core.Components.CrawlerFramework
 2 {
 3     using MikeWare.Core.Components.CrawlerFramework.Policies;
 4     using System;
 5     using System.Collections.Concurrent;
 6     using System.Threading;
 7     using System.Threading.Tasks;
 8 
 9     public class LeaderAnt : Ant
10     {
11         private ConcurrentQueue<JobContext> Queue;
12         private ManualResetEvent mre = new ManualResetEvent(false);
13         public AConcurrentPolicy EnqueuePolicy { get; set; }
14 
15         ……
16 
17         public void Enqueue(JobContext context)
18         {
19             if (null != EnqueuePolicy)
20             {
21                 while (!EnqueuePolicy.WaitOne(TimeSpan.FromMilliseconds(3)) && !mre.WaitOne(1))
22                     continue;
23             }
24 
25             Queue.Enqueue(context);
26         }
27 
28         ……
29 }

领队 —— LeaderAnt

重如果在入队的时刻，增加了拿钥匙的环节；

 1 namespace MikeWare.Crawlers.EBooks.Bizs
 2 {
 3     using MikeWare.Core.Components.CrawlerFramework;
 4     using MikeWare.Core.Components.CrawlerFramework.Policies;
 5     using MikeWare.Crawlers.EBooks.Entities;
 6     using System;
 7     using System.Collections.Generic;
 8     using System.Net;
 9 
10     public class EBooksCrawler
11     {
12         public static void Start(int pageIndex, DateTime lastUpdateTime)
13         {
14             var leader = new LeaderAnt()
15             {
16                 EnqueuePolicy = new SemaphoreConcurrentPolicy(100, 100)
17                 //EnqueuePolicy = new PeriodEnqueuePolicy(TimeSpan.FromMilliseconds(150))
18             };
19 
20             var newContext = new JobContext
21             {
22                 JobName = $"奇书网-最新电子书-列表-第{pageIndex.ToString("00000")}页",
23                 Uri = $"http://www.xqishuta.com/s/new/index_{pageIndex}.html",
24                 Method = WebRequestMethods.Http.Get,
25                 InParams = new Dictionary<string, object>(),
26                 Analizer = new BooksListAnalizer(),
27             };
28             newContext.InParams.Add(Consts.PAGE_INDEX, 1);
29             newContext.InParams.Add(Consts.LAST_UPDATE_TIME, DateTime.MinValue);
30 
31             leader.Enqueue(newContext);
32 
33             leader.Work();
34         }
35     }
36 }

营业层 —— EBooksCrawler

重如果在组织LeaderAnt的时刻，为其指定了我们要运用的战略；

同时须要注重的是，这个SemaphoreConcurrentPolicy并发战略的完成，并没有划定入队的时候距离，而是掌握了最大的行列长度，所以，并发的频次大概高，大概低，这个战略能够用来制衡资本的运用状况。关于入队时候距离，能够运用PeriodConcurrentPolicy或本身完成战略来掌握；

另一个战略的完成，我们就不在这里细说了。有兴致的同砚能够看看源码。

好了，本节的内容就这么多吧，置信人人对并发战略的制订与完成，都有了各自的明白。

后续章节一样出色，敬请期待……

手把手带你阅读Mybatis源码（一）构造篇

《C# 爬虫破境之道》：第二境爬虫应用 — 第七节：并发控制与计谋

小白学 Python 数据剖析（6）：Pandas （五）基础操作（2）数据选择

一篇文章带你「重新认识」线程上下文切换怎么玩儿

参与评论

快速导航

《C# 爬虫 破境之道》：第二境 爬虫应用 — 第七节：并发控制与计谋

分享本文封面

相关文章

参与评论

登录

快速导航

《C# 爬虫破境之道》：第二境爬虫应用 — 第七节：并发控制与计谋