Scrapy_redis源码介绍和分析 分布式以及去重的原理解析

本文会对scrapy_redis爬虫的实现原理进行详细介绍，通过查看源码。在读这篇文章之前，你需要补充一些知识点，包括（redis数据库的命令、hashlib模块）

源码分析

pycharm查看源文件的方式不再详细说明，如果不知道的可参考：Pycharm的基本使用。查看源码就需要找一个切入点，源文件肯定是非常的多，我们不能一个一个看。要使用scrapy_redis就需要在settings配置相关信息；

# -*- coding: utf-8 -*-


DUPEFILTER_CLASS='scrapy_redis.dupefilter.RFPDupeFilter'
SCHEDULER='scrapy_redis.scheduler.Scheduler'
SCHEDULER_PERSIST=True

ITEM_PIPELINES = {
   'tencent.pipelines.TencentPipeline': 300,
   'scrapy_redis.pipelines.RedisPipeline':400,
}

DOWNLOAD_DELAY = 1

RedisPipeline

我们从这里切入进行分析，首先来看pipelines文件中的类RedisPipelin源码片段，我只选取重要的部分，其余的自己看:

class RedisPipeline(object):
    .......
    .......  
    def process_item(self, item, spider):#使用了process_item的方法，实现数据的保存
        return deferToThread(self._process_item, item, spider)  #调用一个异步线程去处理这个item

    def _process_item(self, item, spider):
        key = self.item_key(item, spider)
        data = self.serialize(item)
        self.server.rpush(key, data)  #向 爬虫名字：items 中添加item
        return item

这一部分比较简单，我们可以使用RedisPipeline,也可以自定义存放的位置。

RFPDupeFilter

dupefilter文件中存在类RFPDupeFilter，我们来看代码片段：

class RFPDupeFilter(BaseDupeFilter):
    ......
    ......
    def request_seen(self, request):   #判断request对象是否已经存在
       
        fp = self.request_fingerprint(request)  #调用request_fingerprint(self, request)函数
        # This returns the number of values added, zero if already exists.
        added = self.server.sadd(self.key, fp)
        return added == 0   #返回0表示添加失败，即已经存在，否则表示不存在

    def request_fingerprint(self, request):
        
        return request_fingerprint(request)
        #返回request_fingerprint(request)函数

    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()  #sha1加密
        fp.update(to_bytes(request.method))  #请求方法
        fp.update(to_bytes(canonicalize_url(request.url)))  #请求url地址
        fp.update(request.body or b'')  #请求体，post请求才会有
        if include_headers:  #添加请求头，默认不添加请求头（因为header的cookies中含有session
                             #id，这在不同的网站中是随机的，会给sha1的计算带来误差）
            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()  #返回加密之后的16进制
    return cache[include_headers]

那么scrapy_redis去重方法，就是使用sha1加密request得到指纹，再把指纹存在redis集合中，下一次新来一个request，同样的方式生成指纹，判断指纹是否存在redis的集合中。

判断数据是否存在redis数据库的集合中，不存在插入是通过request_seen函数来实现，原理你需要知道redis的相关用法，这里不再介绍，向reids插入数据，返回0，表示已经存在，即该函数会返回True。

Schedulter

schedulter文件夹中存在类Schedulter，代码片段：

class Scheduler(object):
    ......
    ......

    def close(self, reason):
        if not self.persist: #如果在settings中设置为不持久，那么在退出的时候会清空
            self.flush()

    def flush(self):
        self.df.clear() 指的是存放dupefilter的redis
        self.queue.clear()#指的是存放requests的redis
    def enqueue_request(self, request):
        if not request.dont_filter and self.df.request_seen(request):  #不能加入待爬队列的条件
        #当前url需要（经过allow_domain）过滤，且request不存在
        #对于页面会更新的网址，可以设置dont_filter为True让其能够被反复抓取。
        #dont_filter=False True True request指纹已经存在，不会入队
        #dont_filter=False Ture False request指纹已经存在 会入队
        #dont_filter=True False 会入队
            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