class Scheduler(object):

 def __init__(self, dupefilter, jobdir=None, dqclass=None, mqclass=None,
 logunser=False, stats=None, pqclass=None):
 self.df = dupefilter
 self.dqdir = self._dqdir(jobdir)
 self.pqclass = pqclass
 self.dqclass = dqclass
 self.mqclass = mqclass
 self.logunser = logunser
 self.stats = stats
 # 注意在scrpy中优先注意这个方法，此方法是一个钩子 用于访问当前爬虫的配置
 @classmethod
 def from_crawler(cls, crawler):
 settings = crawler.settings
 # 获取去重用的类 默认：scrapy.dupefilters.RFPDupeFilter
 dupefilter_cls = load_object(settings['DUPEFILTER_CLASS'])
 # 对去重类进行配置from_settings 在 scrapy.dupefilters.RFPDupeFilter 43行
 # 这种调用方式对于IDE跳转不是很好  所以需要自己去找
 # @classmethod
 # def from_settings(cls, settings):
 # debug = settings.getbool('DUPEFILTER_DEBUG')
 # return cls(job_dir(settings), debug)
 # 上面就是from_settings方法 其实就是设置工作目录 和是否开启debug
 dupefilter = dupefilter_cls.from_settings(settings)
 # 获取优先级队列 类对象 默认：queuelib.pqueue.PriorityQueue
 pqclass = load_object(settings['SCHEDULER_PRIORITY_QUEUE'])
 # 获取磁盘队列 类对象（SCHEDULER使用磁盘存储 重启不会丢失）
 dqclass = load_object(settings['SCHEDULER_DISK_QUEUE'])
 # 获取内存队列 类对象（SCHEDULER使用内存存储 重启会丢失）
 mqclass = load_object(settings['SCHEDULER_MEMORY_QUEUE'])
 # 是否开启debug
 logunser = settings.getbool('LOG_UNSERIALIZABLE_REQUESTS', settings.getbool('SCHEDULER_DEBUG'))
 # 将这些参数传递给 __init__方法
 return cls(dupefilter, jobdir=job_dir(settings), logunser=logunser,
 stats=crawler.stats, pqclass=pqclass, dqclass=dqclass, mqclass=mqclass)


 def has_pending_requests(self):
 """检查是否有没处理的请求"""
 return len(self) > 0

 def open(self, spider):
 """Engine创建完毕之后会调用这个方法"""
 self.spider = spider
 # 创建一个有优先级的内存队列 实例化对象
 # self.pqclass 默认是：queuelib.pqueue.PriorityQueue
 # self._newmq 会返回一个内存队列的 实例化对象 在110  111 行
 self.mqs = self.pqclass(self._newmq)
 # 如果self.dqdir 有设置 就创建一个磁盘队列 否则self.dqs 为空
 self.dqs = self._dq() if self.dqdir else None
 # 获得一个去重实例对象 open 方法是从BaseDupeFilter继承的
 # 现在我们可以用self.df来去重啦
 return self.df.open()

 def close(self, reason):
 """当然Engine关闭时"""
 # 如果有磁盘队列 则对其进行dump后保存到active.json文件中
 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)

 def enqueue_request(self, request):
 """添加一个Requests进调度队列"""
 # self.df.request_seen是检查这个Request是否已经请求过了 如果有会返回True
 if not request.dont_filter and self.df.request_seen(request):
 # 如果Request的dont_filter属性没有设置（默认为False）和 已经存在则去重
 # 不push进队列
 self.df.log(request, self.spider)
 return False
 # 先尝试将Request push进磁盘队列
 dqok = self._dqpush(request)
 if dqok:
 # 如果成功 则在记录一次状态
 self.stats.inc_value('scheduler/enqueued/disk', spider=self.spider)
 else:
 # 不能添加进磁盘队列则会添加进内存队列
 self._mqpush(request)
 self.stats.inc_value('scheduler/enqueued/memory', spider=self.spider)
 self.stats.inc_value('scheduler/enqueued', spider=self.spider)
 return True

 def next_request(self):
 """从队列中获取一个Request"""
 # 优先从内存队列中获取
 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)
 # 将获取的到Request返回给Engine
 return request

 def __len__(self):
 return len(self.dqs) + len(self.mqs) if self.dqs else len(self.mqs)

 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

class RFPDupeFilter(BaseDupeFilter):
 """Request Fingerprint duplicates filter"""

 def __init__(self, path=None, debug=False):
 self.file = None
 self.fingerprints = set()
 self.logdupes = True
 self.debug = debug
 self.logger = logging.getLogger(__name__)
 if path:
 # 此处可以看到去重其实打开了一个名叫 requests.seen的文件
 # 如果是使用的磁盘的话
 self.file = open(os.path.join(path, 'requests.seen'), 'a+')
 self.file.seek(0)
 self.fingerprints.update(x.rstrip() for x in self.file)

 @classmethod
 def from_settings(cls, settings):
 debug = settings.getbool('DUPEFILTER_DEBUG')
 return cls(job_dir(settings), debug)

 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)

class Scheduler(object):
 """Redis-based scheduler

 Settings
 --------
 SCHEDULER_PERSIST : bool (default: False)
 Whether to persist or clear redis queue.
 SCHEDULER_FLUSH_ON_START : bool (default: False)
 Whether to flush redis queue on start.
 SCHEDULER_IDLE_BEFORE_CLOSE : int (default: 0)
 How many seconds to wait before closing if no message is received.
 SCHEDULER_QUEUE_KEY : str
 Scheduler redis key.
 SCHEDULER_QUEUE_CLASS : str
 Scheduler queue class.
 SCHEDULER_DUPEFILTER_KEY : str
 Scheduler dupefilter redis key.
 SCHEDULER_DUPEFILTER_CLASS : str
 Scheduler dupefilter class.
 SCHEDULER_SERIALIZER : str
 Scheduler serializer.

 """

 def __init__(self, server,
 persist=False,
 flush_on_start=False,
 queue_key=defaults.SCHEDULER_QUEUE_KEY,
 queue_cls=defaults.SCHEDULER_QUEUE_CLASS,
 dupefilter_key=defaults.SCHEDULER_DUPEFILTER_KEY,
 dupefilter_cls=defaults.SCHEDULER_DUPEFILTER_CLASS,
 idle_before_close=0,
 serializer=None):
 """Initialize scheduler.

 Parameters
 ----------
 server : Redis
 这是Redis实例
 persist : bool
 是否在关闭时清空Requests.默认值是False。
 flush_on_start : bool
 是否在启动时清空Requests。 默认值是False。
 queue_key : str
 Request队列的Key名字
 queue_cls : str
 队列的可导入路径（就是使用什么队列）
 dupefilter_key : str
 去重队列的Key
 dupefilter_cls : str
 去重类的可导入路径。
 idle_before_close : int
 等待多久关闭

 """
 if idle_before_close < 0:
 raise TypeError("idle_before_close cannot be negative")

 self.server = server
 self.persist = persist
 self.flush_on_start = flush_on_start
 self.queue_key = queue_key
 self.queue_cls = queue_cls
 self.dupefilter_cls = dupefilter_cls
 self.dupefilter_key = dupefilter_key
 self.idle_before_close = idle_before_close
 self.serializer = serializer
 self.stats = None

 def __len__(self):
 return len(self.queue)

 @classmethod
 def from_settings(cls, settings):
 kwargs = {
 'persist': settings.getbool('SCHEDULER_PERSIST'),
 'flush_on_start': settings.getbool('SCHEDULER_FLUSH_ON_START'),
 'idle_before_close': settings.getint('SCHEDULER_IDLE_BEFORE_CLOSE'),
 }

 # If these values are missing, it means we want to use the defaults.
 optional = {
 # TODO: Use custom prefixes for this settings to note that are
 # specific to scrapy-redis.
 'queue_key': 'SCHEDULER_QUEUE_KEY',
 'queue_cls': 'SCHEDULER_QUEUE_CLASS',
 'dupefilter_key': 'SCHEDULER_DUPEFILTER_KEY',
 # We use the default setting name to keep compatibility.
 'dupefilter_cls': 'DUPEFILTER_CLASS',
 'serializer': 'SCHEDULER_SERIALIZER',
 }
 # 从setting中获取配置组装成dict(具体获取那些配置是optional字典中key)
 for name, setting_name in optional.items():
 val = settings.get(setting_name)
 if val:
 kwargs[name] = val

 # Support serializer as a path to a module.
 if isinstance(kwargs.get('serializer'), six.string_types):
 kwargs['serializer'] = importlib.import_module(kwargs['serializer'])
 # 或得一个Redis连接
  server = connection.from_settings(settings)
 # Ensure the connection is working.
 server.ping()

 return cls(server=server, **kwargs)

 @classmethod
 def from_crawler(cls, crawler):
  instance = cls.from_settings(crawler.settings)
 # FIXME: for now, stats are only supported from this constructor
 instance.stats = crawler.stats
 return instance

 def open(self, spider):
 self.spider = spider

 try:
 # 根据self.queue_cls这个可以导入的类 实例化一个队列
 self.queue = load_object(self.queue_cls)(
 server=self.server,
 spider=spider,
 key=self.queue_key % {'spider': spider.name},
 serializer=self.serializer,
 )
 except TypeError as e:
 raise ValueError("Failed to instantiate queue class '%s': %s",
 self.queue_cls, e)

 try:
 # 根据self.dupefilter_cls这个可以导入的类 实例一个去重集合
 # 默认是集合 可以实现自己的去重方式 比如 bool 去重
 self.df = load_object(self.dupefilter_cls)(
 server=self.server,
 key=self.dupefilter_key % {'spider': spider.name},
 debug=spider.settings.getbool('DUPEFILTER_DEBUG'),
 )
 except TypeError as e:
 raise ValueError("Failed to instantiate dupefilter class '%s': %s",
 self.dupefilter_cls, e)

 if self.flush_on_start:
 self.flush()
 # notice if there are requests already in the queue to resume the crawl
 if len(self.queue):
 spider.log("Resuming crawl (%d requests scheduled)" % len(self.queue))

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

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

 def enqueue_request(self, request):
 """这个和Scrapy本身的一样"""
 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)
 # 向队列里面添加一个Request
 self.queue.push(request)
 return True

 def next_request(self):
 """获取一个Request"""
 block_pop_timeout = self.idle_before_close
 # block_pop_timeout 是一个等待参数 队列没有东西会等待这个时间  超时就会关闭
 request = self.queue.pop(block_pop_timeout)
 if request and self.stats:
 self.stats.inc_value('scheduler/dequeued/redis', spider=self.spider)
 return request

 def has_pending_requests(self):
 return len(self) > 0

class PriorityQueue(Base):
 """Per-spider priority queue abstraction using redis' sorted set"""
 """其实就是使用Redis的有序集合 来对Request进行排序，这样就可以优先级高的在有序集合的顶层 我们只需要"""
 """从上往下依次获取Request即可"""
 def __len__(self):
 """Return the length of the queue"""
 return self.server.zcard(self.key)

 def push(self, request):
 """Push a request"""
 """添加一个Request进队列"""
 # self._encode_request 将Request请求进行序列化
 data = self._encode_request(request)
 """
 d = {
 'url': to_unicode(request.url),  # urls should be safe (safe_string_url)
 'callback': cb,
 'errback': eb,
 'method': request.method,
 'headers': dict(request.headers),
 'body': request.body,
 'cookies': request.cookies,
 'meta': request.meta,
 '_encoding': request._encoding,
 'priority': request.priority,
 'dont_filter': request.dont_filter,
 'flags': request.flags,
 '_class': request.__module__ + '.' + request.__class__.__name__
 }

 data就是上面这个字典的序列化
 在Scrapy.utils.reqser.py 中的request_to_dict方法中处理
 """

 # 在Redis有序集合中数值越小优先级越高(就是会被放在顶层)所以这个位置是取得 相反数
 score = -request.priority
 # We don't use zadd method as the order of arguments change depending on
 # whether the class is Redis or StrictRedis, and the option of using
 # kwargs only accepts strings, not bytes.
 # ZADD 是添加进有序集合
 self.server.execute_command('ZADD', self.key, score, data)

 def pop(self, timeout=0):
 """
 Pop a request
 timeout not support in this queue class
 有序集合不支持超时所以就木有使用timeout了  这个timeout就是挂羊头卖狗肉
 """
 """从有序集合中取出一个Request"""
 # use atomic range/remove using multi/exec
 """使用multi的原因是为了将获取Request和删除Request合并成一个操作(原子性的)在获取到一个元素之后 删除它，因为有序集合 不像list 有pop 这种方式啊"""
 pipe = self.server.pipeline()
 pipe.multi()
 # 取出 顶层第一个
 # zrange :返回有序集 key 中，指定区间内的成员。0,0 就是第一个了
 # zremrangebyrank：移除有序集 key 中，指定排名(rank)区间内的所有成员 0，0也就是第一个了
 # 更多请参考Redis官方文档
 pipe.zrange(self.key, 0, 0).zremrangebyrank(self.key, 0, 0)
 results, count = pipe.execute()
 if results:
 return self._decode_request(results[0])

def request_seen(self, request):
  """Returns True if request was already seen.

 Parameters
 ----------
 request : scrapy.http.Request

 Returns
 -------
 bool

 """
  # 通过self.request_fingerprint 会生一个sha1的指纹
  fp = self.request_fingerprint(request)
  # This returns the number of values added, zero if already exists.
  # 添加进一个Redis集合如果self.key这个集合中存在fp这个指纹会返回1  不存在返回0
  added = self.server.sadd(self.key, fp)
  return added == 0