scrapy分布式去重组件源码及其实现过程

简介: scrapy_redis在继承scrapy去重组件的基础上覆盖了某些方法,原scrapy去重是基于单机情况下的内部去重,但是分布式是多机条件下的多爬虫协同去重,因此需要让不同及其上的同一个爬虫能够在同一个地方进行去重,这就是Redis的集合。

scrapy_redis在继承scrapy去重组件的基础上覆盖了某些方法,原scrapy去重是基于单机情况下的内部去重,但是分布式是多机条件下的多爬虫协同去重,因此需要让不同及其上的同一个爬虫能够在同一个地方进行去重,这就是Redis的集合。

先看看scrapy_redis 去重组件dupefilter的源码:

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


from_settings、from_crawler方法不用解释,就是读取配置文件连接Redis设置key,关键在request_seen、request_fingerprint这两个方法;request_seen调用self.request_fingerprint进而调用from scrapy.utils.request import request_fingerprint生成request的指纹。

再看scrapy.utils.request 中的request_fingerprint源码:

"""
This module provides some useful functions for working with
scrapy.http.Request objects
"""


from __future__ import print_function
import hashlib
import weakref
from six.moves.urllib.parse import urlunparse

from w3lib.http import basic_auth_header
from scrapy.utils.python import to_bytes, to_native_str

from w3lib.url import canonicalize_url
from scrapy.utils.httpobj import urlparse_cached


_fingerprint_cache = weakref.WeakKeyDictionary()
def request_fingerprint(request, include_headers=None):
    """
    Return the request fingerprint.
    The request fingerprint is a hash that uniquely identifies the resource the
    request points to. For example, take the following two urls:
    http://www.example.com/query?id=111&cat=222
    http://www.example.com/query?cat=222&id=111
    Even though those are two different URLs both point to the same resource
    and are equivalent (ie. they should return the same response).
    Another example are cookies used to store session ids. Suppose the
    following page is only accesible to authenticated users:
    http://www.example.com/members/offers.html
    Lot of sites use a cookie to store the session id, which adds a random
    component to the HTTP Request and thus should be ignored when calculating
    the fingerprint.
    For this reason, request headers are ignored by default when calculating
    the fingeprint. If you want to include specific headers use the
    include_headers argument, which is a list of Request headers to include.
    """

    if include_headers:
        include_headers = tuple(to_bytes(h.lower())
                                 for h in sorted(include_headers))
    cache = _fingerprint_cache.setdefault(request, {})
    if include_headers not in cache:
        fp = hashlib.sha1()
        fp.update(to_bytes(request.method))
        fp.update(to_bytes(canonicalize_url(request.url)))
        fp.update(request.body or b'')
        if include_headers:
            for hdr in include_headers:
                if hdr in request.headers:
                    fp.update(hdr)
                    for v in request.headers.getlist(hdr):
                        fp.update(v)
        cache[include_headers] = fp.hexdigest()
    return cache[include_headers]


def request_authenticate(request, username, password):
    """Autenticate the given request (in place) using the HTTP basic access
    authentication mechanism (RFC 2617) and the given username and password
    """

    request.headers['Authorization'] = basic_auth_header(username, password)


def request_httprepr(request):
    """Return the raw HTTP representation (as bytes) of the given request.
    This is provided only for reference since it's not the actual stream of
    bytes that will be send when performing the request (that's controlled
    by Twisted).
    """

    parsed = urlparse_cached(request)
    path = urlunparse(('''', parsed.path or '/', parsed.params, parsed.query, ''))
    s = to_bytes(request.method) + b" " + to_bytes(path) + b" HTTP/1.1\r\n"
    s += b"Host: " + to_bytes(parsed.hostname or b'') + b"\r\n"
    if request.headers:
        s += request.headers.to_string() + b"\r\n"
    s += b"\r\n"
    s += request.body
    return s


def referer_str(request):
    """ Return Referer HTTP header suitable for logging. """
    referrer = request.headers.get('Referer')
    if referrer is None:
        return referrer
    return to_native_str(referrer, errors='replace')


request_fingerprint用于返回唯一指纹,并且对于携带参数顺序不同的URL返回相同的指纹,通过哈希计算实现,参与哈希计算的有request.method、request.url、request.body,同时提供了一个备选列表参数,该列表存放需要加入计算的request.headers中的字段信息,也就是cookie也可以参与指纹计算,针对同一个URL但是header中的字段不同那么也可以视为不同的请求,从而不被去重。

当然获得一个请求的指纹之后,便可与Redis中的去重列表比对,如果存在,那么抛弃,如果不存在那么将指纹加入去重集合,同时通知调度器该request可以进入队列,这便是去重的实现过程。


相关文章
|
数据采集 存储 数据可视化
分布式爬虫框架Scrapy-Redis实战指南
本文介绍如何使用Scrapy-Redis构建分布式爬虫系统,采集携程平台上热门城市的酒店价格与评价信息。通过代理IP、Cookie和User-Agent设置规避反爬策略,实现高效数据抓取。结合价格动态趋势分析,助力酒店业优化市场策略、提升服务质量。技术架构涵盖Scrapy-Redis核心调度、代理中间件及数据解析存储,提供完整的技术路线图与代码示例。
2017 0
分布式爬虫框架Scrapy-Redis实战指南
|
人工智能 安全 Java
智慧工地源码,Java语言开发,微服务架构,支持分布式和集群部署,多端覆盖
智慧工地是“互联网+建筑工地”的创新模式,基于物联网、移动互联网、BIM、大数据、人工智能等技术,实现对施工现场人员、设备、材料、安全等环节的智能化管理。其解决方案涵盖数据大屏、移动APP和PC管理端,采用高性能Java微服务架构,支持分布式与集群部署,结合Redis、消息队列等技术确保系统稳定高效。通过大数据驱动决策、物联网实时监测预警及AI智能视频监控,消除数据孤岛,提升项目可控性与安全性。智慧工地提供专家级远程管理服务,助力施工质量和安全管理升级,同时依托可扩展平台、多端应用和丰富设备接口,满足多样化需求,推动建筑行业数字化转型。
485 5
|
数据采集 存储 NoSQL
基于Scrapy-Redis的分布式景点数据爬取与热力图生成
基于Scrapy-Redis的分布式景点数据爬取与热力图生成
1128 67
|
安全
【📕分布式锁通关指南 07】源码剖析redisson利用看门狗机制异步维持客户端锁
Redisson 的看门狗机制是解决分布式锁续期问题的核心功能。当通过 `lock()` 方法加锁且未指定租约时间时,默认启用 30 秒的看门狗超时时间。其原理是在获取锁后创建一个定时任务,每隔 1/3 超时时间(默认 10 秒)通过 Lua 脚本检查锁状态并延长过期时间。续期操作异步执行,确保业务线程不被阻塞,同时仅当前持有锁的线程可成功续期。锁释放时自动清理看门狗任务,避免资源浪费。学习源码后需注意:避免使用带超时参数的加锁方法、控制业务执行时间、及时释放锁以优化性能。相比手动循环续期,Redisson 的定时任务方式更高效且安全。
1286 24
【📕分布式锁通关指南 07】源码剖析redisson利用看门狗机制异步维持客户端锁
|
NoSQL 安全 调度
【📕分布式锁通关指南 10】源码剖析redisson之MultiLock的实现
Redisson 的 MultiLock 是一种分布式锁实现,支持对多个独立的 RLock 同时加锁或解锁。它通过“整锁整放”机制确保所有锁要么全部加锁成功,要么完全回滚,避免状态不一致。适用于跨多个 Redis 实例或节点的场景,如分布式任务调度。其核心逻辑基于遍历加锁列表,失败时自动释放已获取的锁,保证原子性。解锁时亦逐一操作,降低死锁风险。MultiLock 不依赖 Lua 脚本,而是封装多锁协调,满足高一致性需求的业务场景。
588 0
【📕分布式锁通关指南 10】源码剖析redisson之MultiLock的实现
【📕分布式锁通关指南 08】源码剖析redisson可重入锁之释放及阻塞与非阻塞获取
本文深入剖析了Redisson中可重入锁的释放锁Lua脚本实现及其获取锁的两种方式(阻塞与非阻塞)。释放锁流程包括前置检查、重入计数处理、锁删除及消息发布等步骤。非阻塞获取锁(tryLock)通过有限时间等待返回布尔值,适合需快速反馈的场景;阻塞获取锁(lock)则无限等待直至成功,适用于必须获取锁的场景。两者在等待策略、返回值和中断处理上存在显著差异。本文为理解分布式锁实现提供了详实参考。
653 11
【📕分布式锁通关指南 08】源码剖析redisson可重入锁之释放及阻塞与非阻塞获取
|
NoSQL Java Redis
分布式锁—6.Redisson的同步器组件
Redisson提供了多种分布式同步工具,包括分布式锁、Semaphore和CountDownLatch。分布式锁包括可重入锁、公平锁、联锁、红锁和读写锁,适用于不同的并发控制场景。Semaphore允许多个线程同时获取锁,适用于资源池管理。CountDownLatch则用于线程间的同步,确保一组线程完成操作后再继续执行。Redisson通过Redis实现这些同步机制,提供了高可用性和高性能的分布式同步解决方案。源码剖析部分详细介绍了这些组件的初始化和操作流程,展示了Redisson如何利用Redis命令和
|
存储 安全 NoSQL
【📕分布式锁通关指南 09】源码剖析redisson之公平锁的实现
本文深入解析了 Redisson 中公平锁的实现原理。公平锁通过确保线程按请求顺序获取锁,避免“插队”现象。在 Redisson 中,`RedissonFairLock` 类的核心逻辑包含加锁与解锁两部分:加锁时,线程先尝试直接获取锁,失败则将自身信息加入 ZSet 等待队列,只有队首线程才能获取锁;解锁时,验证持有者身份并减少重入计数,最终删除锁或通知等待线程。其“公平性”源于 Lua 脚本的原子性操作:线程按时间戳排队、仅队首可尝试加锁、实时发布锁释放通知。这些设计确保了分布式环境下的线程安全与有序执行。
625 0
【📕分布式锁通关指南 09】源码剖析redisson之公平锁的实现
|
NoSQL Java Redis
【📕分布式锁通关指南 06】源码剖析redisson可重入锁之加锁
本文详细解析了Redisson可重入锁的加锁流程。首先从`RLock.lock()`方法入手,通过获取当前线程ID并调用`tryAcquire`尝试加锁。若加锁失败,则订阅锁释放通知并循环重试。核心逻辑由Lua脚本实现:检查锁是否存在,若不存在则创建并设置重入次数为1;若存在且为当前线程持有,则重入次数+1。否则返回锁的剩余过期时间。此过程展示了Redisson高效、可靠的分布式锁机制。
541 0
【📕分布式锁通关指南 06】源码剖析redisson可重入锁之加锁
|
SpringCloudAlibaba JavaScript 前端开发
谷粒商城笔记+踩坑(2)——分布式组件、前端基础,nacos+feign+gateway+ES6+vue脚手架
分布式组件、nacos注册配置中心、openfegin远程调用、网关gateway、ES6脚本语言规范、vue、elementUI
谷粒商城笔记+踩坑(2)——分布式组件、前端基础,nacos+feign+gateway+ES6+vue脚手架