我们了解了 Scrapy 爬虫框架的用法。这些框架都是在同一台主机上运行的，爬取效率比较有限。如果多台主机协同爬取，那么爬取效率必然会成倍增长，这就是分布式爬虫的优势。

本章我们就来了解一下分布式爬虫的基本原理，以及 Scrapy 实现分布式爬虫的流程。

一、分布式爬虫原理

我们在前面已经实现了 Scrapy 微博爬虫，虽然爬虫是异步加多线程的，但是我们只能在一台主机上运行，所以爬取效率还是有限的，分布式爬虫则是将多台主机组合起来，共同完成一个爬取任务，这将大大提高爬取的效率。

1.1 分布式爬虫架构

Scrapy 单机爬虫中有一个本地爬取队列 Queue，这个队列是利用 deque 模块实现的。如果新的 Request 生成就会放到队列里面，随后 Request 被 Scheduler 调度。之后，Request 交给 Downloader 执行爬取，简单的调度架构如图所示。

如果两个 Scheduler 同时从队列里面取 Request，每个 Scheduler 都有其对应的 Downloader，那么在带宽足够、正常爬取且不考虑队列存取压力的情况下，爬取效率会有什么变化？没错，爬取效率会翻倍。

这样，Scheduler 可以扩展多个，Downloader 也可以扩展多个。而爬取队列 Queue 必须始终为一个，也就是所谓的共享爬取队列。这样才能保证 Scheduer 从队列里调度某个 Request 之后，其他 Scheduler 不会重复调度此 Request，就可以做到多个 Schduler 同步爬取。这就是分布式爬虫的基本雏形，简单调度架构如图所示。

我们需要做的就是在多台主机上同时运行爬虫任务协同爬取，而协同爬取的前提就是共享爬取队列。这样各台主机就不需要各自维护爬取队列，而是从共享爬取队列存取 Request。但是各台主机还是有各自的 Scheduler 和 Downloader，所以调度和下载功能分别完成。如果不考虑队列存取性能消耗，爬取效率还是会成倍提高。

1.2 维护爬取队列

那么这个队列用什么维护来好呢？我们首先需要考虑的就是性能问题，什么数据库存取效率高？我们自然想到基于内存存储的 Redis，而且 Redis 还支持多种数据结构，例如列表 List、集合 Set、有序集合 Sorted Set 等等，存取的操作也非常简单，所以在这里我们采用 Redis 来维护爬取队列。

这几种数据结构存储实际各有千秋，分析如下：

• 列表数据结构有 lpush()、lpop()、rpush()、rpop() 方法，所以我们可以用它来实现一个先进先出式爬取队列，也可以实现一个先进后出栈式爬取队列。
• 集合的元素是无序的且不重复的，这样我们可以非常方便地实现一个随机排序的不重复的爬取队列。
• 有序集合带有分数表示，而 Scrapy 的 Request 也有优先级的控制，所以用有集合我们可以实现一个带优先级调度的队列。

这些不同的队列我们需要根据具体爬虫的需求灵活选择。

1.3 怎样来去重

Scrapy 有自动去重，它的去重使用了 Python 中的集合。这个集合记录了 Scrapy 中每个 Request 的指纹，这个指纹实际上就是 Request 的散列值。我们可以看看 Scrapy 的源代码，如下所示：

import hashlib
def request_fingerprint(request, include_headers=None):
    if include_headers:
        include_headers = tuple(to_bytes(h.lower())
                                 for h in sorted(include_headers))
    cache = _fingerprint_cache.setdefault(request, {})
    if include_headers not in cache:
        fp = hashlib.sha1()
        fp.update(to_bytes(request.method))
        fp.update(to_bytes(canonicalize_url(request.url)))
        fp.update(request.body or b'')
        if include_headers:
            for hdr in include_headers:
                if hdr in request.headers:
                    fp.update(hdr)
                    for v in request.headers.getlist(hdr):
                        fp.update(v)
        cache[include_headers] = fp.hexdigest()
    return cache[include_headers]

request_fingerprint() 就是计算 Request 指纹的方法，其方法内部使用的是 hashlib 的 sha1() 方法。计算的字段包括 Request 的 Method、URL、Body、Headers 这几部分内容，这里只要有一点不同，那么计算的结果就不同。计算得到的结果是加密后的字符串，也就是指纹。每个 Request 都有独有的指纹，指纹就是一个字符串，判定字符串是否重复比判定 Request 对象是否重复容易得多，所以指纹可以作为判定 Request 是否重复的依据。

那么我们如何判定重复呢？Scrapy 是这样实现的，如下所示：

def __init__(self):
    self.fingerprints = set()

def request_seen(self, request):
    fp = self.request_fingerprint(request)
    if fp in self.fingerprints:
        return True
    self.fingerprints.add(fp)

在去重的类 RFPDupeFilter 中，有一个 request_seen() 方法，这个方法有一个参数 request，它的作用就是检测该 Request 对象是否重复。这个方法调用 request_fingerprint() 获取该 Request 的指纹，检测这个指纹是否存在于 fingerprints 变量中，而 fingerprints 是一个集合，集合的元素都是不重复的。如果指纹存在，那么就返回 True，说明该 Request 是重复的，否则这个指纹加入到集合中。如果下次还有相同的 Request 传递过来，指纹也是相同的，那么这时指纹就已经存在于集合中，Request 对象就会直接判定为重复。这样去重的目的就实现了。

Scrapy 的去重过程就是，利用集合元素的不重复特性来实现 Request 的去重。

对于分布式爬虫来说，我们肯定不能再用每个爬虫各自的集合来去重了。因为这样还是每个主机单独维护自己的集合，不能做到共享。多台主机如果生成了相同的 Request，只能各自去重，各个主机之间就无法做到去重了。

那么要实现去重，这个指纹集合也需要是共享的，Redis 正好有集合的存储数据结构，我们可以利用 Redis 的集合作为指纹集合，那么这样去重集合也是利用 Redis 共享的。每台主机新生成 Request 之后，把该 Request 的指纹与集合比对，如果指纹已经存在，说明该 Request 是重复的，否则将 Request 的指纹加入到这个集合中即可。利用同样的原理不同的存储结构我们也实现了分布式 Reqeust 的去重。

1.4 防止中断

在 Scrapy 中，爬虫运行时的 Request 队列放在内存中。爬虫运行中断后，这个队列的空间就被释放，此队列就被销毁了。所以一旦爬虫运行中断，爬虫再次运行就相当于全新的爬取过程。

要做到中断后继续爬取，我们可以将队列中的 Request 保存起来，下次爬取直接读取保存数据即可获取上次爬取的队列。我们在 Scrapy 中指定一个爬取队列的存储路径即可，这个路径使用 JOB_DIR 变量来标识，我们可以用如下命令来实现：

scrapy crawl spider -s JOBDIR=crawls/spider

更加详细的使用方法可以参见官方文档，链接为：https://doc.scrapy.org/en/latest/topics/jobs.html。

在 Scrapy 中，我们实际是把爬取队列保存到本地，第二次爬取直接读取并恢复队列即可。那么在分布式架构中我们还用担心这个问题吗？不需要。因为爬取队列本身就是用数据库保存的，如果爬虫中断了，数据库中的 Request 依然是存在的，下次启动就会接着上次中断的地方继续爬取。

所以，当 Redis 的队列为空时，爬虫会重新爬取；当 Redis 的队列不为空时，爬虫便会接着上次中断之处继续爬取。

1.5 架构实现

我们接下来就需要在程序中实现这个架构了。首先实现一个共享的爬取队列，还要实现去重的功能。另外，重写一个 Scheduer 的实现，使之可以从共享的爬取队列存取 Request。

幸运的是，已经有人实现了这些逻辑和架构，并发布成叫 Scrapy-Redis 的 Python 包。接下来，我们看看 Scrapy-Redis 的源码实现，以及它的详细工作原理。

二、Scrapy-Redis 源码解析

Scrapy-Redis 库已经为我们提供了 Scrapy 分布式的队列、调度器、去重等功能，其 GitHub 地址为：https://github.com/rmax/scrapy-redis。

本节我们深入了解一下，利用 Redis 如何实现 Scrapy 分布式。

2.1 获取源码

可以把源码克隆下来，执行如下命令：

git clone https://github.com/rmax/scrapy-redis.git

核心源码在 scrapy-redis/src/scrapy_redis 目录下。

2.2 爬取队列

从爬取队列入手，看看它的具体实现。源码文件为 queue.py，它有三个队列的实现，首先它实现了一个父类 Base，提供一些基本方法和属性，如下所示：

class Base(object):
    """Per-spider base queue class"""
    def __init__(self, server, spider, key, serializer=None):
        if serializer is None:
            serializer = picklecompat
        if not hasattr(serializer, 'loads'):
            raise TypeError("serializer does not implement 'loads' function: % r"
                            % serializer)
        if not hasattr(serializer, 'dumps'):
            raise TypeError("serializer '% s' does not implement 'dumps' function: % r"
                            % serializer)
        self.server = server
        self.spider = spider
        self.key = key % {'spider': spider.name}
        self.serializer = serializer

    def _encode_request(self, request):
        obj = request_to_dict(request, self.spider)
        return self.serializer.dumps(obj)

    def _decode_request(self, encoded_request):
        obj = self.serializer.loads(encoded_request)
        return request_from_dict(obj, self.spider)

    def __len__(self):
        """Return the length of the queue"""
        raise NotImplementedError

    def push(self, request):
        """Push a request"""
        raise NotImplementedError

    def pop(self, timeout=0):
        """Pop a request"""
        raise NotImplementedError

    def clear(self):
        """Clear queue/stack"""
        self.server.delete(self.key)

首先看一下 _encode_request() 和 _decode_request() 方法，因为我们需要把一 个 Request 对象存储到数据库中，但数据库无法直接存储对象，所以需要将 Request 序列化转成字符串再存储，而这两个方法就分别是序列化和反序列化的操作，利用 pickle 库来实现，一般在调用 push() 将 Request 存入数据库时会调用 _encode_request() 方法进行序列化，在调用 pop() 取出 Request 的时候会调用 _decode_request() 进行反序列化。

在父类中 __len__()、push() 和 pop() 方法都是未实现的，会直接抛出 NotImplementedError，因此这个类是不能直接被使用的，所以必须要实现一个子类来重写这三个方法，而不同的子类就会有不同的实现，也就有着不同的功能。

那么接下来就需要定义一些子类来继承 Base 类，并重写这几个方法，那在源码中就有三个子类的实现，它们分别是 FifoQueue、PriorityQueue、LifoQueue，我们分别来看下它们的实现原理。

首先是 FifoQueue：

class FifoQueue(Base):
    """Per-spider FIFO queue"""

    def __len__(self):
        """Return the length of the queue"""
        return self.server.llen(self.key)

    def push(self, request):
        """Push a request"""
        self.server.lpush(self.key, self._encode_request(request))

    def pop(self, timeout=0):
        """Pop a request"""
        if timeout > 0:
            data = self.server.brpop(self.key, timeout)
            if isinstance(data, tuple):
                data = data[1]
        else:
            data = self.server.rpop(self.key)
        if data:
            return self._decode_request(data)

可以看到这个类继承了 Base 类，并重写了 __len__()、push()、pop() 这三个方法，在这三个方法中都是对 server 对象的操作，而 server 对象就是一个 Redis 连接对象，我们可以直接调用其操作 Redis 的方法对数据库进行操作，可以看到这里的操作方法有 llen()、lpush()、rpop() 等，那这就代表此爬取队列是使用的 Redis 的列表，序列化后的 Request 会被存入列表中，就是列表的其中一个元素，__len__() 方法是获取列表的长度，push() 方法中调用了 lpush() 操作，这代表从列表左侧存入数据，pop() 方法中调用了 rpop() 操作，这代表从列表右侧取出数据。

所以 Request 在列表中的存取顺序是左侧进、右侧出，所以这是有序的进出，即先进先出，英文叫做 First Input First Output，也被简称作 Fifo，而此类的名称就叫做 FifoQueue。

另外还有一个与之相反的实现类，叫做 LifoQueue，实现如下：

class LifoQueue(Base):
    """Per-spider LIFO queue."""

    def __len__(self):
        """Return the length of the stack"""
        return self.server.llen(self.key)

    def push(self, request):
        """Push a request"""
        self.server.lpush(self.key, self._encode_request(request))

    def pop(self, timeout=0):
        """Pop a request"""
        if timeout > 0:
            data = self.server.blpop(self.key, timeout)
            if isinstance(data, tuple):
                data = data[1]
        else:
            data = self.server.lpop(self.key)

        if data:
            return self._decode_request(data)

与 FifoQueue 不同的就是它的 pop() 方法，在这里使用的是 lpop() 操作，也就是从左侧出，而 push() 方法依然是使用的 lpush() 操作，是从左侧入。那么这样达到的效果就是先进后出、后进先出，英文叫做 Last In First Out，简称为 Lifo，而此类名称就叫做 LifoQueue。同时这个存取方式类似栈的操作，所以其实也可以称作 StackQueue。

另外在源码中还有一个子类实现，叫做 PriorityQueue，顾名思义，它叫做优先级队列，实现如下：

class PriorityQueue(Base):
    """Per-spider priority queue abstraction using redis' sorted set"""

    def __len__(self):
        """Return the length of the queue"""
        return self.server.zcard(self.key)

    def push(self, request):
        """Push a request"""
        data = self._encode_request(request)
        score = -request.priority
        self.server.execute_command('ZADD', self.key, score, data)

    def pop(self, timeout=0):
        """
        Pop a request
        timeout not support in this queue class
        """
        pipe = self.server.pipeline()
        pipe.multi()
        pipe.zrange(self.key, 0, 0).zremrangebyrank(self.key, 0, 0)
        results, count = pipe.execute()
        if results:
            return self._decode_request(results[0])

在这里我们可以看到 __len__()、push()、pop() 方法中使用了 server 对象的 zcard()、zadd()、zrange() 操作，可以知道这里使用的存储结果是有序集合 Sorted Set，在这个集合中每个元素都可以设置一个分数，那么这个分数就代表优先级。

在 __len__() 方法里调用了 zcard() 操作，返回的就是有序集合的大小，也就是爬取队列的长度，在 push() 方法中调用了 zadd() 操作，就是向集合中添加元素，这里的分数指定成 Request 的优先级的相反数，因为分数低的会排在集合的前面，所以这里高优先级的 Request 就会存在集合的最前面。pop() 方法是首先调用了 zrange() 操作取出了集合的第一个元素，因为最高优先级的 Request 会存在集合最前面，所以第一个元素就是最高优先级的 Request，然后再调用 zremrangebyrank() 操作将这个元素删除，这样就完成了取出并删除的操作。

此队列是默认使用的队列，也就是爬取队列默认是使用有序集合来存储的。

2.3 去重过滤

前面说过 Scrapy 的去重是利用集合来实现的，而在 Scrapy 分布式中的去重就需要利用共享的集合，那么这里使用的就是 Redis 中的集合数据结构。我们来看看去重类是怎样实现的，源码文件是 dupefilter.py，其内实现了一个 RFPDupeFilter 类，如下所示：

class RFPDupeFilter(BaseDupeFilter):
    """Redis-based request duplicates filter.
    This class can also be used with default Scrapy's scheduler.
    """
    logger = logger
    def __init__(self, server, key, debug=False):
        """Initialize the duplicates filter.
        Parameters
        ----------
        server : redis.StrictRedis
            The redis server instance.
        key : str
            Redis key Where to store fingerprints.
        debug : bool, optional
            Whether to log filtered requests.
        """
        self.server = server
        self.key = key
        self.debug = debug
        self.logdupes = True

    @classmethod
    def from_settings(cls, settings):
        """Returns an instance from given settings.
        This uses by default the key ``dupefilter:<timestamp>``. When using the
        ``scrapy_redis.scheduler.Scheduler`` class, this method is not used as
        it needs to pass the spider name in the key.
        Parameters
        ----------
        settings : scrapy.settings.Settings
        Returns
        -------
        RFPDupeFilter
            A RFPDupeFilter instance.
        """
        server = get_redis_from_settings(settings)
        key = defaults.DUPEFILTER_KEY % {'timestamp': int(time.time())}
        debug = settings.getbool('DUPEFILTER_DEBUG')
        return cls(server, key=key, debug=debug)

    @classmethod
    def from_crawler(cls, crawler):
        """Returns instance from crawler.
        Parameters
        ----------
        crawler : scrapy.crawler.Crawler
        Returns
        -------
        RFPDupeFilter
            Instance of RFPDupeFilter.
        """
        return cls.from_settings(crawler.settings)

    def request_seen(self, request):
        """Returns True if request was already seen.
        Parameters
        ----------
        request : scrapy.http.Request
        Returns
        -------
        bool
        """
        fp = self.request_fingerprint(request)
        added = self.server.sadd(self.key, fp)
        return added == 0

    def request_fingerprint(self, request):
        """Returns a fingerprint for a given request.
        Parameters
        ----------
        request : scrapy.http.Request

        Returns
        -------
        str

        """
        return request_fingerprint(request)

    def close(self, reason=''):
        """Delete data on close. Called by Scrapy's scheduler.
        Parameters
        ----------
        reason : str, optional
        """
        self.clear()

    def clear(self):
        """Clears fingerprints data."""
        self.server.delete(self.key)

    def log(self, request, spider):
        """Logs given request.
        Parameters
        ----------
        request : scrapy.http.Request
        spider : scrapy.spiders.Spider
        """
        if self.debug:
            msg = "Filtered duplicate request: %(request) s"
            self.logger.debug(msg, {'request': request}, extra={'spider': spider})
        elif self.logdupes:
            msg = ("Filtered duplicate request %(request) s"
                   "- no more duplicates will be shown"
                   "(see DUPEFILTER_DEBUG to show all duplicates)")
            self.logger.debug(msg, {'request': request}, extra={'spider': spider})
            self.logdupes = False

这里同样实现了一个 request_seen() 方法，和 Scrapy 中的 request_seen() 方法实现极其类似。不过这里集合使用的是 server 对象的 sadd() 操作，也就是集合不再是一个简单数据结构了，而是直接换成了数据库的存储方式。

鉴别重复的方式还是使用指纹，指纹同样是依靠 request_fingerprint() 方法来获取的。获取指纹之后就直接向集合添加指纹，如果添加成功，说明这个指纹原本不存在于集合中，返回值 1。代码中最后的返回结果是判定添加结果是否为 0，如果刚才的返回值为 1，那这个判定结果就是 False，也就是不重复，否则判定为重复。

这样我们就成功利用 Redis 的集合完成了指纹的记录和重复的验证。

2.4 调度器

Scrapy-Redis 还帮我们实现了配合 Queue、DupeFilter 使用的调度器 Scheduler，源文件名称是 scheduler.py。我们可以指定一些配置，如 SCHEDULER_FLUSH_ON_START 即是否在爬取开始的时候清空爬取队列，SCHEDULER_PERSIST 即是否在爬取结束后保持爬取队列不清除。我们可以在 settings.py 里自由配置，而此调度器很好地实现了对接。

接下来我们看看两个核心的存取方法，实现如下所示：

def enqueue_request(self, request):
    if not request.dont_filter and self.df.request_seen(request):
        self.df.log(request, self.spider)
        return False
    if self.stats:
        self.stats.inc_value('scheduler/enqueued/redis', spider=self.spider)
    self.queue.push(request)
    return True

def next_request(self):
    block_pop_timeout = self.idle_before_close
    request = self.queue.pop(block_pop_timeout)
    if request and self.stats:
        self.stats.inc_value('scheduler/dequeued/redis', spider=self.spider)
    return request

enqueue_request() 可以向队列中添加 Request，核心操作就是调用 Queue 的 push() 操作，还有一些统计和日志操作。next_request() 就是从队列中取 Request，核心操作就是调用 Queue 的 pop() 操作，此时如果队列中还有 Request，则 Request 会直接取出来，爬取继续，否则如果队列为空，爬取则会重新开始。

那么到现在为止我们就把之前所说的三个分布式的问题解决了，总结如下：

• 爬取队列的实现，在这里提供了三种队列，使用了 Redis 的列表或有序集合来维护。
• 去重的实现，使用了 Redis 的集合来保存 Request 的指纹来提供重复过滤。
• 中断后重新爬取的实现，中断后 Redis 的队列没有清空，再次启动时调度器的 next_request() 会从队列中取到下一个 Request，继续爬取。

以上内容便是 Scrapy-Redis 的核心源码解析。Scrapy-Redis 中还提供了 Spider、Item Pipeline 的实现，不过它们并不是必须使用。

在下一节，我们会将 Scrapy-Redis 集成到之前所实现的 Scrapy 新浪微博项目中，实现多台主机协同爬取。

三、Scrapy 分布式实现

接下来，我们会利用 Scrapy-Redis 来实现分布式的对接。

3.1 准备工作

请确保已经成功实现了 Scrapy 新浪微博爬虫，Scrapy-Redis 库已经正确安装；

3.2 搭建 Redis 服务器

要实现分布式部署，多台主机需要共享爬取队列和去重集合，而这两部分内容都是存于 Redis 数据库中的，我们需要搭建一个可公网访问的 Redis 服务器。

推荐使用 Linux 服务器，可以购买阿里云、腾讯云、Azure 等提供的云主机，一般都会配有公网 IP，具体的搭建方式可以参考第 1 章中 Redis 数据库的安装方式。

Redis 安装完成之后就可以远程连接了，注意部分商家（如阿里云、腾讯云）的服务器需要配置安全组放通 Redis 运行端口才可以远程访问。如果遇到不能远程连接的问题，可以排查安全组的设置。

需要记录 Redis 的运行 IP、端口、地址，供后面配置分布式爬虫使用。当前配置好的 Redis 的 IP 为服务器的 IP 120.27.34.25，端口为默认的 6379，密码为 foobared。

3.3 部署代理池和 Cookies 池

新浪微博项目需要用到代理池和 Cookies 池，而之前我们的代理池和 Cookies 池都是在本地运行的。所以我们需要将二者放到可以被公网访问的服务器上运行，将代码上传到服务器，修改 Redis 的连接信息配置，用同样的方式运行代理池和 Cookies 池。

远程访问代理池和 Cookies 池提供的接口，来获取随机代理和 Cookies。如果不能远程访问，先确保其在 0.0.0.0 这个 Host 上运行，再检查安全组的配置。

如我当前配置好的代理池和 Cookies 池的运行 IP 都是服务器的 IP，120.27.34.25，端口分别为 5555 和 5556，如图所示。

所以接下来我们就需要把 Scrapy 新浪微博项目中的访问链接修改如下：

PROXY_URL = 'http://120.27.34.25:5555/random'
COOKIES_URL = 'http://120.27.34.25:5556/weibo/random'

具体的修改方式根据实际配置的 IP 和端口做相应调整。

3.4 配置 Scrapy-Redis

配置 Scrapy-Redis 非常简单，只需要修改一下 settings.py 配置文件即可。

核心配置

首先最主要的是，需要将调度器的类和去重的类替换为 Scrapy-Redis 提供的类，在 settings.py 里面添加如下配置即可：

SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

Redis 连接配置

接下来配置 Redis 的连接信息，这里有两种配置方式。

第一种方式是通过连接字符串配置。我们可以用 Redis 的地址、端口、密码来构造一个 Redis 连接字符串，支持的连接形式如下所示：

redis://[:password]@host:port/db
rediss://[:password]@host:port/db
unix://[:password]@/path/to/socket.sock?db=db

password 是密码，比如要以冒号开头，中括号代表此选项可有可无，host 是 Redis 的地址，port 是运行端口，db 是数据库代号，其值默认是 0。

根据上文中提到我的 Redis 连接信息，构造这个 Redis 的连接字符串如下所示：

redis://:foobared@120.27.34.25:6379

直接在 settings.py 里面配置为 REDIS_URL 变量即可：

REDIS_URL = 'redis://:foobared@120.27.34.25:6379'

第二种配置方式是分项单独配置。这个配置就更加直观明了，如根据我的 Redis 连接信息，可以在 settings.py 中配置如下代码：

REDIS_HOST = '120.27.34.25'
REDIS_PORT = 6379
REDIS_PASSWORD = 'foobared'

这段代码分开配置了 Redis 的地址、端口和密码。

注意，如果配置了 REDIS_URL，那么 Scrapy-Redis 将优先使用 REDIS_URL 连接，会覆盖上面的三项配置。如果想要分项单独配置的话，请不要配置 REDIS_URL。

在本项目中，我选择的是配置 REDIS_URL。

配置调度队列

此项配置是可选的，默认使用 PriorityQueue。如果想要更改配置，可以配置 SCHEDULER_QUEUE_CLASS 变量，如下所示：

SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue'
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.FifoQueue'
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.LifoQueue'

以上三行任选其一配置，即可切换爬取队列的存储方式。

在本项目中不进行任何配置，我们使用默认配置。

配置持久化

此配置是可选的，默认是 False。Scrapy-Redis 默认会在爬取全部完成后清空爬取队列和去重指纹集合。

如果不想自动清空爬取队列和去重指纹集合，可以增加如下配置：

SCHEDULER_PERSIST = True

将 SCHEDULER_PERSIST 设置为 True 之后，爬取队列和去重指纹集合不会在爬取完成后自动清空，如果不配置，默认是 False，即自动清空。

值得注意的是，如果强制中断爬虫的运行，爬取队列和去重指纹集合是不会自动清空的。

在本项目中不进行任何配置，我们使用默认配置。

配置重爬

此配置是可选的，默认是 False。如果配置了持久化或者强制中断了爬虫，那么爬取队列和指纹集合不会被清空，爬虫重新启动之后就会接着上次爬取。如果想重新爬取，我们可以配置重爬的选项：

SCHEDULER_FLUSH_ON_START = True

这样将 SCHEDULER_FLUSH_ON_START 设置为 True 之后，爬虫每次启动时，爬取队列和指纹集合都会清空。所以要做分布式爬取，我们必须保证只能清空一次，否则每个爬虫任务在启动时都清空一次，就会把之前的爬取队列清空，势必会影响分布式爬取。

注意，此配置在单机爬取的时候比较方便，分布式爬取不常用此配置。

在本项目中不进行任何配置，我们使用默认配置。

Pipeline 配置

此配置是可选的，默认不启动 Pipeline。Scrapy-Redis 实现了一个存储到 Redis 的 Item Pipeline，启用了这个 Pipeline 的话，爬虫会把生成的 Item 存储到 Redis 数据库中。在数据量比较大的情况下，我们一般不会这么做。因为 Redis 是基于内存的，我们利用的是它处理速度快的特性，用它来做存储未免太浪费了，配置如下：

ITEM_PIPELINES = {'scrapy_redis.pipelines.RedisPipeline': 300}

本项目不进行任何配置，即不启动 Pipeline。

到此为止，Scrapy-Redis 的配置就完成了。有的选项我们没有配置，但是这些配置在其他 Scrapy 项目中可能用到，要根据具体情况而定。

配置存储目标

之前 Scrapy 新浪微博爬虫项目使用的存储是 MongoDB，而且 MongoDB 是本地运行的，即连接的是 localhost。但是，当爬虫程序分发到各台主机运行的时候，爬虫就会连接各自的的 MongoDB。所以我们需要在各台主机上都安装 MongoDB，这样有两个缺点：一是搭建 MongoDB 环境比较烦琐；二是这样各台主机的爬虫会把爬取结果分散存到各自主机上，不方便统一管理。

所以我们最好将存储目标存到同一个地方，例如都存到同一个 MongoDB 数据库中。我们可以在服务器上搭建一个 MongoDB 服务，或者直接购买 MongoDB 数据存储服务。

这里使用的就是服务器上搭建的的 MongoDB 服务，IP 仍然为 120.27.34.25，用户名为 admin，密码为 admin123。

修改配置 MONGO_URI 为如下：

MONGO_URI = 'mongodb://admin:admin123@120.27.34.25:27017'

到此为止，我们就成功完成了 Scrapy 分布式爬虫的配置了。

运行

接下来将代码部署到各台主机上，记得每台主机都需要配好对应的 Python 环境。

每台主机上都执行如下命令，即可启动爬取：

scrapy crawl weibocn

每台主机启动了此命令之后，就会从配置的 Redis 数据库中调度 Request，做到爬取队列共享和指纹集合共享。同时每台主机占用各自的带宽和处理器，不会互相影响，爬取效率成倍提高。

结果

一段时间后，我们可以用 RedisDesktop 观察远程 Redis 数据库的信息。这里会出现两个 Key：一个叫作 weibocn:dupefilter，用来储存指纹；另一个叫作 weibocn:requests，即爬取队列，如图所示。

随着时间的推移，指纹集合会不断增长，爬取队列会动态变化，爬取的数据也会被储存到 MongoDB 数据库中。

至此 Scrapy 分布式的配置已全部完成。

本节代码地址为：https://github.com/Python3WebSpider/Weibo/tree/distributed，注意这里是 distributed 分支。

本节通过对接 Scrapy-Redis 成功实现了分布式爬虫，但是部署还是有很多不方便的地方。另外，如果爬取量特别大的话，Redis 的内存也是个问题。在后文我们会继续了解相关优化方案。

四、Bloom Filter 的对接

首先回顾一下 Scrapy-Redis 的去重机制。Scrapy-Redis 将 Request 的指纹存储到了 Redis 集合中，每个指纹的长度为 40，例如 27adcc2e8979cdee0c9cecbbe8bf8ff51edefb61 就是一个指纹，它的每一位都是 16 进制数。

我们计算一下用这种方式耗费的存储空间。每个十六进制数占用 4 b，1 个指纹用 40 个十六进制数表示，占用空间为 20 B，1 万个指纹即占用空间 200 KB，1 亿个指纹占用 2 GB。当爬取数量达到上亿级别时，Redis 的占用的内存就会变得很大，而且这仅仅是指纹的存储。Redis 还存储了爬取队列，内存占用会进一步提高，更别说有多个 Scrapy 项目同时爬取的情况了。当爬取达到亿级别规模时，Scrapy-Redis 提供的集合去重已经不能满足我们的要求。所以我们需要使用一个更加节省内存的去重算法 Bloom Filter。

4.1 了解 BloomFilter

Bloom Filter，中文名称叫作布隆过滤器，是 1970 年由 Bloom 提出的，它可以被用来检测一个元素是否在一个集合中。Bloom Filter 的空间利用效率很高，使用它可以大大节省存储空间。Bloom Filter 使用位数组表示一个待检测集合，并可以快速地通过概率算法判断一个元素是否存在于这个集合中。利用这个算法我们可以实现去重效果。

本节我们来了解 Bloom Filter 的基本算法，以及 Scrapy-Redis 中对接 Bloom Filter 的方法。

4.2 BloomFilter 的算法

在 Bloom Filter 中使用位数组来辅助实现检测判断。在初始状态下，我们声明一个包含 m 位的位数组，它的所有位都是 0，如图所示。

现在我们有了一个待检测集合，我们表示为 S={x1, x2, …, xn}，我们接下来需要做的就是检测一个 x 是否已经存在于集合 S 中。在 BloomFilter 算法中首先使用 k 个相互独立的、随机的哈希函数来将这个集合 S 中的每个元素 x1、x2、…、xn 映射到这个长度为 m 的位数组上，哈希函数得到的结果记作位置索引，然后将位数组该位置索引的位置 1。例如这里我们取 k 为 3，即有三个哈希函数，x1 经过三个哈希函数映射得到的结果分别为 1、4、8，x2 经过三个哈希函数映射得到的结果分别为 4、6、10，那么就会将位数组的 1、4、6、8、10 这五位置 1，如图所示：

这时如果再有一个新的元素 x，我们要判断 x 是否属于 S 这个集合，我们便会将仍然用 k 个哈希函数对 x 求映射结果，如果所有结果对应的位数组位置均为 1，那么我们就认为 x 属于 S 这个集合，否则如果有一个不为 1，则 x 不属于 S 集合。

例如一个新元素 x 经过三个哈希函数映射的结果为 4、6、8，对应的位置均为 1，则判断 x 属于 S 这个集合。如果结果为 4、6、7，7 对应的位置为 0，则判定 x 不属于 S 这个集合。

注意这里 m、n、k 满足的关系是 m>nk，也就是说位数组的长度 m 要比集合元素 n 和哈希函数 k 的乘积还要大。

这样的判定方法很高效，但是也是有代价的，它可能把不属于这个集合的元素误认为属于这个集合，我们来估计一下它的错误率。当集合 S={x1, x2,…, xn} 的所有元素都被 k 个哈希函数映射到 m 位的位数组中时，这个位数组中某一位还是 0 的概率是：

因为哈希函数是随机的，所以任意一个哈希函数选中这一位的概率为 1/m，那么 1-1/m 就代表哈希函数一次没有选中这一位的概率，要把 S 完全映射到 m 位数组中，需要做 kn 次哈希运算，所以最后的概率就是 1-1/m 的 kn 次方。

一个不属于 S 的元素 x 如果要被误判定为在 S 中，那么这个概率就是 k 次哈希运算得到的结果对应的位数组位置都为 1，所以误判概率为：

根据：

可以将误判概率转化为：

在给定 m、n 时，可以求出使得 f 最小化的 k 值，在这里将误判概率归纳如下：

m/n	最优 k	k=1	k=2	k=3	k=4	k=5	k=6	k=7	k=8
2	1.39	0.393	0.400
3	2.08	0.283	0.237	0.253
4	2.77	0.221	0.155	0.147	0.160
5	3.46	0.181	0.109	0.092	0.092	0.101
6	4.16	0.154	0.0804	0.0609	0.0561	0.0578	0.0638
7	4.85	0.133	0.0618	0.0423	0.0359	0.0347	0.0364
8	5.55	0.118	0.0489	0.0306	0.024	0.0217	0.0216	0.0229
9	6.24	0.105	0.0397	0.0228	0.0166	0.0141	0.0133	0.0135	0.0145
10	6.93	0.0952	0.0329	0.0174	0.0118	0.00943	0.00844	0.00819	0.00846
11	7.62	0.0869	0.0276	0.0136	0.00864	0.0065	0.00552	0.00513	0.00509
12	8.32	0.08	0.0236	0.0108	0.00646	0.00459	0.00371	0.00329	0.00314
13	9.01	0.074	0.0203	0.00875	0.00492	0.00332	0.00255	0.00217	0.00199
14	9.7	0.0689	0.0177	0.00718	0.00381	0.00244	0.00179	0.00146	0.00129
15	10.4	0.0645	0.0156	0.00596	0.003	0.00183	0.00128	0.001	0.000852
16	11.1	0.0606	0.0138	0.005	0.00239	0.00139	0.000935	0.000702	0.000574
17	11.8	0.0571	0.0123	0.00423	0.00193	0.00107	0.000692	0.000499	0.000394
18	12.5	0.054	0.0111	0.00362	0.00158	0.000839	0.000519	0.00036	0.000275
19	13.2	0.0513	0.00998	0.00312	0.0013	0.000663	0.000394	0.000264	0.000194
20	13.9	0.0488	0.00906	0.0027	0.00108	0.00053	0.000303	0.000196	0.00014
21	14.6	0.0465	0.00825	0.00236	0.000905	0.000427	0.000236	0.000147	0.000101
22	15.2	0.0444	0.00755	0.00207	0.000764	0.000347	0.000185	0.000112	7.46e-05
23	15.9	0.0425	0.00694	0.00183	0.000649	0.000285	0.000147	8.56e-05	5.55e-05
24	16.6	0.0408	0.00639	0.00162	0.000555	0.000235	0.000117	6.63e-05	4.17e-05
25	17.3	0.0392	0.00591	0.00145	0.000478	0.000196	9.44e-05	5.18e-05	3.16e-05
26	18	0.0377	0.00548	0.00129	0.000413	0.000164	7.66e-05	4.08e-05	2.42e-05
27	18.7	0.0364	0.0051	0.00116	0.000359	0.000138	6.26e-05	3.24e-05	1.87e-05
28	19.4	0.0351	0.00475	0.00105	0.000314	0.000117	5.15e-05	2.59e-05	1.46e-05
29	20.1	0.0339	0.00444	0.000949	0.000276	9.96e-05	4.26e-05	2.09e-05	1.14e-05
30	20.8	0.0328	0.00416	0.000862	0.000243	8.53e-05	3.55e-05	1.69e-05	9.01e-06
31	21.5	0.0317	0.0039	0.000785	0.000215	7.33e-05	2.97e-05	1.38e-05	7.16e-06
32	22.2	0.0308	0.00367	0.000717	0.000191	6.33e-05	2.5e-05	1.13e-05	5.73e-06

表中第一列为 m/n 的值，第二列为最优 k 值，其后列为不同 k 值的误判概率，可以看到当 k 值确定时，随着 m/n 的增大，误判概率逐渐变小。当 m/n 的值确定时，当 k 越靠近最优 K 值，误判概率越小。另外误判概率总体来看都是极小的，在容忍此误判概率的情况下，大幅减小存储空间和判定速度是完全值得的。

接下来我们就将 BloomFilter 算法应用到 Scrapy-Redis 分布式爬虫的去重过程中，以解决 Redis 内存不足的问题。

4.3 对接 Scrapy-Redis

实现 BloomFilter 时，我们首先要保证不能破坏 Scrapy-Redis 分布式爬取的运行架构，所以我们需要修改 Scrapy-Redis 的源码，将它的去重类替换掉。同时 BloomFilter 的实现需要借助于一个位数组，所以既然当前架构还是依赖于 Redis 的，那么正好位数组的维护直接使用 Redis 就好了。

首先我们实现一个基本的哈希算法，可以实现将一个值经过哈希运算后映射到一个 m 位位数组的某一位上，代码实现如下：

class HashMap(object):
    def __init__(self, m, seed):
        self.m = m
        self.seed = seed

    def hash(self, value):
        """
        Hash Algorithm
        :param value: Value
        :return: Hash Value
        """
        ret = 0
        for i in range(len(value)):
            ret += self.seed * ret + ord(value[i])
        return (self.m - 1) & ret

在这里新建了一个 HashMap 类，构造函数传入两个值，一个是 m 位数组的位数，另一个是种子值 seed，不同的哈希函数需要有不同的 seed，这样可以保证不同的哈希函数的结果不会碰撞。

在 hash() 方法的实现中，value 是要被处理的内容，在这里我们遍历了该字符的每一位并利用 ord() 方法取到了它的 ASCII 码值，然后混淆 seed 进行迭代求和运算，最终会得到一个数值。这个数值的结果就由 value 和 seed 唯一确定，然后我们再将它和 m 进行按位与运算，即可获取到 m 位数组的映射结果，这样我们就实现了一个由字符串和 seed 来确定的哈希函数。当 m 固定时，只要 seed 值相同，就代表是同一个哈希函数，相同的 value 必然会映射到相同的位置。所以如果我们想要构造几个不同的哈希函数，只需要改变其 seed 就好了，以上便是一个简易的哈希函数的实现。

接下来我们再实现 BloomFilter，BloomFilter 里面需要用到 k 个哈希函数，所以在这里我们需要对这几个哈希函数指定相同的 m 值和不同的 seed 值，在这里构造如下：

BLOOMFILTER_HASH_NUMBER = 6
BLOOMFILTER_BIT = 30

class BloomFilter(object):
    def __init__(self, server, key, bit=BLOOMFILTER_BIT, hash_number=BLOOMFILTER_HASH_NUMBER):
        """
        Initialize BloomFilter
        :param server: Redis Server
        :param key: BloomFilter Key
        :param bit: m = 2 ^ bit
        :param hash_number: the number of hash function
        """
        # default to 1 << 30 = 10,7374,1824 = 2^30 = 128MB, max filter 2^30/hash_number = 1,7895,6970 fingerprints
        self.m = 1 << bit
        self.seeds = range(hash_number)
        self.maps = [HashMap(self.m, seed) for seed in self.seeds]
        self.server = server
        self.key = key

由于我们需要亿级别的数据的去重，即前文介绍的算法中的 n 为 1 亿以上，哈希函数的个数 k 大约取 10 左右的量级，而 m>kn，所以这里 m 值大约保底在 10 亿，由于这个数值比较大，所以这里用移位操作来实现，传入位数 bit，定义 30，然后做一个移位操作 1 << 30，相当于 2 的 30 次方，等于 1073741824，量级也是恰好在 10 亿左右，由于是位数组，所以这个位数组占用的大小就是 2^30b=128MB，而本文开头我们计算过 Scrapy-Redis 集合去重的占用空间大约在 2G 左右，可见 BloomFilter 的空间利用效率之高。

随后我们再传入哈希函数的个数，用它来生成几个不同的 seed，用不同的 seed 来定义不同的哈希函数，这样我们就可以构造一个哈希函数列表，遍历 seed，构造带有不同 seed 值的 HashMap 对象，保存成变量 maps 供后续使用。

另外 server 就是 Redis 连接对象，key 就是这个 m 位数组的名称。

接下来我们就要实现比较关键的两个方法了，一个是判定元素是否重复的方法 exists()，另一个是添加元素到集合中的方法 insert()，实现如下：

def exists(self, value):
    """
    if value exists
    :param value:
    :return:
    """
    if not value:
        return False
    exist = 1
    for map in self.maps:
        offset = map.hash(value)
        exist = exist & self.server.getbit(self.key, offset)
    return exist

def insert(self, value):
    """
    add value to bloom
    :param value:
    :return:
    """
    for f in self.maps:
        offset = f.hash(value)
        self.server.setbit(self.key, offset, 1)

首先我们先看下 insert() 方法，BloomFilter 算法中会逐个调用哈希函数对放入集合中的元素进行运算得到在 m 位位数组中的映射位置，然后将位数组对应的位置置 1，所以这里在代码中我们遍历了初始化好的哈希函数，然后调用其 hash() 方法算出映射位置 offset，再利用 Redis 的 setbit() 方法将该位置 1。

在 exists() 方法中我们就需要实现判定是否重复的逻辑了，方法参数 value 即为待判断的元素，在这里我们首先定义了一个变量 exist，然后遍历了所有哈希函数对 value 进行哈希运算，得到映射位置，然后我们用 getbit() 方法取得该映射位置的结果，依次进行与运算。这样只有每次 getbit() 得到的结果都为 1 时，最后的 exist 才为 True，即代表 value 属于这个集合。如果其中只要有一次 getbit() 得到的结果为 0，即 m 位数组中有对应的 0 位，那么最终的结果 exist 就为 False，即代表 value 不属于这个集合。这样此方法最后的返回结果就是判定重复与否的结果了。

到现在为止 BloomFilter 的实现就已经完成了，我们可以用一个实例来测试一下，代码如下：

conn = StrictRedis(host='localhost', port=6379, password='foobared')
bf = BloomFilter(conn, 'testbf', 5, 6)
bf.insert('Hello')
bf.insert('World')
result = bf.exists('Hello')
print(bool(result))
result = bf.exists('Python')
print(bool(result))

在这里我们首先定义了一个 Redis 连接对象，然后传递给 BloomFilter，为了避免内存占用过大这里传的位数 bit 比较小，设置为 5，哈希函数的个数设置为 6。

首先我们调用 insert() 方法插入了 Hello 和 World 两个字符串，随后判断了一下 Hello 和 Python 这两个字符串是否存在，最后输出它的结果，运行结果如下：

True
False

很明显，结果完全没有问题，这样我们就借助于 Redis 成功实现了 BloomFilter 的算法。

接下来我们需要继续修改 Scrapy-Redis 的源码，将它的 dupefilter 逻辑替换为 BloomFilter 的逻辑，在这里主要是修改 RFPDupeFilter 类的 request_seen() 方法，实现如下：

def request_seen(self, request):
    fp = self.request_fingerprint(request)
    if self.bf.exists(fp):
        return True
    self.bf.insert(fp)
    return False

首先还是利用 request_fingerprint() 方法获取了 Request 的指纹，然后调用 BloomFilter 的 exists() 方法判定了该指纹是否存在，如果存在，则证明该 Request 是重复的，返回 True，否则调用 BloomFilter 的 insert() 方法将该指纹添加并返回 False，这样就成功利用 BloomFilter 替换了 Scrapy-Redis 的集合去重。

对于 BloomFilter 的初始化定义，我们可以将 init() 方法修改为如下内容：

def __init__(self, server, key, debug, bit, hash_number):
    self.server = server
    self.key = key
    self.debug = debug
    self.bit = bit
    self.hash_number = hash_number
    self.logdupes = True
    self.bf = BloomFilter(server, self.key, bit, hash_number)

其中 bit 和 hash_number 需要使用 from_settings() 方法传递，修改如下：

@classmethod
def from_settings(cls, settings):
    server = get_redis_from_settings(settings)
    key = defaults.DUPEFILTER_KEY % {'timestamp': int(time.time())}
    debug = settings.getbool('DUPEFILTER_DEBUG', DUPEFILTER_DEBUG)
    bit = settings.getint('BLOOMFILTER_BIT', BLOOMFILTER_BIT)
    hash_number = settings.getint('BLOOMFILTER_HASH_NUMBER', BLOOMFILTER_HASH_NUMBER)
    return cls(server, key=key, debug=debug, bit=bit, hash_number=hash_number)

其中常量的定义 DUPEFILTER_DEBUG 和 BLOOMFILTER_BIT 统一定义在 defaults.py 中，默认如下：

BLOOMFILTER_HASH_NUMBER = 6
BLOOMFILTER_BIT = 30

到此为止我们就成功实现了 BloomFilter 和 Scrapy-Redis 的对接。

本节代码地址为：https://github.com/Python3WebSpider/ScrapyRedisBloomFilter。

4.4 使用

为了方便使用，本节的代码已经打包成了一个 Python 包并发布到了 PyPi，链接为：https://pypi.python.org/pypi/scrapy-redis-bloomfilter，因此我们以后如果想使用 ScrapyRedisBloomFilter 直接使用就好了，不需要再自己实现一遍。

我们可以直接使用 Pip 来安装，命令如下：

pip3 install scrapy-redis-bloomfilter

使用的方法和 Scrapy-Redis 基本相似，在这里说明几个关键配置：

# 去重类，要使用 BloomFilter 请替换 DUPEFILTER_CLASS
DUPEFILTER_CLASS = "scrapy_redis_bloomfilter.dupefilter.RFPDupeFilter"
# 哈希函数的个数，默认为 6，可以自行修改
BLOOMFILTER_HASH_NUMBER = 6
# BloomFilter 的 bit 参数，默认 30，占用 128MB 空间，去重量级 1 亿
BLOOMFILTER_BIT = 30

DUPEFILTER_CLASS 是去重类，如果要使用 BloomFilter 需要将 DUPEFILTER_CLASS 修改为该包的去重类。

BLOOMFILTER_HASH_NUMBER 是 BloomFilter 使用的哈希函数的个数，默认为 6，可以根据去重量级自行修改。

BLOOMFILTER_BIT 即前文所介绍的 BloomFilter 类的 bit 参数，它决定了位数组的位数，如果 BLOOMFILTER_BIT 为 30，那么位数组位数为 2 的 30 次方，将占用 Redis 128MB 的存储空间，去重量级在 1 亿左右，即对应爬取量级 1 亿左右。如果爬取量级在 10 亿、20 亿甚至 100 亿，请务必将此参数对应调高。

4.5 测试

在源代码中附有一个测试项目，放在 tests 文件夹，该项目使用了 Scrapy-RedisBloomFilter 来去重，Spider 的实现如下：

from scrapy import Request, Spider

class TestSpider(Spider):
    name = 'test'
    base_url = 'https://www.baidu.com/s?wd='

    def start_requests(self):
        for i in range(10):
            url = self.base_url + str(i)
            yield Request(url, callback=self.parse)

        # Here contains 10 duplicated Requests
        for i in range(100):
            url = self.base_url + str(i)
            yield Request(url, callback=self.parse)

    def parse(self, response):
        self.logger.debug('Response of ' + response.url)

在 start_requests() 方法中首先循环 10 次，构造参数为 0-9 的 URL，然后重新循环了 100 次，构造了参数为 0-99 的 URL，那么这里就会包含 10 个重复的 Request，我们运行项目测试一下：

scrapy crawl test

可以看到最后的输出结果如下：

{'bloomfilter/filtered': 10,
 'downloader/request_bytes': 34021,
 'downloader/request_count': 100,
 'downloader/request_method_count/GET': 100,
 'downloader/response_bytes': 72943,
 'downloader/response_count': 100,
 'downloader/response_status_count/200': 100,
 'finish_reason': 'finished',
 'finish_time': datetime.datetime(2017, 8, 11, 9, 34, 30, 419597),
 'log_count/DEBUG': 202,
 'log_count/INFO': 7,
 'memusage/max': 54153216,
 'memusage/startup': 54153216,
 'response_received_count': 100,
 'scheduler/dequeued/redis': 100,
 'scheduler/enqueued/redis': 100,
 'start_time': datetime.datetime(2017, 8, 11, 9, 34, 26, 495018)}

可以看到最后统计的第一行的结果：

'bloomfilter/filtered': 10,

这就是 BloomFilter 过滤后的统计结果，可以看到它的过滤个数为 10 个，也就是它成功将重复的 10 个 Reqeust 识别出来了，测试通过。

以上便是 BloomFilter 的原理及对接实现，使用了 BloomFilter 可以大大节省 Redis 内存，在数据量大的情况下推荐使用此方案。