《C# 爬虫 破境之道》:第二境 爬虫应用 — 第七节:并发控制与策略

2019/04/10 10:10
阅读数 16

我们在第五节中提到一个问题,任务队列增长速度太快,与之对应的采集、分析、处理速度远远跟不上,造成内存快速增长,带宽占用过高,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或自己实现策略来控制;

另一个策略的实现,我们就不在这里细说了。有兴趣的同学可以看看源码。

 

好了,本节的内容就这么多吧,相信大家对并发策略的制定与实现,都有了各自的理解。

后续章节同样精彩,敬请期待……

 

 

喜欢本系列丛书的朋友,可以点击链接加入QQ交流群(994761602)【C# 破境之道】
方便各位在有疑问的时候可以及时给我个反馈。同时,也算是给各位志同道合的朋友提供一个交流的平台。
需要源码的童鞋,也可以在群文件中获取最新源代码。

原文出处:https://www.cnblogs.com/mikecheers/p/12340397.html

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部