深入分析 Scrapy 下載器原理

今天我們來完整分析下 Scrapy 中下載器模塊的代碼，深入理解下載器的實現(xiàn)原理以及用到的 Twisted 相關模塊。本節(jié)的內(nèi)容會有些枯燥，請耐心閱讀下去。

1. Twisted 中的 Web Client 模塊

本小節(jié)內(nèi)容主要參考官方文檔對于 Web Client 模塊的介紹，也就是文獻1。這部分內(nèi)容正是 Scrapy 下載器的核心，為了能更好的理解下載器，我們需要先學習下 Twisted 中的這塊內(nèi)容。

1.1 發(fā)出請求

注意 twisted.web.client.Agent 這個類，它是客戶端 API 的入口點，請求是使用 request() 方法發(fā)出的，該方法以請求方法、請求URI、請求頭和可以生成請求體的對象作為參數(shù)。代理負責連接設置。因此，它需要一個 reactor 作為初始值設定項的參數(shù)。來看官方給的第一個簡單例子：

from __future__ import print_function

from twisted.internet import reactor
from twisted.web.client import Agent
from twisted.web.http_headers import Headers

agent = Agent(reactor)

d = agent.request(
    b'GET',
    b'http://idcbgp.cn/wiki/',
    Headers({'User-Agent': ['Twisted Web Client Example']}),
    None)

def cbResponse(ignored):
    print('Response received')
d.addCallback(cbResponse)

# 關閉reactor()
def cbShutdown(ignored):
    reactor.stop()
d.addBoth(cbShutdown)

reactor.run()

上述代碼簡單實例化一個 agent，然后調(diào)用 request() 方法請求 http://idcbgp.cn/wiki/ 這個地址，這個動作也是一個延遲加載的方式；接下來的回調(diào)鏈中還會有請求完成后打印收到響應的方法以及最后關閉 reactor 的方法；執(zhí)行的結果如下：

[root@server2 scrapy-test]# python3 request.py 
Response received

如果想要給請求帶上參數(shù)，就需要傳遞一個 twisted.web.iweb.IBodyProducer 類型的對象到 Agent.request。我們繼續(xù)來學習官方給出的第二個例子：

下面的代碼給出了一個簡單的 IBodyProducer 實現(xiàn)，它向使用者寫入內(nèi)存中的字符串

# 代碼文件命名為：bytesprod.py 

from zope.interface import implementer

from twisted.internet.defer import succeed
from twisted.web.iweb import IBodyProducer

@implementer(IBodyProducer)
class BytesProducer(object):
    def __init__(self, body):
        self.body = body
        self.length = len(body)

    def startProducing(self, consumer):
        consumer.write(self.body)
        return succeed(None)

    def pauseProducing(self):
        pass

    def stopProducing(self):
        pass

下面的代碼則在請求中帶上了 body 體：

# 代碼文件：sendbody.py

from twisted.internet import reactor
from twisted.web.client import Agent
from twisted.web.http_headers import Headers

from bytesprod import BytesProducer

agent = Agent(reactor)
# 構造請求體
body = BytesProducer(b"hello, world")
d = agent.request(
    b'POST',
    b'http://httpbin.org/post',
    Headers({'User-Agent': ['Twisted Web Client Example'],
             'Content-Type': ['text/x-greeting']}),
    # 帶上body
    body)

# 回調(diào)鏈，收到上個request的請求響應
def cbResponse(ignored):
    print('Response received')
d.addCallback(cbResponse)

# 關閉reactor
def cbShutdown(ignored):
    reactor.stop()
d.addBoth(cbShutdown)

reactor.run()

1.2 接收響應

接下來一個內(nèi)容就是關于數(shù)據(jù)的接收。前面的代碼都只有請求，沒有接收響應數(shù)據(jù)。如果 Agent.request 請求成功，則 Deferred 將觸發(fā)一個響應。一旦收到所有響應頭，就會發(fā)生這種情況。它發(fā)生在處理任何響應體 (如果有)之前。Response 對象有一個使響應體可用的方法：deliverBody，接下來我們給出一個使用實例：

from __future__ import print_function

from pprint import pformat

from twisted.internet import reactor
from twisted.internet.defer import Deferred
from twisted.internet.protocol import Protocol
from twisted.web.client import Agent
from twisted.web.http_headers import Headers

# 繼承Protocol
class BeginningPrinter(Protocol):
    def __init__(self, finished):
        self.finished = finished
        self.remaining = 1024 * 10

    def dataReceived(self, bytes):
        """
        響應的數(shù)據(jù)從該方法獲取，最終獲取的數(shù)據(jù)大小不超過self.remaining 
        """
        if self.remaining:
            display = bytes[:self.remaining]
            print('Some data received:')
            print(str(display, encoding='utf8'))
            self.remaining -= len(display)

    def connectionLost(self, reason):
        print('Finished receiving body:', reason.getErrorMessage())
        self.finished.callback(None)

# 獲取agent實例，傳入reactor        
agent = Agent(reactor)
# 請求慕課網(wǎng)wiki
d = agent.request(
    b'GET',
    b'http://idcbgp.cn/wiki/',
    Headers({'User-Agent': ['Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.106 Safari/537.36']}),
    None)

def cbRequest(response):
    finished = Deferred()
    # 獲取響應數(shù)據(jù)
    response.deliverBody(BeginningPrinter(finished))
    return finished
# 加入回調(diào)鏈
d.addCallback(cbRequest)

def cbShutdown(ignored):
    reactor.stop()
d.addBoth(cbShutdown)

reactor.run()

我們可以直接來看看這個：

其實仔細分析 Scrapy 下載器源碼我們可以在 scrapy/core/downloader/handlers/http11.py 的代碼中看到如下非常類似的代碼：

2. Scrapy 中下載器模塊源碼分析

2.1 下載器的__init__() 方法分析

我們直接看源碼目錄，關于下載器模塊的代碼全在 scrapy/core/downloader 目錄下：

其中主要的核心代碼就是這個 __init__.py 文件，其中定義了 Downloader 這個類，它就是我們的下載器。首先來看這個類的初始化過程：

# ...

class Downloader:

    DOWNLOAD_SLOT = 'download_slot'

    def __init__(self, crawler):
        # 獲取配置
        self.settings = crawler.settings
        self.signals = crawler.signals
        self.slots = {}
        self.active = set()
        # 獲取handlers
        self.handlers = DownloadHandlers(crawler)
        # 獲取配置中的請求并發(fā)數(shù)
        self.total_concurrency = self.settings.getint('CONCURRENT_REQUESTS')
        # 每個域的請求并發(fā)數(shù)
        self.domain_concurrency = self.settings.getint('CONCURRENT_REQUESTS_PER_DOMAIN')
        # ip并發(fā)請求數(shù)
        self.ip_concurrency = self.settings.getint('CONCURRENT_REQUESTS_PER_IP')
        # 是否設置隨機延遲
        self.randomize_delay = self.settings.getbool('RANDOMIZE_DOWNLOAD_DELAY')
        # 中間件管理器
        self.middleware = DownloaderMiddlewareManager.from_crawler(crawler)
        self._slot_gc_loop = task.LoopingCall(self._slot_gc)
        self._slot_gc_loop.start(60)

從這個初始化的過程中，我們要重點關注幾個屬性值：

• settings：對應的 scrapy 項目配置；

• handlers：實例化 DownloadHandlers， 而這個類則是網(wǎng)頁下載的最終處理類；

• 各種配置屬性：從 settings.py 中獲取控制并發(fā)的參數(shù)；

• middleware：下載中間件的管理器，在下一節(jié)中我們會詳細分析這個屬性；

2.2 下載器實例的 handlers 屬性值分析

重點先看看這個 handlers 屬性值，它是 DownloadHandlers 類的一個實例，對應的代碼位置為：scrapy/core/downloader/handlers/__init__.py 。其類定義如下：

# ...

class DownloadHandlers:

    def __init__(self, crawler):
        self._crawler = crawler
        self._schemes = {}  # stores acceptable schemes on instancing
        self._handlers = {}  # stores instanced handlers for schemes
        self._notconfigured = {}  # remembers failed handlers
        handlers = without_none_values(
            crawler.settings.getwithbase('DOWNLOAD_HANDLERS'))
        for scheme, clspath in handlers.items():
            self._schemes[scheme] = clspath
            self._load_handler(scheme, skip_lazy=True)

        crawler.signals.connect(self._close, signals.engine_stopped)

    def _get_handler(self, scheme):
        """Lazy-load the downloadhandler for a scheme
        only on the first request for that scheme.
        """
        if scheme in self._handlers:
            return self._handlers[scheme]
        if scheme in self._notconfigured:
            return None
        if scheme not in self._schemes:
            self._notconfigured[scheme] = 'no handler available for that scheme'
            return None

        return self._load_handler(scheme)

    def _load_handler(self, scheme, skip_lazy=False):
        path = self._schemes[scheme]
        try:
            dhcls = load_object(path)
            if skip_lazy and getattr(dhcls, 'lazy', True):
                return None
            dh = create_instance(
                objcls=dhcls,
                settings=self._crawler.settings,
                crawler=self._crawler,
            )
        except NotConfigured as ex:
            self._notconfigured[scheme] = str(ex)
            return None
        except Exception as ex:
            # ...
        else:
            self._handlers[scheme] = dh
            return dh

    def download_request(self, request, spider):
        scheme = urlparse_cached(request).scheme
        handler = self._get_handler(scheme)
        if not handler:
            raise NotSupported("Unsupported URL scheme '%s': %s" %
                               (scheme, self._notconfigured[scheme]))
        return handler.download_request(request, spider)

    @defer.inlineCallbacks
    def _close(self, *_a, **_kw):
        # ...

我們分別來解析這個類中定義的方法，都是非常重要的。此外，每個方法含義明確，而且代碼精煉，我們對此一一進行說明。

首先是 __init__() 方法，我們可以看到一個核心的語句：handlers = without_none_values(crawler.settings.getwithbase('DOWNLOAD_HANDLERS'))。其中 getwithbase() 方法的定義位于 scapy/settings/__init__.py 文件中，其內(nèi)容如下：

# 源碼位置：scrapy/settings/__init__.py
# ...

class BaseSettings(MutableMapping):
    # ...
    
    def getwithbase(self, name):
        """Get a composition of a dictionary-like setting and its `_BASE`
        counterpart.

        :param name: name of the dictionary-like setting
        :type name: string
        """
        compbs = BaseSettings()
        compbs.update(self[name + '_BASE'])
        compbs.update(self[name])
        return compbs
    
    # ...

從上面的代碼中，可以知道 crawler.settings.getwithbase('DOWNLOAD_HANDLERS') 語句其實是獲取了 settings.py 配置中的 DOWNLOAD_HANDLERS 和 DOWNLOAD_HANDLERS_BASE 值，并返回包含這兩個屬性值的 BaseSettings 實例。我們從默認的 scrapy/settings/default_settings.py 中可以看到這兩個屬性值如下：

# 源碼位置：scrapy/settings/default_settings.py

DOWNLOAD_HANDLERS = {}
DOWNLOAD_HANDLERS_BASE = {
    'data': 'scrapy.core.downloader.handlers.datauri.DataURIDownloadHandler',
    'file': 'scrapy.core.downloader.handlers.file.FileDownloadHandler',
    'http': 'scrapy.core.downloader.handlers.http.HTTPDownloadHandler',
    'https': 'scrapy.core.downloader.handlers.http.HTTPDownloadHandler',
    's3': 'scrapy.core.downloader.handlers.s3.S3DownloadHandler',
    'ftp': 'scrapy.core.downloader.handlers.ftp.FTPDownloadHandler',
}

通常我們項目中的下載請求一般是 http 或者 https，對應的都是 scrapy.core.downloader.handlers.http.HTTPDownloadHandler 這個位置的 handler，從這里也可以看到 scrapy 框架其實是支持非常多協(xié)議下載的，比如 s3 下載、文件下載以及 ftp下載等。緊接著的兩句就是在設置 self._schemes 和 self._handlers 的值。其中 self._handlers 會加載對應 Handler 類：

    def _load_handler(self, scheme, skip_lazy=False):
        # 獲取協(xié)議對應的handler類路徑，比如 ftp協(xié)議對應著的handler類為
        # scrapy.core.downloader.handlers.ftp.FTPDownloadHandler
        path = self._schemes[scheme]
        try:
            # 獲取相應的handler類,非字符串形式
            dhcls = load_object(path)
            # 默認__init__()方法調(diào)用時參數(shù)skip_lazy為True，我們需要檢查對應handler類中的lazy屬性值，沒有設置時默認為True；如果handler類中l(wèi)azy屬性設置為True或者不設置，則該handler不會加入到self._handlers中
            if skip_lazy and getattr(dhcls, 'lazy', True):
                return None
            # 創(chuàng)建相應的handler類實例，需要的參數(shù)為：
            dh = create_instance(
                objcls=dhcls,
                settings=self._crawler.settings,
                crawler=self._crawler,
            )
        except NotConfigured as ex:
            # ...
        except Exception as ex:
            # ...
        else:
            # 設置self._handlers
            self._handlers[scheme] = dh
            return dh

上面的注釋已經(jīng)非常清楚，在眾多 handlers 中，只有 S3DownloadHandler 類中沒有設置 lazy。所以在 __init__() 方法執(zhí)行完后，self._handlers 中不會有鍵 “s3” 及其對應的實例。

_get_handler() 就非常明顯了，由于我們在初始化方法 __init__() 中已經(jīng)得到了 self._handlers，此時該方法就是根據(jù)傳入的協(xié)議獲取相應 Handler 類的實例。 例如傳入的 scheme="http"，則返回就是 HTTPDownloadHandler 類的一個實例。

最后一個 download_request() 方法可以說是下載網(wǎng)頁的核心調(diào)用方法。我們來逐步分析該方法中的語句：

scheme = urlparse_cached(request).scheme
handler = self._get_handler(scheme)

上面兩句是獲取對應的下載請求的 Handler 實例，比較容易理解。接下來的 return 就是調(diào)用 handler 實例中的 download_request() 方法按照對應協(xié)議方式下載請求數(shù)據(jù)：

return handler.download_request(request, spider)

每個 handler 類都會有對應的 download_request() 方法。我們重點看看 http 和 https 協(xié)議對應的 handler 類：

# 源碼位置：scrapy/core/downloader/handlers/http.py
from scrapy.core.downloader.handlers.http10 import HTTP10DownloadHandler
from scrapy.core.downloader.handlers.http11 import (
    HTTP11DownloadHandler as HTTPDownloadHandler,
)

可以看到，這里的 HTTPDownloadHandler 類實際上是 http11.py 中的 HTTP11DownloadHandler 類。這里面內(nèi)容有點多，我們簡要地分析下：

# 源碼位置：scrapy/core/downloader/handlers/http11.py
# ...

class HTTP11DownloadHandler:
    # ...
    
    def download_request(self, request, spider):
        """Return a deferred for the HTTP download"""
        agent = ScrapyAgent(
            contextFactory=self._contextFactory,
            pool=self._pool,
            maxsize=getattr(spider, 'download_maxsize', self._default_maxsize),
            warnsize=getattr(spider, 'download_warnsize', self._default_warnsize),
            fail_on_dataloss=self._fail_on_dataloss,
            crawler=self._crawler,
        )
        return agent.download_request(request)
    
    # ...

是不是挺簡單的兩條語句：得到 agent，然后調(diào)用 agent.download_request() 方法得到請求的結果？我們繼續(xù)追蹤這個 ScrapyAgent 類：

# 源碼位置：scrapy/core/downloader/handlers/http11.py
# ...
from twisted.web.client import Agent, HTTPConnectionPool, ResponseDone, ResponseFailed, URI
# ...

class ScrapyAgent:

    _Agent = Agent
    # ...
    
    def _get_agent(self, request, timeout):
        from twisted.internet import reactor
        bindaddress = request.meta.get('bindaddress') or self._bindAddress
        # 從請求的meta參數(shù)中獲取proxy
        proxy = request.meta.get('proxy')
        if proxy:
            # 有代理的情況
            # ...
        
        # 沒有代理返回Agent的一個實例
        return self._Agent(
            reactor=reactor,
            contextFactory=self._contextFactory,
            connectTimeout=timeout,
            bindAddress=bindaddress,
            pool=self._pool,
        )
    
    def download_request(self, request):
        from twisted.internet import reactor
        # 從meta中獲取下載超時時間
        timeout = request.meta.get('download_timeout') or self._connectTimeout
        # 獲取agent
        agent = self._get_agent(request, timeout)

        # request details
        url = urldefrag(request.url)[0]
        method = to_bytes(request.method)
        headers = TxHeaders(request.headers)
        if isinstance(agent, self._TunnelingAgent):
            headers.removeHeader(b'Proxy-Authorization')
        if request.body:
            bodyproducer = _RequestBodyProducer(request.body)
        else:
            bodyproducer = None
        start_time = time()
        # 使用agent發(fā)起請求
        d = agent.request(method, to_bytes(url, encoding='ascii'), headers, bodyproducer)
        # set download latency
        d.addCallback(self._cb_latency, request, start_time)
        # response body is ready to be consumed
        d.addCallback(self._cb_bodyready, request)
        d.addCallback(self._cb_bodydone, request, url)
        # check download timeout
        self._timeout_cl = reactor.callLater(timeout, d.cancel)
        d.addBoth(self._cb_timeout, request, url, timeout)
        return d

看完上面的代碼就應該比較清楚了，我們先不考慮代理的相關代碼。直接看 _get_agent() 方法就是獲取 twisted 模塊中 Agent 的一個實例，然后通過這個 agent 去請求網(wǎng)頁 (在 download_request() 方法中完成 )，最后返回的仍然是一個 Deferred 對象。在 download_request() 中的代碼就和我們在第一小節(jié)中介紹的類似，這里便是 Scrapy 最后完成網(wǎng)頁抓取的地方，就是基于 Twisted 的 web client 部分的方法。其中，最核心的一行語句就是：

d = agent.request(method, to_bytes(url, encoding='ascii'), headers, bodyproducer)

這個分析已經(jīng)走到了 Scrapy 下載器的最后一層，在往下就是研究 Twisted 框架的源碼了。同學們有興趣的話，可以繼續(xù)追蹤下去，對本節(jié)課而言，追蹤到此已經(jīng)結束了。我們跳出 handlers，繼續(xù)研究 Downloader 類。

2.3 下載器的 _download() 分析

我們回到 Downloader 類上繼續(xù)學習，該下載器類中最核心的有如下三個方法：

• _enqueue_request()：請求入隊；
• _process_queue()：處理隊列中的請求；
• _download()：下載網(wǎng)頁；

從代碼中很明顯可以看到，三個函數(shù)的關系如下：

Downloader中三個核心函數(shù)關系

我們來重點看看這個 _download() 方法：

_download()方法分析

看圖中注釋部分，_download() 方法先創(chuàng)建一個下載的 deferred，注意這里的方法正是 self.handlers 的 download_request() 方法，這是網(wǎng)頁下載的主要語句。 接下來又使用 deferred 的 addCallback() 方法添加一個回調(diào)函數(shù)：_downloaded()。很明顯，_downloaded() 就是下載完成后調(diào)用的方法，其中 response 就是下載的結果，也就是后續(xù)會返回給 spider 中的 parse() 方法的那個。我們可以簡單做個實驗，看看是不是真的會在這里打印出響應的結果：

創(chuàng)建一個名為 test_downloader 的 scrapy 的項目：

[root@server2 scrapy-test]# scrapy startproject test_downloader

生成一個名為 downloader 的 spider：

# 進入到spider目錄
[root@server2 scrapy-test]# cd test_downloader/test_downloader/spiders/
# 新建一個spider文件
[root@server2 spiders]# scrapy genspider downloader idcbgp.cn/wiki/
[root@server2 spiders]# cat downloader.py 
import scrapy


class DownloaderSpider(scrapy.Spider):
    name = 'downloader'
    allowed_domains = ['idcbgp.cn/wiki/']
    start_urls = ['http://idcbgp.cn/wiki/']

    def parse(self, response):
        pass

我們添加幾個配置，將 scrapy 的日志打到文件中，避免影響我們打印一些結果：

# test_download/settings.py
# ...

#是否啟動日志記錄，默認True
LOG_ENABLED = True 
LOG_ENCODING = 'UTF-8'
#日志輸出文件，如果為NONE，就打印到控制臺
LOG_FILE = 'downloader.log'
#日志級別，默認DEBUG
LOG_LEVEL = 'INFO'
# 日志日期格式 
LOG_DATEFORMAT = "%Y-%m-%d %H:%M:%S"
#日志標準輸出，默認False，如果True所有標準輸出都將寫入日志中，比如代碼中的print輸出也會被寫入到
LOG_STDOUT = False

最重要的步驟來啦，我們在 scrapy 的源碼 scrapy/core/downloader/__init__.py 的中添加一些代碼，用于查看下載器獲取的結果：

我們來對添加的這部分代碼進行下說明：

# ...

class Downloader:
    # ...
    
    def _download(self, slot, request, spider):
        print('下載請求:{}, {}'.format(request.url, spider.name))
        # The order is very important for the following deferreds. Do not change!

        # 1. Create the download deferred
        dfd = mustbe_deferred(self.handlers.download_request, request, spider)

        # 2. Notify response_downloaded listeners about the recent download
        # before querying queue for next request
        def _downloaded(response):
            self.signals.send_catch_log(signal=signals.response_downloaded,
                                        response=response,
                                        request=request,
                                        spider=spider)
            ###############################新增代碼########################################
            print('__downloaded()中 response 結果類型:{}'.format(type(response)))
            import gzip
            from io import BytesIO
            from scrapy.http.response.text import TextResponse
            if isinstance(response, TextResponse):
                text = response.text
            else:
                # 解壓縮文本，這部分會在后續(xù)的下載中間件中被處理，傳給parse()方法時會變成解壓后的數(shù)據(jù)
                f = gzip.GzipFile(fileobj = BytesIO(response.body))
                text = f.read().decode('utf-8')
            print('得到結果:{}'.format(text[:3000]))
            ############################################################################
            
            return response

但就我們新建的項目而言，只是簡單的爬取慕課網(wǎng)的 wiki 頁面，獲取相應的頁面數(shù)據(jù)。由于我們沒有禁止 robot 協(xié)議，所以項目第一次會爬取 /robots.txt 地址，檢查 wiki 頁面是否允許爬??；接下來才會爬取 wiki 頁面。測試發(fā)現(xiàn)，第一次請求 /robots.txt 地址時，在 _downloaded() 中得到的結果直接就是 TextResponse 實例，我們可以用 response.text 方式直接拿到結果；但是第二次請求 http://idcbgp.cn/wiki/ 時，返回的結果是經(jīng)過壓縮的，其結果的前三個字節(jié)碼為：b'\x1f\x8b\x08' 開頭的 ，說明它是 gzip 壓縮過的數(shù)據(jù)。為了能查看相應的數(shù)據(jù)，我們可以在這里解碼查看，對應的就是上面的 else 部分代碼。我們現(xiàn)在來進行演示：

3. 小結

本小節(jié)中我們詳盡的剖析了 Scrapy 的下載器模塊，找出了最終請求網(wǎng)頁數(shù)據(jù)的方式。整個下載器的代碼不算特別復雜，主要是對 Twisted 的 web client 模塊中的類和方法的進一步封裝，我們可以通過參考文獻1來掌握該模塊的學習。當掌握了 Twisted 的這些基礎后，在來看 Scrapy 的代碼就會一目了然。下一小節(jié)我們將繼續(xù)剖析 Scrapy 的中間件模塊，進一步學習 Scrapy 框架源碼。

4. 參考文獻

[1] Using the Twisted Web Client