小白進階之Scrapy第六篇Scrapy-Redis詳解（轉）

Scrapy-Redis 詳解

通常我們在一個站站點進行采集的時候，如果是小站的話 我們使用scrapy本身就可以滿足。

但是如果在面對一些比較大型的站點的時候，單個scrapy就顯得力不從心了。

要是我們能夠多個Scrapy一起采集該多好啊 人多力量大。

很遺憾Scrapy官方並不支持多個同時采集一個站點，雖然官方給出一個方法：

**將一個站點的分割成幾部分 交給不同的scrapy去采集**

似乎是個解決辦法，但是很麻煩誒！畢竟分割很麻煩的哇

下面就改輪到我們的額主角Scrapy-Redis登場了！

什么？？你這么就登場了？還沒說為什么呢？

好吧 為了簡單起見 就用官方圖來簡單說明一下：

這張圖大家相信大家都很熟悉了。重點看一下SCHEDULER

1. 先來看看官方對於SCHEDULER的定義：

**SCHEDULER接受來自Engine的Requests,並將它們放入隊列（可以按順序優先級），以便在之后將其提供給Engine**

點我看文檔

2. 現在我們來看看SCHEDULER都提供了些什么功能：

根據官方文檔說明 在我們沒有沒有指定 SCHEDULER 參數時，默認使用：’scrapy.core.scheduler.Scheduler’ 作為SCHEDULER(調度器)

scrapy.core.scheduler.py:

Python

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

 

只挑了一些重點的寫了一些注釋剩下大家自己領會(才不是我懶哦 )

從上面的代碼 我們可以很清楚的知道 SCHEDULER的主要是完成了 push Request pop Request 和 去重的操作。

而且queue 操作是在內存隊列中完成的。

大家看queuelib.queue就會發現基於內存的（deque）

那么去重呢？

scrapy.dupefilters.py

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)

 

按照正常流程就是大家都會進行重復的采集；我們都知道進程之間內存中的數據不可共享的，那么你在開啟多個Scrapy的時候，它們相互之間並不知道對方采集了些什么那些沒有沒采集。那就大家伙兒自己玩自己的了。完全沒沒有效率的提升啊！

怎么解決呢？

這就是我們Scrapy-Redis解決的問題了，不能協作不就是因為Request 和 去重這兩個 不能共享嗎？

那我把這兩個獨立出來好了。

將Scrapy中的SCHEDULER組件獨立放到大家都能訪問的地方不就OK啦！加上scrapy-redis后流程圖就應該變成這樣了?

So············· 這樣是不是看起來就清楚多了？？？

下面我們來看看Scrapy-Redis是怎么處理的?

scrapy_redis.scheduler.py：

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

 

來先來看看

 

以上就是Scrapy-Redis中的SCHEDULER模塊。下面我們來看看queue和本身的什么不同：

scrapy_redis.queue.py

以最常用的優先級隊列 PriorityQueue 舉例：

 

scrapy_redis.queue.py

 

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

 

以上就是SCHEDULER在處理Request的時候做的操作了。

是時候來看看SCHEDULER是怎么處理去重的了！

只需要注意這個?方法即可：

scrapy_redis.dupefilter.py

 

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

 

這樣大家就都可以訪問同一個Redis 獲取同一個spider的Request 在同一個位置去重，就不用擔心重復啦

大概就像這樣：

1. spider1：檢查一下這個Request是否在Redis去重，如果在就證明其它的spider采集過啦！如果不在就添加進調度隊列，等待別 人獲取。自己繼續干活抓取網頁 產生新的Request了 重復之前步驟。
2. spider2：以相同的邏輯執行

可能有些小伙兒會產生疑問了~~！spider2拿到了別人的Request了 怎么能正確的執行呢？邏輯不會錯嗎？

這個不用擔心啦 因為整Request當中包含了，所有的邏輯，回去看看上面那個序列化的字典。

總結一下：

1. 1. Scrapy-Reids 就是將Scrapy原本在內存中處理的 調度(就是一個隊列Queue)、去重、這兩個操作通過Redis來實現
2. 多個Scrapy在采集同一個站點時會使用相同的redis key（可以理解為隊列）添加Request 獲取Request 去重Request，這樣所有的spider不會進行重復采集。效率自然就嗖嗖的上去了。
3. 3. Redis是原子性的，好處不言而喻(一個Request要么被處理 要么沒被處理，不存在第三可能)

另外Scrapy-Redis本身不支持Redis-Cluster，大量網站去重的話會給單機很大的壓力（就算使用boolfilter 內存也不夠整啊！）

改造方式很簡單：

1.  使用 **rediscluster** 這個包替換掉本身的Redis連接
2. Redis-Cluster 不支持事務，可以使用lua腳本進行代替（lua腳本是原子性的哦）
3. **注意使用lua腳本 不能寫占用時間很長的操作**（畢竟一大群人等着操作Redis 你總不能讓人家等着吧）

以上！完畢

對於懶人小伙伴兒 看看這個我改好的: 集群版Scrapy-Redis **PS: 支持Python3.6+ 哦 ！ 其余的版本沒測試過**

 

轉載請注明：靜覓 » 小白進階之Scrapy第六篇Scrapy-Redis詳解