scrapy-redis源码浅析

原文链接

前言

分析这个项目的源码原因是需要有去重过滤，增量爬取两个功能，而scrapy-redis项目已经帮我们实现了，想看看他是怎么实现的。这里只贴出部分主要代码，查看时请打开源码对照，笔记有点长，建议看的童鞋按部分看。这是第一次分析源码，限于个人水平，如有错误恳请指正，谢谢！

地址：https://github.com/rmax/scrapy-redis/tree/master/src/scrapy_redis

tips: 源码涉及scrapy的方法，不知道的在文档里搜一下就知道它的作用了，redis也是。

• scrapy:https://docs.scrapy.org/en/latest/
• redis:https://www.tutorialspoint.com/redis/redis_quick_guide.htm

正文部分

克隆源码

安装git：https://git-scm.com/download/win

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

查看源码的话还是visual studio code 和pycharm方便，只要安装了某个包，在源码里用Ctrl+鼠标左键点击方法名就可以跳转到包的源码里。在linux下，则麻烦点要安装vim跳转的插件exuberant-ctags，有兴趣可以自己百度。

• 在Windows环境安装需要用到的包

pip install --upgrade scrapy
pip install --upgrade redis

一、init部分

这里仅copy部分关键代码用于理清原理，全部代码请查看GitHub源码。

• 首先看下init文件__init__.py。从当前目录下的connection.py文件中import两个函数get_redis和get_redis_from_settings

1. __init__.py文件

from .connection import (  # NOQA
    get_redis,
    get_redis_from_settings,
)

• 接着先看看get_redis函数。此函数返回一个redis客户端实例。此函数定义了一个redis_cls类，其值为redis.StrictRedis，是从default.py里的设置的值，所有默认值都放在了这个文件。(这里就忽略了)

• 还有一个url，默认为None。如果scrapy的settings.py启用了REDIS_URL这个参数，就会传递到这里，然后调用redis.StrictRedis的类方法from_url，这个方法返回一个连接到传入url的redis客户端对象。如果scrapy的settings.py没有启用REDIS_URL这个参数，则返回一个redis的默认客户端对象，即默认连接到redis://[:password]@localhost:6379/0，而不是给定的redis地址。

2. connection.py文件

def get_redis(**kwargs):
    redis_cls = kwargs.pop('redis_cls', defaults.REDIS_CLS)
    url = kwargs.pop('url', None)
    if url:
        return redis_cls.from_url(url, **kwargs)
    else:
        return redis_cls(**kwargs)

• 再看第二个函数get_redis_from_settings，它有个参数settings。这个函数首先设置了一个从当前目录下的defaults.py获取默认参数的副本，然后再用从scrapy项目的settings.py中获取的有关redis字典型的配置参数来更新替换默认参数；然后引用six库做了python版本兼容，最后返回个redis客户端实例。注：scrapy的getdict方法用于将settings里的配置转为字典

• connection.py

def get_redis_from_settings(settings):
    params = defaults.REDIS_PARAMS.copy()
    params.update(settings.getdict('REDIS_PARAMS'))
    # XXX: Deprecate REDIS_* settings.
    for source, dest in SETTINGS_PARAMS_MAP.items():
        val = settings.get(source)
        if val:
            params[dest] = val

    # Allow ``redis_cls`` to be a path to a class.
    if isinstance(params.get('redis_cls'), six.string_types):
        params['redis_cls'] = load_object(params['redis_cls'])

    return get_redis(**params)

小结

这样__init__之后，就能够实例化一个连接到自己设置的redis地址的redis客户端实例了。

二、scheduler部分

能够连接redis后就要将scrapy请求的url存到redis。这里作者实现了个调度器Scheduler类来替换scrapy默认的调度器scrapy.core.scheduler.Scheduler。在自己项目的配置文件settings.py中设置成SCHEDULER = "scrapy_redis.scheduler.Scheduler"来替换默认的调度器。

SCHEDULER = "scrapy_redis.scheduler.Scheduler"

• scheduler.py就只有一个Scheduler类。先看__init__函数，只需传入一个server参数，即自己的redis实例，其他均是选默认参数。

    def __init__(self, server, ...):
        self.server = server
        ...

• 接着看Scheduler类的类方法from_settings，设置了个字典kargs用于从自己项目的settings.py中读取参数SCHEDULER_PERSIST，SCHEDULER_FLUSH_ON_START，SCHEDULER_IDLE_BEFORE_CLOSE等，其中SCHEDULER_PERSIST这个参数就是用于实现增量爬取功能的，如果为TRUE则已经存入redis队列里的url就会一直保存不会清空，在我们停止了爬虫，下次再继续运行时就可以直接跳过已经在redis队列里的url了；接着这里还有个可选字典optional用于替换刚才提到的初始化函数init里的默认参数；接着将optional设置了的值加入到kwargs里；接着作者为了支持本地文件能像包一样导入，就用importlib.import_module函数转化了下；最后实例化一个连接到自己的redis实例对象，检查对象连通性；return cls()中的cls返回的是一个redis实例作为参数的Scheduler类对象本身，它被下一个类方法from_crawler调用，这样调用类方法后返回cls就会调用这个类的__init__方法再次初始化。

    @classmethod
    def from_settings(cls, settings):
        kwargs = {
            'persist': settings.getbool('SCHEDULER_PERSIST'),
            ...
            }
            
        optional = {
            'queue_key': 'SCHEDULER_QUEUE_KEY',
            ...
            }
            
        for name, setting_name in optional.items():
            val = settings.get(setting_name)
            if val:
                kwargs[name] = val
            
        server = connection.from_settings(settings)
        server.ping()

        return cls(server=server, **kwargs)

scrapy是如何调用自定义scheduler的

• from_crawler也是类方法，它需要传入一个crawler对象作为参数，即是自己项目中的crawler；接着调用该类本身的类方法from_settings，并将crawler.settings作为参数传入，像上面说的一样，就会得到一个包含自己项目配置的redis实例；获取自己项目crawler的stats状态，返回实例。

    @classmethod
    def from_crawler(cls, crawler):
        instance = cls.from_settings(crawler.settings)
        instance.stats = crawler.stats
        return instance

• 接着看open，传入一个spider对象作为参数；用load_object模块加载scrapy-redis项目默认配置中的队列类，默认为scrapy_redis.queue.PriorityQueue，传入参数队列类必须的参数server, spider, key, serializer等（作者在queue.py定义了3中队列类，都是操作redis的，等下我们再看）。跟队列类一样，加载去重过滤类，调用这个类的类方法from_spider，即dupefilter.py里的类方法，而from_spider，传入一个自己项目的spider对象，获取spider对象对应的配置，获取一个连接到自己配置的redis地址的redis对象，然后将自己项目的spider name结合scrapy-redis的默认配置生成spider_name:dupefilter作为过滤去重的redis key，搜索官网可知debug为自己项目的默认值False，最后返回调用此类方法的对象本身，这样就得到了一个连接到自己配置的redis实例和自己项目配置及默认配置的spider对象这两者结合的对象；接着判断是否清空redis的去重队列，默认不清空；通过判断队列长度判断是否还有请求在爬取。(这个过程感觉挺难理解的，下一篇笔记会用视频记录pycharm debug类方法调用的过程)

def open(self, spider):
        self.spider = spider
        ...
        self.df = load_object(self.dupefilter_cls).from_spider(spider)
        ...

dupefilter.py的类方法from_spider

class RFPDupeFilter(BaseDupeFilter):
    @classmethod
    def from_spider(cls, spider):
        settings = spider.settings
        server = get_redis_from_settings(settings)
        dupefilter_key = settings.get("SCHEDULER_DUPEFILTER_KEY", defaults.SCHEDULER_DUPEFILTER_KEY)
        key = dupefilter_key % {'spider': spider.name}
        debug = settings.getbool('DUPEFILTER_DEBUG')
        return cls(server, key=key, debug=debug)

• 接着看close和flush函数，通过persist来确定是否清空去重队列和请求队列，默认False，但是语句为if not False，即为True,所以默认会清空；close的参数reason为scrapy默认异常cancelled操作。

    def close(self, reason):
        if not self.persist:
            self.flush()

    def flush(self):
        self.df.clear()
        self.queue.clear()

queue.py的实例方法clear

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

dupfilter.py的实例方法clear

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

• 再看enqueue_request，顾名思义即为入列请求的函数；传入scrapy的request对象，判断request的dont_filter参数，默认为False和上面得到的去重过滤对象self.df的request_seen方法，它又调用request_fingerprint方法，request_fingerprint方法调用request的默认方法request_fingerprint来获取请求的指纹，然后将指纹作为值存入redis的去重队列中，如果存入成功，则redis返回0，即该请求的指纹没有重复，返回added == 0，if not request.dont_filter and self.df.request_seen(request)也就是if not False and 0 == 0时，调用去重过滤对象self.df的log函数，当这个函数的参数debug为True时启用log库的debug模式记录日志，否则记录日志添加no more duplicates will be shown，记录后将参数self.logdupes设置为False，之后返回False；接着，用scrapy的stats.inc_value收集统计spider的状态(不太理解这个函数，有知道的童鞋可以告知下，谢谢)，最后入列请求后返回True。

    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

dupefilter.py的实例方法request_seen

    def request_seen(self, request):
        fp = self.request_fingerprint(request)
        added = self.server.sadd(self.key, fp)
        return added == 0

dupefilter.py的实例方法log

    def log(self, request, 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

• 最后看看next_request函数；block_pop_timeout为默认值0，调用redis的pop每隔0秒从队列取出一个请求，取出操作使用redis的pipeline，要先执行multi()操作，然后执行取请求操作zrange(0, 0)取出一个请求并用zremrangebyrank(0, 0)删除这个索引对应的请求，然后执行execute获取结果；最后返回解码json格式后的结果；如果取出了请求并且状态不为None时，用scrapy的stats.inc_value收集统计spider的状态，之后返回request请求。

    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

queue.py的实例方法pop

   def pop(self, timeout=0):
        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])

小结

在自己项目的settings替换成scrapy-redis的Scheduler后，就会将项目的crawler对象传入到Scheduler类中，Scheduler类会生成一个自己定义的redis对象，一个去重过滤的df对象，一个存取请求的队列对象queue，再进行一些spider对象状态值的统计，请求的入列、清空等操作，经过这部分就实现了去重过滤功能了。dupefilter.py的方法基本在这里都用到了，所以就不单独再分析了。

三、queue部分

scheduler部分只用到了PriorityQueue有优先级的队列。作者其实在queue.py中实现了FifoQueue、PriorityQueue、LifoQueue3中操作redis的方案，下面逐一看一下。

1. Base类

• 先看3个类都继承了的基类Base；初始化需要一个redis客户端实例server，一个spider实例，一个key，一个默认为None的serializer；如果serializer为None则pass，异常处理，如果没有loads、dumps则报错；初始化参数。

• 接下来是私有方法_encode_request和_decode_request，这里用到了scrapy内置的函数request_to_dict和request_from_dict来实现；_encode_request将请求转为字典类型，用serializer.dumps转换成json格式类型的数据并返回；_decode_request也类似，过程相反，将json类型数据转为字典类型后返回。

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

    def _decode_request(self, encoded_request):
        """Decode an request previously encoded"""
        obj = self.serializer.loads(encoded_request)
        return request_from_dict(obj, self.spider)

• 再有就是__len__、push、pop等，它们都是没有实现的，clear则是通用的实现，用于删除redis指定key

2. FifoQueue类

• FifoQueue类实现了基类Base没有实现的__len__、push、pop方法；__len__返回列表类型的redis队列长度；

• push将编码成json格式的数据存入列表类型的redis队列；

• pop判断timeout参数是否大于0，是则使用redis的brpop(key, timeout)方法，当这个key里面没有值时会等待n秒后才返回tuple类型的数据，返回第一个是key键，第二个是值；如果timeout不大于0，则用rpop方法删除并获取列表中的最后一个元素，当队列里面没有值时，2种方法都会返回None，即data为None，最后如果data不为None返回解码后的请求数据。

    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)

3. PriorityQueue类

• 与FifoQueue类类似，PriorityQueue类也实现了基类Base没有实现的__len__、push、pop方法，不过这里用的是redis的有序集合sorted set；__len__方法用zcard获取有序集合长度；

• push方法用_encode_request获取请求，设置了score值为scrapy的内置属性-request.priority默认值为0，最后用redis的zadd方法将key即spider.name，score，data即request请求等添加到有序集合中。

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

• pop在scheduler部分调用的时候已经分析过了。就是调用redis的pop每隔0秒从队列取出一个请求，取出操作使用redis的pipeline，要先执行multi()操作，然后执行取请求操作zrange(0, 0)取出一个请求并用zremrangebyrank(0, 0)删除这个索引对应的请求，然后执行execute获取结果，如果取出了结果就返回解码后的results[0]即request对象。

    def pop(self, timeout=0):
        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])

4. LifoQueue类

• 同样LifoQueue类也实现了__len__、push、pop方法；__len__、push方法跟FifoQueue类的一致，不再赘述；pop方法，其实也差不多，不过FifoQueue类用的是redis的brpop和rpop用于从最先进入队列删除key（最旧），LifoQueue类用的是blpop和lpop用于从最后进入队列删除key（最新）。

    def pop(self, timeout=0):
        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

小结

作者实现了3中队列类，先进先出队列、优先级队列、后进先出队列，项目用到的是优先级队列。

其实到这里已经能够满足我的去重过滤、增量爬取的需求了。但作者还提供了自己的spiders.py来执行爬取请求，和pipelines.py来将数据存储到redis的功能，有兴趣的接着看。

可选部分

四、spiders部分

先看spider部分，这里定义了3个类，RedisMixin类用于实现从redis队列读取urls，RedisSpider类继承RedisMixin和scrapy的Spider类；RedisCrawlSpider继承RedisMixin和scrapy的CrawlerSpider类；都用于空闲时从redis队列中读取爬取的请求urls。

1. RedisSpider类和RedisCrawlSpider类

• 先看看RedisSpider类和RedisCrawlSpider。它们都实现了scrapy的一个类方法from_crawler。这个类方法是干嘛的呢，不知道，所以去官网文档搜索下Spider from_crawler，发现method -- scrapy.spiders.Spider.from_crawler -- in Spiders这个内容比较符合我们想找的类方法。进去文档搜索from_crawler，的确找到一模一样的类方法from_crawler(crawler, *args, **kwargs)，它是scrapy下的class scrapy.spiders.Spider类下的类方法，用来创建spider对象。

• RedisSpider类和RedisCrawlSpider功能是一样的，就拿RedisSpider这个类来说吧，调用from_crawler类方法，里面的super会继承scrapy.spiders.CrawlSpider创建的spider对象，然后RedisSpider类也就具有了spider对象的所有属性和方法，同时又继承了RedisMixin类，那么类RedisSpider又具有了RedisMixin类的所有属性和方法，所以就可以调用RedisMixin类里的setup_redis方法了。(这个过程感觉挺难理解的，下一篇笔记会用视频记录pycharm debug类方法调用的过程)

from scrapy.spiders import Spider, CrawlSpider

class RedisMixin(object):
    def setup_redis(self, crawler=None):
        pass

class RedisSpider(RedisMixin, Spider):
    @classmethod
    def from_crawler(self, crawler, *args, **kwargs):
        obj = super(RedisSpider, self).from_crawler(crawler, *args, **kwargs)
        obj.setup_redis(crawler)
        return obj

class RedisCrawlSpider(RedisMixin, CrawlSpider):
    @classmethod
    def from_crawler(self, crawler, *args, **kwargs):
        obj = super(RedisCrawlSpider, self).from_crawler(crawler, *args, **kwargs)
        obj.setup_redis(crawler)
        return obj

2. RedisMixin类

• 接着说RedisMixin类，经过调用类方法from_crawler，RedisMixin类已经具有了spider对象的所有属性和方法，那么就可以在RedisMixin类里面使用它们了。

• 首先这个类定义了start_requests函数直接返回next_requests函数，next_requests函数返回一个要调度的request或返回none；

• next_requests函数具体实现：先设置了个标志位use_set，其名为REDIS_START_URLS_AS_SET，其值为default.py设置的默认值False；因为use_set为False，所以fetch_one调用上面说过的init部分生成的redis实例的spop方法，否则调用lpop方法；初始化时found为0，进入循环，redis_batch_size的值为scrapy项目的settings.py设置的CONCURRENT_REQUESTS的值，默认并发值是16；调用fetch_one从redis获取一个redis_key即经过去重过滤的请求url，如果没有获取到请求就说明队列为空，跳出循环；有的话接着调用make_request_from_data方法将字节类型url编码成str类型再返回(这个函数返回make_requests_from_url，但我找不到哪里有定义，不知道是不是作者写错了，在github问也没人回答。。)；如果有返回，则用yield同时处理多个请求url，然后将请求个数加一，并将日志输出。

    def next_requests(self):
        use_set = self.settings.getbool('REDIS_START_URLS_AS_SET', defaults.START_URLS_AS_SET)
        fetch_one = self.server.spop if use_set else self.server.lpop
        found = 0
        while found < self.redis_batch_size:
            data = fetch_one(self.redis_key)
            if not data:
                # Queue empty.
                break
            req = self.make_request_from_data(data)
            if req:
                yield req
                found += 1
            else:
                self.logger.debug("Request not made from data: %r", data)

        if found:
            self.logger.debug("Read %s requests from '%s'", found, self.redis_key)

    def make_request_from_data(self, data):
        url = bytes_to_str(data, self.redis_encoding)
        return self.make_requests_from_url(url)

• 接着看setup_redis方法，按照注释，这是用于设置redis连接和空闲信号的，需要在spider对象设置了它的crawler对象之后才可以被调用，也就是要使用上面提到的RedisSpider和RedisCrawlSpider两个类之后，继承了spider和crawler之后才行；这个方法需要传入参数crawler，默认为None，如果传入值为None则会报错，提示crawler is required；如果没报错说明已经继承了spider和crawler对象，那接着就是从crawler对象获取配置信息等属性；接着判断redis队列中是否有请求url，如果没有则将本项目 的配置赋值给redis_key，然后格式化成'name': 自己项目爬虫名这样的格式；接着用字符串的strip()方法判断redis_key是否为空字符串，是则报错；然后判断redis_batch_size，如果为None则将默认值赋值给它，然后异常处理redis_batch_size是否为整形；再来就是判断redis_encoding，为None则将默认值赋值给它；判断参数都ok后，将参数信息作为日志输出；然后如init部分分析的，生成一个redis客户端对象；最后调用crawler.signals.connect方法，这个方法调用spider_idle方法，spider_idle方法又调用schedule_next_requests方法，schedule_next_requests方法调用next_requests方法从reids队列来获取新的请求url，然后用scrapy的crawler.engine.crawl方法在爬虫空闲时来获取并执行爬取请求，执行完了返回到spider_idle执行raise DontCloseSpider来禁止关闭爬虫spider，正常来说执行完了请求就会关闭爬虫，到无法再获取到新的请求时，也就是redis请求队列没有请求了才会关闭spider。(其实这里有点蒙，作者用了DontCloseSpider来禁止关闭爬虫spider，最后是怎么关闭爬虫的？有知道的童鞋望告知，谢谢！)

class RedisMixin(object):
    """Mixin class to implement reading urls from a redis queue."""
    redis_key = None
    redis_batch_size = None
    redis_encoding = None
    server = None
    def setup_redis(self, crawler=None):
        #...忽略部分内容
        self.server = connection.from_settings(crawler.settings)
        # The idle signal is called when the spider has no requests left,
        # that's when we will schedule new requests from redis queue
        crawler.signals.connect(self.spider_idle, signal=signals.spider_idle)
    
    def schedule_next_requests(self):
        # TODO: While there is capacity, schedule a batch of redis requests.
        for req in self.next_requests():
            self.crawler.engine.crawl(req, spider=self)

    def spider_idle(self):
        """Schedules a request if available, otherwise waits."""
        # XXX: Handle a sentinel to close the spider.
        self.schedule_next_requests()
        raise DontCloseSpider

五、pipelines部分

根据scrapy的原理可知，经过spider模块engine模块scheduler模块后到达pipelines模块，请求url爬取的内容将在这里被处理。

根据文档，自定义pipeline要实现from_crawler、process_item、open_spider、close_spider这几个方法(这里没有实现open_spider、close_spider，有点不理解)。

• 先看__init__，初始化一个server对象，也就是自己的redis客户端对象，一个用于存储爬取数据的item的key，还有个用于编码成json格式的序列化函数，默认使用ScrapyJSONEncoder().encode

• 接着是2个类方法，from_crawler类方法调用from_settings类方法，from_settings类方法首先从项目配置settings.py读取配置，如果读取到有REDIS_ITEMS_KEY这个关键字就将其作为参数添加到params字典中，REDIS_ITEMS_SERIALIZER也是一样，最后return cls(**params)，此时类方法就会再次初始化，将params字典中的参数赋值给__init__中的参数。(这里的类方法跟上面scheduler,spiders提到的是类似的)

• 然后是实现process_item方法，它调用deferToThread方法，这个方法的作用是在线程中运行函数，并将结果作为延迟返回。这个方法传入一个私有方法_process_item来处理item，首先调用item_key格式化，将spidername:items作为key，然后序列化items的内容作为值，最后用rpush将键值对存入redis并返回item，让它继续走完scrapy的流程。

小结

经过spiders.py和pipelines.py后就可以将爬虫爬取的内容存储到自己的redis了。

结语

如果有童鞋看到了最后，手动给你点个赞！真有耐心，哈哈。本人水平有限，有问题欢迎留言交流，谢谢。

公众号往期文章

scrapy过滤重复数据和增量爬取

redis基础笔记

scrapy电影天堂实战(二)创建爬虫项目

scrapy电影天堂实战(一)创建数据库

scrapy基础笔记

在docker镜像中加入环境变量

笔记 | mongodb 入门操作

笔记 | python元类

笔记 | python2和python3使用super()

那些你在python3中可能没用到但应该用的东西

superset docker 部署

开机启动容器里面的程序

博客 | 三步部署hitchhiker-api