scrapy源码2：scheduler的源码分析

绝地灬酷狼 2023-02-22 10:37 1阅读 0赞

### 一. scheduler核心 ###

Scheduler主要负责scrapy请求队列的管理，即进队与出队。进一步来说，会涉及到队列的选择，队列去重，序列化。

from_crawler(cls, crawler):
            settings = crawler.settings
            dupefilter_cls = load_object(settings['DUPEFILTER_CLASS'])
            dupefilter = dupefilter_cls.from_settings(settings)
            pqclass = load_object(settings['SCHEDULER_PRIORITY_QUEUE'])
            dqclass = load_object(settings['SCHEDULER_DISK_QUEUE'])
            mqclass = load_object(settings['SCHEDULER_MEMORY_QUEUE'])
            logunser = settings.getbool('LOG_UNSERIALIZABLE_REQUESTS', settings.getbool('SCHEDULER_DEBUG'))
            return cls(dupefilter, jobdir=job_dir(settings), logunser=logunser,
                       stats=crawler.stats, pqclass=pqclass, dqclass=dqclass, mqclass=mqclass)

创建了4个对象，分别是dupefilter,pqclass,dqclass,mqclass。

##### 1. dupefilter过滤器（url去重） #####

DUPEFILTER\_CLASS = ‘scrapy.dupefilters.RFPDupeFilter’这个类的含义是"Request Fingerprint duplicates filter"，请求指纹副本过滤。也就是对每个request请求做一个指纹，保证相同的请求有相同的指纹。对重复的请求进行过滤。包含查询字符串、cookies字段的相同url也会被去重。

class RFPDupeFilter(BaseDupeFilter):
    
        def request_seen(self, request):
            fp = self.request_fingerprint(request)
            if fp in self.fingerprints:
                return True
            self.fingerprints.add(fp)
            if self.file:
                self.file.write(fp + os.linesep)
    
        def request_fingerprint(self, request):
            return request_fingerprint(request)

scrapy默认的去重方案：利用request生成fingerprint, 存入set，每次利用set判断，如果用了 disk queue 追加至文件

#### 2. pqclass优先级队列 ####

SCHEDULER\_PRIORITY\_QUEUE = ‘queuelib.PriorityQueue’这是一个优先级队列，使用的是开源的第三方queuelib.它的作用就是对request请求按优先级进行排序，这样我们可以对不同重要性的URL指定优先级（通过设置Request的priority属性）。优先级是一个整数，虽然queuelib使用小的数做为高优化级，但是由于scheduler入队列时取了负值，所以对于我们来说，数值越大优先级越高。

#### 3. dqclass支持序列化的后进先出的磁盘队列 ####

SCHEDULER\_DISK\_QUEUE = ‘scrapy.squeues.PickleLifoDiskQueue’  
这是一个支持序列化的后进先出的磁盘队列。主要用来帮助我们在停止爬虫后可以接着上一次继续开始爬虫。

#### 4. mqclass后进先出的内存队列 ####

SCHEDULER\_MEMORY\_QUEUE = ‘scrapy.squeues.LifoMemoryQueue’从名字上看，是后进先出的内存队列。这个队列是为了使用2中的队列而存在的，不必单独分析。

### 二. scheduler源码解释笔记 ###

# scheduler.py
    import os
    import json
    import logging
    from os.path import join, exists
    
    # request_to_dict: 将请求对象转换成dict，如果给定了一个spider，它将尝试找出回调中使用的spider方法的名称，并将其存储为回调。
    # request_from_dict：从dict创建请求对象，如果给定了一个spider，它将尝试解析在spider中查找同名方法的回调。
    from scrapy.utils.reqser import request_to_dict, request_from_dict
    from scrapy.utils.misc import load_object, create_instance
    from scrapy.utils.job import job_dir
    
    logger = logging.getLogger(__name__)  # 获得一个全局的logger对象。
    
    
    class Scheduler(object):
    
        def __init__(self, dupefilter, jobdir=None, dqclass=None, mqclass=None,
                     logunser=False, stats=None, pqclass=None):
            self.df = dupefilter  # 去重模块 默认利用set在内存去重
            self.dqdir = self._dqdir(jobdir)  # 磁盘队列路径 持久化队列至硬盘
            self.pqclass = pqclass  # 带优先级队列 默认来自queuelib
            self.dqclass = dqclass  # 磁盘队列 持久化队列至硬盘
            self.mqclass = mqclass  # 内存队列 默认来自queuelib
            self.logunser = logunser
            self.stats = stats  # 状态记录 状态记录通用模块
    
        # 从crawler的设置获取各个属性， 然后使用load_object 获取对应类。
        # 主要有以下几个名词， 调度优先级队列，调度磁盘队列，调度内存队列。调度debug开启是否，日志非序列化请求，重复类。
        @classmethod
        def from_crawler(cls, crawler):  # 实例化入口 scrapy风格的实例化入口
            settings = crawler.settings
            dupefilter_cls = load_object(settings['DUPEFILTER_CLASS'])
            dupefilter = create_instance(dupefilter_cls, settings, crawler)
            pqclass = load_object(settings['SCHEDULER_PRIORITY_QUEUE'])  # 'queuelib.PriorityQueue'
            dqclass = load_object(settings['SCHEDULER_DISK_QUEUE'])  # 'scrapy.squeues.PickleLifoDiskQueue'
            mqclass = load_object(settings['SCHEDULER_MEMORY_QUEUE'])  # 'scrapy.squeues.LifoMemoryQueue'
            logunser = settings.getbool('LOG_UNSERIALIZABLE_REQUESTS', settings.getbool('SCHEDULER_DEBUG'))
            return cls(dupefilter, jobdir=job_dir(settings), logunser=logunser,
                       stats=crawler.stats, pqclass=pqclass, dqclass=dqclass, mqclass=mqclass)
    
        # 获取是否还有请求没处理。 返回true或者false.
        def has_pending_requests(self):  # 检查队列数 指向len
            return len(self) > 0
    
        # 打开调度器方法： 设置当前的爬虫，设置当前的内存队列，磁盘队列，内存队列初始值为调度优先级队列。
        def open(self, spider):  # 初始化队列 scrapy模块的初始化入口
            self.spider = spider
            self.mqs = self.pqclass(self._newmq)
            self.dqs = self._dq() if self.dqdir else None
            return self.df.open()
    
        # 关闭调度器方法： 判断dqs， 关闭dqs，打开active.json文件， 把prios信息写进去。关闭df
        def close(self, reason):  # 安全退出接口 scrapy模块的安全入口
            if self.dqs:
                prios = self.dqs.close()
                with open(join(self.dqdir, 'active.json'), 'w') as f:
                    json.dump(prios, f)
            return self.df.close(reason)
    
        # enqueue_request: 请求进队列
        # 如果请求是不过滤的，过滤器df的请求处理过。记录日志， 返回false
        def enqueue_request(self, request):  # 进队api 调度进队
            if not request.dont_filter and self.df.request_seen(request):
                self.df.log(request, self.spider)
                return False
            dqok = self._dqpush(request)  # self._dqpush 这个是磁盘队列加入这个请求
            if dqok:  # 如果成功，就给统计信息的disk的对应爬虫加1
                self.stats.inc_value('scheduler/enqueued/disk', spider=self.spider)
            else:  # 其他情况的话，就给统计信息的memory的对应爬虫加1
                self._mqpush(request)
                self.stats.inc_value('scheduler/enqueued/memory', spider=self.spider)
            self.stats.inc_value('scheduler/enqueued', spider=self.spider)  # 总的也需要加1，然后返回true
            return True
    
        # next_request：从队列里取出数据进行处理
        # 获取下一个请求， 先从内存队列mqs里面pop一个，给memory加1，如果内存中为空就从磁盘队列dq里面pop一个。
        # 然后disk加1，如果request不为空， 就给dequed加1
        # 注意这个方法和上个方法， 一个是入，一个是出 的。
        # 统计信息也是， 一个统计到en队列中， 一个统计到de队列去。
        def next_request(self):  # 出队api 调度出队，t优先从内存队列里取，然后才是磁盘队列
            request = self.mqs.pop()
            if request:
                self.stats.inc_value('scheduler/dequeued/memory', spider=self.spider)
            else:
                request = self._dqpop()
                if request:
                    self.stats.inc_value('scheduler/dequeued/disk', spider=self.spider)
            if request:
                self.stats.inc_value('scheduler/dequeued', spider=self.spider)
            return request
    
        # 三目运算， 如果磁盘队列不为空的话， 就是磁盘队列加内存队列的长度， 否则就是内存队列的长度。
        def __len__(self):
            return len(self.dqs) + len(self.mqs) if self.dqs else len(self.mqs)
    
        """ 下面的几个方法，_dqpush，_mqpush，_dqpop，_newmq，_newdq，_dq，_dqdir 从方法名字，看看有push,you pop， new ， 大概可以知道这个是构造，添加，删除操作。 _newmq: 构造一个内存队列， _newdq: 构造一个磁盘队列。 具体的列可以看出从setting里面读取过来的。实例化的在这个方面里面的。 返回值就是对应的对象。 _mqpush: 内存队列里面push一个请求， 优先级为请求的负值 _dqpush: 核心就是请求转化为字典， 然后把dict放到磁盘队列中， 如果有异常，说明是无法序列化请求，构造msg信息。 记录警告信息。并记录序列化失败个数，最总返回true，有异常那就返回none _dqpop: 从磁盘队列获取pop一个dict，然后将dict转为request。 返回回去。 _dqdir: 获取dqdir，如果有设置的话， 就会创建一个目录，并返回这个目录 """
    
        def _dqpush(self, request):
            if self.dqs is None:
                return
            try:
                reqd = request_to_dict(request, self.spider)
                self.dqs.push(reqd, -request.priority)
            except ValueError as e:  # non serializable request
                if self.logunser:
                    msg = ("Unable to serialize request: %(request)s - reason:"
                           " %(reason)s - no more unserializable requests will be"
                           " logged (stats being collected)")
                    logger.warning(msg, { 'request': request, 'reason': e},
                                   exc_info=True, extra={ 'spider': self.spider})
                    self.logunser = False
                self.stats.inc_value('scheduler/unserializable',
                                     spider=self.spider)
                return
            else:
                return True
    
        def _mqpush(self, request):
            self.mqs.push(request, -request.priority)
    
        def _dqpop(self):
            if self.dqs:
                d = self.dqs.pop()
                if d:
                    return request_from_dict(d, self.spider)
    
        def _newmq(self, priority):
            return self.mqclass()
    
        def _newdq(self, priority):
            return self.dqclass(join(self.dqdir, 'p%s' % priority))
    
        def _dq(self):
            activef = join(self.dqdir, 'active.json')
            if exists(activef):
                with open(activef) as f:
                    prios = json.load(f)
            else:
                prios = ()
            q = self.pqclass(self._newdq, startprios=prios)
            if q:
                logger.info("Resuming crawl (%(queuesize)d requests scheduled)",
                            { 'queuesize': len(q)}, extra={ 'spider': self.spider})
            return q
    
        def _dqdir(self, jobdir):
            if jobdir:
                dqdir = join(jobdir, 'requests.queue')
                if not exists(dqdir):
                    os.makedirs(dqdir)
                return dqdir