天天看點

scrapy-redis分布式爬蟲原理分析

scrapy是python的一個非常好用的爬蟲庫,功能非常強大,但是當我們要爬取的頁面非常多的時候,單個主機的處理能力就不能滿足我們的需求了(無論是處理速度還是網絡請求的并發數),這時候分布式爬蟲的優勢就顯現出來,人多力量大。而scrapy-redis就是結合了分布式資料庫redis,重寫了scrapy一些比較關鍵的代碼,将scrapy變成一個可以在多個主機上同時運作的分布式爬蟲。

scrapy-redis是github上的一個開源項目,可以直接下載下傳到他的源代碼:

https://github.com/rolando/scrapy-redis

scrapy-redis的官方文檔寫的比較簡潔,沒有提及其運作原理,是以如果想全面的了解分布式爬蟲的運作原理,還是得看scrapy-redis的源代碼才行(還得先了解scrapy的運作原理,不然看scrapy-redis還是比較費勁),不過scrapy-redis的源代碼很少,也比較好懂,很快就能看完。

scrapy-redis工程的主體還是是redis和scrapy兩個庫,工程本身實作的東西不是很多,這個工程就像膠水一樣,把這兩個插件粘結了起來。下面我們來看看,scrapy-redis的每一個源代碼檔案都實作了什麼功能,最後如何實作分布式的爬蟲系統:

connect.py

import redis
import six

from scrapy.utils.misc import load_object


DEFAULT_REDIS_CLS = redis.StrictRedis


# Sane connection defaults.
DEFAULT_PARAMS = {
    'socket_timeout': ,
    'socket_connect_timeout': ,
    '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檔案中配置套接字的逾時時間、等待時間等。

dupefilters.py

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 == 

    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在進入排程之前都要進行判重,如果重複就不需要參加排程,直接舍棄就好了,不然就是白白浪費資源。

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=-)
           

這裡實作了loads和dumps兩個函數,其實就是實作了一個serializer,因為redis資料庫不能存儲複雜對象(value部分隻能是字元串,字元串清單,字元串集合和hash,key部分隻能是字元串),是以我們存啥都要先串行化成文本才行。這裡使用的就是python的pickle子產品,一個相容py2和py3的串行化工具。這個serializer主要用于一會的scheduler存reuqest對象,至于為什麼不實用json格式,我也不是很懂,item pipeline的串行化預設用的就是json。

pipeline.py

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存起來,主要是為了友善我們延後處理資料。

queue.py

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=):
        """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=):
        """Pop a request"""
        if timeout > :
            data = self.server.brpop(self.key, timeout)
            if isinstance(data, tuple):
                data = data[]
        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=):
        """
        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, , ).zremrangebyrank(self.key, , )
        results, count = pipe.execute()
        if results:
            return self._decode_request(results[])


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=):
        """Pop a request"""
        if timeout > :
            data = self.server.blpop(self.key, timeout)
            if isinstance(data, tuple):
                data = data[]
        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排程池,實作了分布式爬蟲之間的統一排程。

scheduler.py

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=,
                 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 < :
            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) > 
           

這個檔案重寫了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進行爬取工作。

同時我們可以看到,如果setting檔案中配置了SCHEDULER_PERSIST為True,那麼在爬蟲關閉的時候scheduler會調用自己的flush函數把redis資料庫中的判重和排程池全部清空,使得我們的爬取進度完全丢失(但是item沒有丢失,item資料在另一個鍵中儲存)。如果設定SCHEDULER_PERSIST為False,爬蟲關閉後,判重池和排程池仍然存在于redis資料庫中,則我們再次開啟爬蟲時,可以接着上一次的進度繼續爬取。

spider.py

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 = 
    # 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 = 
        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 += 

        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入口,然後再次重複上邊的工作。