Scrapy

Scrapy是一個比較好用的Python爬蟲框架，你只需要編寫幾個元件就可以實現網頁資料的爬取。但是當我們要爬取的頁面非常多的時候，單個主機的處理能力就不能滿足我們的需求了（無論是處理速度還是網路請求的併發數），這時候分散式爬蟲的優勢就顯現出來。

而Scrapy-Redis則是一個基於Redis的Scrapy分散式元件。它利用Redis對用於爬取的請求(Requests)進行儲存和排程(Schedule)，並對爬取產生的專案(items)儲存以供後續處理使用。scrapy-redi重寫了scrapy一些比較關鍵的程式碼，將scrapy變成一個可以在多個主機上同時執行的分散式爬蟲。

原生的Scrapy的架構是這樣子的：

加上了Scrapy-Redis之後的架構變成了：

scrapy-redis的官方文件寫的比較簡潔，沒有提及其執行原理，所以如果想全面的理解分散式爬蟲的執行原理，還是得看scrapy-redis的原始碼才行，不過scrapy-redis的原始碼很少，也比較好懂，很快就能看完。

scrapy-redis工程的主體還是是redis和scrapy兩個庫，工程本身實現的東西不是很多，這個工程就像膠水一樣，把這兩個外掛粘結了起來。

scrapy-redis提供了哪些元件？

scrapy-redis所實現的兩種分散式：爬蟲分散式以及item處理分散式。分別是由模組scheduler和模組pipelines實現。

connection.py

負責根據setting中配置例項化redis連線。被dupefilter和scheduler呼叫，總之涉及到redis存取的都要使用到這個模組。

import redis
import six
from scrapy.utils.misc import load_object
DEFAULT_REDIS_CLS = redis.StrictRedis
# Sane connection defaults.
DEFAULT_PARAMS = {
 'socket_timeout': 30,
 'socket_connect_timeout': 30,
 'retry_on_timeout': True,
}
# Shortcut maps 'setting name' -> 'parmater name'.
SETTINGS_PARAMS_MAP = {
 'REDIS_URL': 'url',
 'REDIS_HOST': 'host',
 'REDIS_PORT': 'port',
}
def get_redis_from_settings(settings):
 """Returns a redis client instance from given Scrapy settings object.
 This function uses ``get_client`` to instantiate the client and uses
 ``DEFAULT_PARAMS`` global as defaults values for the parameters. You can
 override them using the ``REDIS_PARAMS`` setting.
 Parameters
 ----------
 settings : Settings
  A scrapy settings object. See the supported settings below.
 Returns
 -------
 server
  Redis client instance.
 Other Parameters
 ----------------
 REDIS_URL : str, optional
  Server connection URL.
 REDIS_HOST : str, optional
  Server host.
 REDIS_PORT : str, optional
  Server port.
 REDIS_PARAMS : dict, optional
  Additional client parameters.
 """
 params = DEFAULT_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)
# Backwards compatible alias.
from_settings = get_redis_from_settings
def get_redis(**kwargs):
 """Returns a redis client instance.
 Parameters
 ----------
 redis_cls : class, optional
  Defaults to ``redis.StrictRedis``.
 url : str, optional
  If given, ``redis_cls.from_url`` is used to instantiate the class.
 **kwargs
  Extra parameters to be passed to the ``redis_cls`` class.
 Returns
 -------
 server
  Redis client instance.
 """
 redis_cls = kwargs.pop('redis_cls', DEFAULT_REDIS_CLS)
 url = kwargs.pop('url', None)
 if url:
  return redis_cls.from_url(url, **kwargs)
 else:
  return redis_cls(**kwargs)

connect檔案引入了redis模組，這個是redis-python庫的介面，用於通過python訪問redis資料庫，可見，這個檔案主要是實現連線redis資料庫的功能（返回的是redis庫的Redis物件或者StrictRedis物件，這倆都是可以直接用來進行資料操作的物件）。這些連線介面在其他檔案中經常被用到。其中，我們可以看到，要想連線到redis資料庫，和其他資料庫差不多，需要一個ip地址、埠號、使用者名稱密碼（可選）和一個整形的資料庫編號，同時我們還可以在scrapy工程的setting檔案中配置套接字的超時時間、等待時間等。

dupefilter.py

負責執行requst的去重，實現的很有技巧性，使用redis的set資料結構。但是注意scheduler並不使用其中用於在這個模組中實現的dupefilter鍵做request的排程，而是使用queue.py模組中實現的queue。當request不重複時，將其存入到queue中，排程時將其彈出。

import logging
import time
from scrapy.dupefilters import BaseDupeFilter
from scrapy.utils.request import request_fingerprint
from .connection import get_redis_from_settings
DEFAULT_DUPEFILTER_KEY = "dupefilter:%(timestamp)s"
logger = logging.getLogger(__name__)
# TODO: Rename class to RedisDupeFilter.
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)
  # XXX: This creates one-time key. needed to support to use this
  # class as standalone dupefilter with scrapy's default scheduler
  # if scrapy passes spider on open() method this wouldn't be needed
  # TODO: Use SCRAPY_JOB env as default and fallback to timestamp.
  key = DEFAULT_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)
  # This returns the number of values added, zero if already exists.
  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)")
   msg = "Filtered duplicate request: %(request)s"
   self.logger.debug(msg, {'request': request}, extra={'spider': spider})
   self.logdupes = False

這個檔案看起來比較複雜，重寫了scrapy本身已經實現的request判重功能。因為本身scrapy單機跑的話，只需要讀取記憶體中的request佇列或者持久化的request佇列（scrapy預設的持久化似乎是json格式的檔案，不是資料庫）就能判斷這次要發出的request url是否已經請求過或者正在排程（本地讀就行了）。而分散式跑的話，就需要各個主機上的scheduler都連線同一個資料庫的同一個request池來判斷這次的請求是否是重複的了。

在這個檔案中，通過繼承BaseDupeFilter重寫他的方法，實現了基於redis的判重。根據原始碼來看，scrapy-redis使用了scrapy本身的一個fingerprint接request_fingerprint，這個介面很有趣，根據scrapy文件所說，他通過hash來判斷兩個url是否相同（相同的url會生成相同的hash結果），但是當兩個url的地址相同，get型引數相同但是順序不同時，也會生成相同的hash結果（這個真的比較神奇。。。）所以scrapy-redis依舊使用url的fingerprint來判斷request請求是否已經出現過。這個類通過連線redis，使用一個key來向redis的一個set中插入fingerprint（這個key對於同一種spider是相同的，redis是一個key-value的資料庫，如果key是相同的，訪問到的值就是相同的，這裡使用spider名字+DupeFilter的key就是為了在不同主機上的不同爬蟲例項，只要屬於同一種spider，就會訪問到同一個set，而這個set就是他們的url判重池），如果返回值為0，說明該set中該fingerprint已經存在（因為集合是沒有重複值的），則返回False，如果返回值為1，說明添加了一個fingerprint到set中，則說明這個request沒有重複，於是返回True，還順便把新fingerprint加入到資料庫中了。 DupeFilter判重會在scheduler類中用到，每一個request在進入排程之前都要進行判重，如果重複就不需要參加排程，直接捨棄就好了，不然就是白白浪費資源。

queue.py

其作用如dupefilter.py所述，但是這裡實現了三種方式的queue：FIFO的SpiderQueue，SpiderPriorityQueue，以及LIFI的SpiderStack。預設使用的是第二種，這也就是出現之前文章中所分析情況的原因（連結）。

from scrapy.utils.reqser import request_to_dict, request_from_dict
from . import picklecompat
class Base(object):
 """Per-spider queue/stack base class"""
 def __init__(self, server, spider, key, serializer=None):
  """Initialize per-spider redis queue.
  Parameters:
   server -- redis connection
   spider -- spider instance
   key -- key for this queue (e.g. "%(spider)s:queue")
  """
  if serializer is None:
   # Backward compatibility.
   # TODO: deprecate pickle.
   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):
  """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)
 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)
class SpiderQueue(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)
class SpiderPriorityQueue(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
  # 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.
  self.server.execute_command('ZADD', self.key, score, data)
 def pop(self, timeout=0):
  """
  Pop a request
  timeout not support in this queue class
  """
  # use atomic range/remove using multi/exec
  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])
class SpiderStack(Base):
 """Per-spider stack"""
 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)
__all__ = ['SpiderQueue', 'SpiderPriorityQueue', 'SpiderStack']

該檔案實現了幾個容器類，可以看這些容器和redis互動頻繁，同時使用了我們上邊picklecompat中定義的serializer。這個檔案實現的幾個容器大體相同，只不過一個是佇列，一個是棧，一個是優先順序佇列，這三個容器到時候會被scheduler物件例項化，來實現request的排程。比如我們使用SpiderQueue最為排程佇列的型別，到時候request的排程方法就是先進先出，而實用SpiderStack就是先進後出了。

我們可以仔細看看SpiderQueue的實現，他的push函式就和其他容器的一樣，只不過push進去的request請求先被scrapy的介面request_to_dict變成了一個dict物件（因為request物件實在是比較複雜，有方法有屬性不好序列化），之後使用picklecompat中的serializer序列化為字串，然後使用一個特定的key存入redis中（該key在同一種spider中是相同的）。而呼叫pop時，其實就是從redis用那個特定的key去讀其值（一個list），從list中讀取最早進去的那個，於是就先進先出了。

這些容器類都會作為scheduler排程request的容器，scheduler在每個主機上都會例項化一個，並且和spider一一對應，所以分散式執行時會有一個spider的多個例項和一個scheduler的多個例項存在於不同的主機上，但是，因為scheduler都是用相同的容器，而這些容器都連線同一個redis伺服器，又都使用spider名加queue來作為key讀寫資料，所以不同主機上的不同爬蟲例項公用一個request排程池，實現了分散式爬蟲之間的統一排程。

picklecompat.py

"""A pickle wrapper module with protocol=-1 by default."""
try:
 import cPickle as pickle # PY2
except ImportError:
 import pickle
def loads(s):
 return pickle.loads(s)
def dumps(obj):
return pickle.dumps(obj, protocol=-1)

這裡實現了loads和dumps兩個函式，其實就是實現了一個serializer，因為redis資料庫不能儲存複雜物件（value部分只能是字串，字串列表，字串集合和hash，key部分只能是字串），所以我們存啥都要先序列化成文字才行。這裡使用的就是python的pickle模組，一個相容py2和py3的序列化工具。這個serializer主要用於一會的scheduler存reuqest物件，至於為什麼不實用json格式，我也不是很懂，item pipeline的序列化預設用的就是json。

pipelines.py

這是是用來實現分散式處理的作用。它將Item儲存在redis中以實現分散式處理。另外可以發現，同樣是編寫pipelines，在這裡的編碼實現不同於文章中所分析的情況，由於在這裡需要讀取配置，所以就用到了from_crawler()函式。

from scrapy.utils.misc import load_object
from scrapy.utils.serialize import ScrapyJSONEncoder
from twisted.internet.threads import deferToThread
from . import connection
default_serialize = ScrapyJSONEncoder().encode
class RedisPipeline(object):
 """Pushes serialized item into a redis list/queue"""
 def __init__(self, server,
     key='%(spider)s:items',
     serialize_func=default_serialize):
  self.server = server
  self.key = key
  self.serialize = serialize_func
 @classmethod
 def from_settings(cls, settings):
  params = {
   'server': connection.from_settings(settings),
  }
  if settings.get('REDIS_ITEMS_KEY'):
   params['key'] = settings['REDIS_ITEMS_KEY']
  if settings.get('REDIS_ITEMS_SERIALIZER'):
   params['serialize_func'] = load_object(
    settings['REDIS_ITEMS_SERIALIZER']
   )
  return cls(**params)
 @classmethod
 def from_crawler(cls, crawler):
  return cls.from_settings(crawler.settings)
 def process_item(self, item, spider):
  return deferToThread(self._process_item, item, spider)
 def _process_item(self, item, spider):
  key = self.item_key(item, spider)
  data = self.serialize(item)
  self.server.rpush(key, data)
  return item
 def item_key(self, item, spider):
  """Returns redis key based on given spider.
  Override this function to use a different key depending on the item
  and/or spider.
  """
  return self.key % {'spider': spider.name}

pipeline檔案實現了一個item pipieline類，和scrapy的item pipeline是同一個物件，通過從settings中拿到我們配置的REDIS_ITEMS_KEY作為key，把item序列化之後存入redis資料庫對應的value中（這個value可以看出出是個list，我們的每個item是這個list中的一個結點），這個pipeline把提取出的item存起來，主要是為了方便我們延後處理資料。

scheduler.py

此擴充套件是對scrapy中自帶的scheduler的替代（在settings的SCHEDULER變數中指出），正是利用此擴充套件實現crawler的分散式排程。其利用的資料結構來自於queue中實現的資料結構。

scrapy-redis所實現的兩種分散式：爬蟲分散式以及item處理分散式就是由模組scheduler和模組pipelines實現。上述其它模組作為為二者輔助的功能模組。

import importlib
import six
from scrapy.utils.misc import load_object
from . import connection
# TODO: add SCRAPY_JOB support.
class Scheduler(object):
 """Redis-based scheduler"""
 def __init__(self, server,
     persist=False,
     flush_on_start=False,
     queue_key='%(spider)s:requests',
     queue_cls='scrapy_redis.queue.SpiderPriorityQueue',
     dupefilter_key='%(spider)s:dupefilter',
     dupefilter_cls='scrapy_redis.dupefilter.RFPDupeFilter',
     idle_before_close=0,
     serializer=None):
  """Initialize scheduler.
  Parameters
  ----------
  server : Redis
   The redis server instance.
  persist : bool
   Whether to flush requests when closing. Default is False.
  flush_on_start : bool
   Whether to flush requests on start. Default is False.
  queue_key : str
   Requests queue key.
  queue_cls : str
   Importable path to the queue class.
  dupefilter_key : str
   Duplicates filter key.
  dupefilter_cls : str
   Importable path to the dupefilter class.
  idle_before_close : int
   Timeout before giving up.
  """
  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',
  }
  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'])
  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 = 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.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):
  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
 def has_pending_requests(self):
  return len(self) > 0

這個檔案重寫了scheduler類，用來代替scrapy.core.scheduler的原有排程器。其實對原有排程器的邏輯沒有很大的改變，主要是使用了redis作為資料儲存的媒介，以達到各個爬蟲之間的統一排程。

scheduler負責排程各個spider的request請求，scheduler初始化時，通過settings檔案讀取queue和dupefilters的型別（一般就用上邊預設的），配置queue和dupefilters使用的key（一般就是spider name加上queue或者dupefilters，這樣對於同一種spider的不同例項，就會使用相同的資料塊了）。每當一個request要被排程時，enqueue_request被呼叫，scheduler使用dupefilters來判斷這個url是否重複，如果不重複，就新增到queue的容器中（先進先出，先進後出和優先順序都可以，可以在settings中配置）。當排程完成時，next_request被呼叫，scheduler就通過queue容器的介面，取出一個request，把他傳送給相應的spider，讓spider進行爬取工作。

spider.py

設計的這個spider從redis中讀取要爬的url，然後執行爬取，若爬取過程中返回更多的url，那麼繼續進行直至所有的request完成。之後繼續從redis中讀取url，迴圈這個過程。

from scrapy import signals
from scrapy.exceptions import DontCloseSpider
from scrapy.spiders import Spider, CrawlSpider
from . import connection
class RedisMixin(object):
 """Mixin class to implement reading urls from a redis queue."""
 redis_key = None # If empty, uses default '<spider>:start_urls'.
 # Fetch this amount of start urls when idle.
 redis_batch_size = 100
 # Redis client instance.
 server = None
 def start_requests(self):
  """Returns a batch of start requests from redis."""
  return self.next_requests()
 def setup_redis(self, crawler=None):
  """Setup redis connection and idle signal.
  This should be called after the spider has set its crawler object.
  """
  if self.server is not None:
   return
  if crawler is None:
   # We allow optional crawler argument to keep backwrads
   # compatibility.
   # XXX: Raise a deprecation warning.
   assert self.crawler, "crawler not set"
   crawler = self.crawler
  if not self.redis_key:
   self.redis_key = '%s:start_urls' % self.name
  self.log("Reading URLs from redis key '%s'" % self.redis_key)
  self.redis_batch_size = self.settings.getint(
   'REDIS_START_URLS_BATCH_SIZE',
   self.redis_batch_size,
  )
  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 next_requests(self):
  """Returns a request to be scheduled or none."""
  use_set = self.settings.getbool('REDIS_START_URLS_AS_SET')
  fetch_one = self.server.spop if use_set else self.server.lpop
  # XXX: Do we need to use a timeout here?
  found = 0
  while found < self.redis_batch_size:
   data = fetch_one(self.redis_key)
   if not data:
    # Queue empty.
    break
   yield self.make_request_from_data(data)
   found += 1
  if found:
   self.logger.debug("Read %s requests from '%s'", found, self.redis_key)
 def make_request_from_data(self, data):
  # By default, data is an URL.
  if '://' in data:
   return self.make_requests_from_url(data)
  else:
   self.logger.error("Unexpected URL from '%s': %r", self.redis_key, data)
 def schedule_next_requests(self):
  """Schedules a request if available"""
  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
class RedisSpider(RedisMixin, Spider):
 """Spider that reads urls from redis queue when idle."""
 @classmethod
 def from_crawler(self, crawler):
  obj = super(RedisSpider, self).from_crawler(crawler)
  obj.setup_redis(crawler)
  return obj
class RedisCrawlSpider(RedisMixin, CrawlSpider):
 """Spider that reads urls from redis queue when idle."""
 @classmethod
 def from_crawler(self, crawler):
  obj = super(RedisCrawlSpider, self).from_crawler(crawler)
  obj.setup_redis(crawler)
  return obj

spider的改動也不是很大，主要是通過connect介面，給spider綁定了spider_idle訊號，spider初始化時，通過setup_redis函式初始化好和redis的連線，之後通過next_requests函式從redis中取出strat url，使用的key是settings中

REDIS_START_URLS_AS_SET定義的（注意了這裡的初始化url池和我們上邊的queue的url池不是一個東西，queue的池是用於排程的，初始化url池是存放入口url的，他們都存在redis中，但是使用不同的key來區分，就當成是不同的表吧），spider使用少量的start url，可以發展出很多新的url，這些url會進入scheduler進行判重和排程。直到spider跑到排程池內沒有url的時候，會觸發spider_idle訊號，從而觸發spider的next_requests函式，再次從redis的start url池中讀取一些url。

元件之間的關係

最後總結一下scrapy-redis的總體思路：這個工程通過重寫scheduler和spider類，實現了排程、spider啟動和redis的互動。實現新的dupefilter和queue類，達到了判重和排程容器和redis的互動，因為每個主機上的爬蟲程序都訪問同一個redis資料庫，所以排程和判重都統一進行統一管理，達到了分散式爬蟲的目的。

當spider被初始化時，同時會初始化一個對應的scheduler物件，這個排程器物件通過讀取settings，配置好自己的排程容器queue和判重工具dupefilter。每當一個spider產出一個request的時候，scrapy核心會把這個reuqest遞交給這個spider對應的scheduler物件進行排程，scheduler物件通過訪問redis對request進行判重，如果不重複就把他新增進redis中的排程池。當排程條件滿足時，scheduler物件就從redis的排程池中取出一個request傳送給spider，讓他爬取。當spider爬取的所有暫時可用url之後，scheduler發現這個spider對應的redis的排程池空了，於是觸發訊號spider_idle，spider收到這個訊號之後，直接連線redis讀取strart url池，拿去新的一批url入口，然後再次重複上邊的工作。

為什麼要提供這些元件？

我們先從scrapy的“待爬佇列”和“Scheduler”入手：玩過爬蟲的同學都多多少少有些瞭解，在爬蟲爬取過程當中有一個主要的資料結構是“待爬佇列”，以及能夠操作這個佇列的排程器（也就是Scheduler）。scrapy官方文件對這二者的描述不多，基本上沒提。

scrapy使用什麼樣的資料結構來存放待爬取的request呢？其實沒用高大上的資料結構，就是python自帶的collection.deque（改造過後的），問題來了，該怎麼讓兩個以上的Spider共用這個deque呢？

scrapy-redis提供了一個解決方法，把deque換成redis資料庫，我們從同一個redis伺服器存放要爬取的request，這樣就能讓多個spider去同一個資料庫裡讀取，這樣分散式的主要問題就解決了嘛。

那麼問題又來了，我們換了redis來存放佇列，哪scrapy就能直接分散式了麼？。scrapy中跟“待爬佇列”直接相關的就是排程器“Scheduler”，它負責對新的request進行入列操作（加入deque），取出下一個要爬取的request（從deque中取出）等操作。在scrapy中，Scheduler並不是直接就把deque拿來就粗暴的使用了，而且提供了一個比較高階的組織方法，它把待爬佇列按照優先順序建立了一個字典結構，比如：

{
 priority0:佇列0
 priority1:佇列2
 priority2:佇列2
}

然後根據request中的priority屬性，來決定該入哪個佇列。而出列時，則按priority較小的優先出列。為了管理這個比較高階的佇列字典，Scheduler需要提供一系列的方法。你要是換了redis做佇列，這個scrapy下的Scheduler就用不了，所以自己寫一個吧。於是就出現了scrapy-redis的專用scheduler。

那麼既然使用了redis做主要資料結構，能不能把其他使用自帶資料結構關鍵功能模組也換掉呢？ 在我們爬取過程當中，還有一個重要的功能模組，就是request去重。scrapy中是如何實現這個去重功能的呢？用集合~scrapy中把已經發送的request指紋放入到一個集合中，把下一個request的指紋拿到集合中比對，如果該指紋存在於集合中，說明這個request傳送過了，如果沒有則繼續操作。

為了分散式，把這個集合也換掉吧，換了redis，照樣也得把去重類給換了。於是就有了scrapy-redis的dupefilter。那麼依次類推，接下來的其他元件（Pipeline和Spider），我們也可以輕鬆的猜到，他們是為什麼要被修改呢。