如何构建一个分布式爬虫：实战篇

本篇文章将是『如何构建一个分布式爬虫』系列文章的最后一篇，拟**从实战角度**来介绍如何构建一个*稳健的分布式微博爬虫*。这里我*没敢谈高效*，抓过微博数据的同学应该都知道微博的反爬虫能力，也知道微博数据抓取的瓶颈在哪里。我在知乎上看过一些同学的说法，把微博的数据抓取难度简单化了，我只能说，那是你太naive，没深入了解和长期抓取而已。

本文将会以PC端微博进行讲解，因为移动端微博数据不如PC短全面，而且抓取和解析难度都会小一些。文章比较长，由于篇幅所限，文章并没有列出所有代码，只是讲了大致流程和思路。

要抓微博数据，第一步便是模拟登陆，因为很多信息（比如用户信息，用户主页微博数据翻页等各种翻页）都需要在登录状态下才能查看。关于模拟登陆进阶，我写过两篇文章，一篇是超详细的Python实现新浪微博模拟登陆(小白都能懂)，是从小白的角度写的；另外一篇是超详细的Python实现百度云盘模拟登陆(模拟登陆进阶)，是从有一定经验的熟手的角度写的。读了这两篇文章，并且根据我写的过程自己动手实现过的同学，应该对于模拟登陆PC端微博是没有太大难度的。那两篇文章没有讲如何处理验证码，这里我简单说一下，做爬虫的同学不要老想着用什么机器学习的方法去识别复杂验证码，真的难度非常大，这应该也不是一个爬虫工程师的工作重点，当然这只是我的个人建议。工程化的项目，我还是建议大家通过*打码平台*来解决验证码的问题。我在 分布式微博爬虫( ResolveWang/WeiboSpider )中就是直接调用打码平台的接口来做的大规模微博账号的模拟登陆，效果还不错，而且打码成本很低。

说完模拟登陆（具体请参见我写的那两篇文章，篇幅所限，我就不copy过来了），我们现在正式进入微博的数据抓取。这里我会以微博用户信息抓取为例来进行分析和讲解。

关于用户信息抓取，可能我们有两个目的。
一个是我们只想抓一些指定用户，另外一个是我们想尽可能多的抓取更多数量的用户的信息。我的目的假定是第二种。那么我们该以什么样的策略来抓取，才能
获得尽可能多的用户信息呢？如果我们初始用户选择有误，选了一些不活跃的用户，很可能会形成一个环，这样就抓不了太多的数据。这里有一个很简单的思路：我们
把一些大V拿来做为种子用户，我们先抓他们的个人信息，然后再抓大V所关注的用户和粉丝，大V关注的用户肯定也是类似大V的用户，这样的话，就不容易形成环了。

策略我们都清楚了。就该是分析和编码了。

我们先来分析如何构造用户信息的URL。这里我以微博名为

一起神吐槽

的博主为例进行分析。做爬虫的话，一个很重要的意识就是
爬虫能抓的数据都是人能看到的数据，反过来，人能在浏览器上看到的数据，爬虫几乎都能抓。这里用的是

几乎

，因为有的数据抓取难度特别。我们首先需要
以正常人的流程看看怎么获取到用户的信息。我们先进入该博主的主页，如下图

点击查看更多，可以查看到该博主的具体信息

这里我们就看到了他的具体信息了。然后，我们看该页面的url构造

Sina Visitor System

我直接copy的地址栏的url。这样做有啥不好的呢？对于老鸟来说，一下就看出来了，这样做的话，可能会导致信息不全，因为可能有些信息是动态加载的。所以，我们需要通过抓包来判断到底微博会通过该url返回所有信息，还是需要请求一些ajax 链接才会返回一些关键信息。这里我就重复一下我的观点：
抓包很重要，抓包很重要，抓包很重要！重要的事情说三遍。关于抓包，我在超详细的Python实现新浪微博模拟登陆(小白都能懂)和超详细的Python实现百度云盘模拟登陆(模拟登陆进阶)都详细讲过了，这里我就不讲了。

我们抓完包，发现并没有ajax请求。那么可以肯定请求前面的url，会返回所有信息。我们通过点击鼠标右键，查看网页源代码，然后

ctrl+a

、

ctrl+c

将所有的页面源码保存到本地，这里我命名为

personinfo.html

。我们用浏览器打开该文件，发现我们需要的所有信息都在这段源码中，这个工作和抓包判断数据是否全面有些重复，但是在我看来是必不可少的，因为我们解析页面数据的时候还可以用到这个html文件，如果我们每次都通过网络请求去解析内容的话，那么可能账号没一会儿就会被封了（因为频繁访问微博信息），所以我们需要
把要解析的文件保存到本地。

从上面分析中我们可以得知

http://weibo.com/p/1005051751195602/info?mod=pedit_more

这个url就是获取用户数据的url。那么我们在只知道用户id的时候怎么构造它呢？我们可以多拿几个用户id来做测试，看构造是否有规律，比如我这里以用户名为网易云音乐的用户做分析，发现它的用户信息页面构造如下

Sina Visitor System

这个就和上面那个不同了。但是我们仔细观察，可以发现上面那个是个人用户，下面是企业微博用户。我们尝试一下把它们url格式都统一为第一种或者第二种的格式
http://weibo.com/1751195602/about

这样会出现404，那么统一成上面那种呢？

http://weibo.com/p/1005051721030997/info?mod=pedit_more

这样子的话，它会被重定向到用户主页，而不是用户详细资料页。所以也就不对了。那么该以什么依据判断何时用第一种url格式，何时用第二种url格式呢？我们多翻几个用户，会发现除了

100505

之外，还有

100305

、

100206

等前缀，那么我猜想这个应该可以区分不同用户。这个前缀在哪里可以得到呢？我们打开我们刚保存的页面源码，搜索

100505

，可以发现

微博应该是根据这个来区分不同用户类型的。这里大家可以自己也可以试试，看不同用户的

domain

是否不同。为了数据能全面，我也是做了大量测试，发现个人用户的domain是

1005051

，作家是

100305

，其他基本都是认证的企业号。前两个个人信息的url构造就是

http://weibo.com/p/domain+uid/info?mod=pedit_more

后者的是

http://weibo.com/uid/about

弄清楚了个人信息url的构造方式，但是还有一个问题。我们已知只有uid啊，没有domain啊。如果是企业号，我们通过

domain=100505

会被重定向到主页，如果是作家等(domain=100305或者100306)，也会被重定向主页。我们在主页把domain提取出来，再请求一次，不就能拿到用户详细信息了吗？

关于如何构造获取用户信息的url的相关分析就到这里了。因为我们是在登录的情况下进行数据抓取的，可能在抓取的时候，某个账号突然就被封了，或者由于网络原因，某次请求失败了，该如何处理？对于前者，我们需要判断每次请求返回的内容是否符合预期，也就是看response url是否正常，看response content是否是404或者让你验证手机号等，对于后者，我们可以做一个简单的重试策略。处理这两种情况的代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
1@timeout_decorator

2def get_page(url, user_verify=True, need_login=True):

3    """

4    :param url: 待抓取url

5    :param user_verify: 是否为可能出现验证码的页面(ajax连接不会出现验证码，如果是请求微博或者用户信息可能出现验证码)，否为抓取转发的ajax连接

6    :param need_login: 抓取页面是否需要登录，这样做可以减小一些账号的压力

7    :return: 返回请求的数据，如果出现404或者403,或者是别的异常，都返回空字符串

8    """

9    crawler.info('本次抓取的url为{url}'.format(url=url))

10    count = 0

11

12    while count < max_retries:

13

14        if need_login:

15            # 每次重试的时候都换cookies,并且和上次不同,如果只有一个账号，那么就允许相同

16            name_cookies = Cookies.fetch_cookies()

17            

18            if name_cookies is None:

19                crawler.warning('cookie池中不存在cookie，正在检查是否有可用账号')

20                rs = get_login_info()

21

22                # 选择状态正常的账号进行登录，账号都不可用就停掉celery worker

23                if len(rs) == 0:

24                    crawler.error('账号均不可用，请检查账号健康状况')

25                    # 杀死所有关于celery的进程

26                    if 'win32' in sys.platform:

27                        os.popen('taskkill /F /IM "celery*"')

28                    else:

29                        os.popen('pkill -f "celery"')

30                else:

31                    crawler.info('重新获取cookie中...')

32                    login.excute_login_task()

33                    time.sleep(10)

34

35        try:

36            if need_login:

37                resp = requests.get(url, headers=headers, cookies=name_cookies[1], timeout=time_out, verify=False)

38

39                if "$CONFIG['islogin'] = '0'" in resp.text:

40                    crawler.warning('账号{}出现异常'.format(name_cookies[0]))

41                    freeze_account(name_cookies[0], 0)

42                    Cookies.delete_cookies(name_cookies[0])

43                    continue

44            else:

45                resp = requests.get(url, headers=headers, timeout=time_out, verify=False)

46

47            page = resp.text

48            if page:

49                page = page.encode('utf-8', 'ignore').decode('utf-8')

50            else:

51                continue

52

53            # 每次抓取过后程序sleep的时间，降低封号危险

54            time.sleep(interal)

55

56            if user_verify:

57                if 'unfreeze' in resp.url or 'accessdeny' in resp.url or 'userblock' in resp.url or is_403(page):

58                    crawler.warning('账号{}已经被冻结'.format(name_cookies[0]))

59                    freeze_account(name_cookies[0], 0)

60                    Cookies.delete_cookies(name_cookies[0])

61                    count += 1

62                    continue

63

64                if 'verifybmobile' in resp.url:

65                    crawler.warning('账号{}功能被锁定，需要手机解锁'.format(name_cookies[0]))

66        

67                    freeze_account(name_cookies[0], -1)

68                    Cookies.delete_cookies(name_cookies[0])

69                    continue

70

71                if not is_complete(page):

72                    count += 1

73                    continue

74

75                if is_404(page):

76                    crawler.warning('url为{url}的连接不存在'.format(url=url))

77                    return ''

78

79        except (requests.exceptions.ReadTimeout, requests.exceptions.ConnectionError, AttributeError) as e:

80            crawler.warning('抓取{}出现异常，具体信息是{}'.format(url, e))

81            count += 1

82            time.sleep(excp_interal)

83

84        else:

85            Urls.store_crawl_url(url, 1)

86            return page

87

88    crawler.warning('抓取{}已达到最大重试次数，请在redis的失败队列中查看该url并检查原因'.format(url))

89    Urls.store_crawl_url(url, 0)

90    return ''

91

92

这里大家把上述代码当一段伪代码读就行了，主要看看如何处理抓取时候的异常。因为如果贴整个用户抓取的代码，不是很现实，代码量有点大。

下面讲页面解析的分析。有一些做PC端微博信息抓取的同学，可能曾经遇到过这么个问题：保存到本地的html文件打开都能看到所有信息啊，为啥在页面源码中找不到呢？因为PC端微博页面的关键信息都是像下图这样，被

FM.view()

包裹起来的，里面的数据可能被*json encode*过。

那么这么多的

FM.view()

，我们怎么知道该提取哪个呢？这里有一个小技巧，
由于只有中文会被编码，英文还是原来的样子，所以我们可以看哪段script中包含了渲染后的页面中的字符，那么那段应该就可能包含所有页面信息。我们这里以顶部的头像为例，如图

我们在页面源码中搜索，只发现一个script中有该字符串，那么就是那段script是页面相关信息。我们可以通过正则表达式把该script提取出来，然后把其中的

html

也提取出来，再保存到本地，看看信息是否全面。这里我就不截图了。感觉还有很多要写的，不然篇幅太长了。

另外，对于具体页面的解析，我也不做太多的介绍了。太细的东西还是建议读读源码。我只讲一下，我觉得的一种处理异常的比较优雅的方式。微博爬虫的话，主要是页面样式太多，如果你打算包含所有不同的用户的模版，那么我觉得几乎不可能，不同用户模版，用到的解析规则就不一样。那么出现解析异常如何处理？尤其是你没有catch到的异常。很可能因为这个问题，程序就崩掉。其实对于Python这门语言来说，我们可以通过 **装饰器** 来捕捉我们没有考虑到的异常，比如我这个装饰器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
1def parse_decorator(return_type):

2    """

3    :param return_type: 用于捕捉页面解析的异常, 0表示返回数字0, 1表示返回空字符串, 2表示返回[],3表示返回False, 4表示返回{}, 5返回None

4    :return: 0,'',[],False,{},None

5    """

6    def page_parse(func):

7        @wraps(func)

8        def handle_error(*keys):

9            try:

10                return func(*keys)

11            except Exception as e:

12                parser.error(e)

13

14                if return_type == 5:

15                    return None

16                elif return_type == 4:

17                    return {}

18                elif return_type == 3:

19                    return False

20                elif return_type == 2:

21                    return []

22                elif return_type == 1:

23                    return ''

24                else:

25                    return 0

26        return handle_error

27    return page_parse

28

29

上面的代码就是处理解析页面发生异常的情况，我们只能在数据的准确性、全面性和程序的健壮性之间做一些取舍。用装饰器的话，程序中不用写太多的

try

语句，代码重复率也会减少很多。

页面的解析由于篇幅所限，我就讲到这里了。没有涉及太具体的解析，其中一个还有一个比较难的点，就是
数据的全面性，读者可以去多观察几个微博用户的个人信息，就会发现有的个人信息，有的用户有填写，有的并没有。解析的时候要考虑完的话，建议
从自己的微博的个人信息入手，看到底有哪些可以填。这样可以保证几乎不会漏掉一些重要的信息。

最后，我再切合本文的标题，讲如何搭建一个分布式的微博爬虫。
开发过程中，我们可以先就做单机单线程的爬虫，然后再改成使用celery的方式。这里这样做是为了方便开发和测试，因为你单机搭起来并且跑得通了，那么分布式的话，就很容易改了，因为celery的API使用本来就很简洁。

我们抓取的是用户信息和他的关注和粉丝uid。用户信息的话，我们一个请求大概能抓取一个用户的信息，而粉丝和关注我们一个请求可以抓取18个左右（因为这个抓的是列表），显然可以发现**用户信息应该多占一些请求的资源**。这时候就该介绍如何构建一个分布式爬虫：理论篇没有介绍的关于celery的一个高级特性了，它叫做
任务路由。
直白点说，它可以规定哪个分布式节点能做哪些任务，不能做哪些任务。它的存在可以让资源分配更加合理，分布式微博爬虫(ResolveWang/WeiboSpider)项目初期，就没有使用任务路由，然后抓了十多万条关注和粉丝，发现用户信息只有几万条，这就是资源分配得不合理。那么如何进行任务路由呢

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
1# coding:utf-8

2import os

3from datetime import timedelta

4from celery import Celery

5from kombu import Exchange, Queue

6from config.conf import get_broker_or_backend

7from celery import platforms

8

9# 允许celery以root身份启动

10platforms.C_FORCE_ROOT = True

11

12worker_log_path = os.path.join(os.path.dirname(os.path.dirname(__file__))+'/logs', 'celery.log')

13beat_log_path = os.path.join(os.path.dirname(os.path.dirname(__file__))+'/logs', 'beat.log')

14

15tasks = ['tasks.login', 'tasks.user']

16

17# include的作用就是注册服务化函数

18app = Celery('weibo_task', include=tasks, broker=get_broker_or_backend(1), backend=get_broker_or_backend(2))

19

20app.conf.update(

21    CELERY_TIMEZONE='Asia/Shanghai',

22    CELERY_ENABLE_UTC=True,

23    CELERYD_LOG_FILE=worker_log_path,

24    CELERYBEAT_LOG_FILE=beat_log_path,

25    CELERY_ACCEPT_CONTENT=['json'],

26    CELERY_TASK_SERIALIZER='json',

27    CELERY_RESULT_SERIALIZER='json',

28    CELERY_QUEUES=(

29        Queue('login_queue', exchange=Exchange('login', type='direct'), routing_key='for_login'),

30        Queue('user_crawler', exchange=Exchange('user_info', type='direct'), routing_key='for_user_info'),

31        Queue('fans_followers', exchange=Exchange('fans_followers', type='direct'), routing_key='for_fans_followers'),

32)

33

34

上述代码我指定了有

login_queue

、

user_crawler

、

fans_followers

三个任务队列。它们分别的作用是登录、用户信息抓取、粉丝和关注抓取。现在假设我有三台爬虫服务器A、B和C。我想让我所有的账号登录任务分散到三台服务器、让用户抓取在A和B上执行，让粉丝和关注抓取在C上执行，那么启动A、B、C三个服务器的celery worker的命令就分别是

celery -A tasks.workers -Q login_queue,user_crawler worker -l info -c 1 # A服务器和B服务器启动worker的命令，它们只会执行登录和用户信息抓取任务
celery -A tasks.workers -Q login_queue,fans_followers worker -l info -c 1 # C服务器启动worker的命令，它只会执行登录、粉丝和关注抓取任务

然后我们通过命令行或者代码（如下）就能发送所有任务给各个节点执行了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
1# coding:utf-8

2from tasks.workers import app

3from page_get import user as user_get

4from db.seed_ids import get_seed_ids, get_seed_by_id, insert_seeds, set_seed_other_crawled

5

6@app.task(ignore_result=True)

7def crawl_follower_fans(uid):

8    seed = get_seed_by_id(uid)

9    if seed.other_crawled == 0:

10        rs = user_get.get_fans_or_followers_ids(uid, 1)

11        rs.extend(user_get.get_fans_or_followers_ids(uid, 2))

12        datas = set(rs)

13        # 重复数据跳过插入

14        if datas:

15            insert_seeds(datas)

16        set_seed_other_crawled(uid)

17@app.task(ignore_result=True)

18def crawl_person_infos(uid):

19    """

20    根据用户id来爬取用户相关资料和用户的关注数和粉丝数（由于微博服务端限制，默认爬取前五页，企业号的关注和粉丝也不能查看）

21    :param uid: 用户id

22    :return: 

23    """

24    if not uid:

25        return

26

27    # 由于与别的任务共享数据表，所以需要先判断数据库是否有该用户信息，再进行抓取

28    user = user_get.get_profile(uid)

29    # 不抓取企业号

30    if user.verify_type == 2:

31        set_seed_other_crawled(uid)

32        return

33    app.send_task('tasks.user.crawl_follower_fans', args=(uid,), queue='fans_followers',

34                  routing_key='for_fans_followers')

35

36

37@app.task(ignore_result=True)

38def excute_user_task():

39    seeds = get_seed_ids()

40    if seeds:

41        for seed in seeds:

42            # 在send_task的时候指定任务队列

43            app.send_task('tasks.user.crawl_person_infos', args=(seed.uid,), queue='user_crawler',

44                          routing_key='for_user_info')

45

46

这里我们是通过

queue='user_crawler',routing_key='for_user_info'

来将任务和worker进行关联的。

关于celery任务路由的更详细的资料请阅读官方文档(Routing Tasks – Celery 4.0.2 documentation)。

到这里，基本把微博信息抓取的过程和分布式进行抓取的过程都讲完了，具体实现分布式的方法，可以读读基础篇(如何构建一个分布式爬虫：基础篇)。由于代码量比较大，我并没有贴上完整的代码，只讲了要点。分析过程是讲的抓取过程的分析和页面解析的分析，并在最后，结合分布式，讲了一下使用任务队列来让分布式爬虫更加灵活和可扩展。

如果有同学想跟着做一遍，可能需要参考分布式微博爬虫( ResolveWang/WeiboSpider )的源码，自己动手实现一下，或者跑一下，印象可能会更加深刻。