鐧界櫆椋庢�庝箞娌荤枟 http://www.baidianfeng51.cn/
Crawler.crawl-Spider.start_requests-Engine.open_spider-CallLaterOnce.scheduler-_next_request-SpiderMiddlewareManager.process_start_requests-Scheduler.open-初始化优先级队列-初始化基于磁盘的队列-调用指纹过滤器的open方法-Scraper.open_spider-调用所有的pipeline的open_spider芳芳-Crawler.open_spider-Signals.send_cache_log_deferred-Engine._next_request-Engine._needs_backout-Engine._next_request_from_scheduler-Engine.crawl-Scheduler.enqueue_request请求入队-RFPDupeFilter.request_seen指纹过滤-Engine._download-Downloader.fetch-DownloaderMiddlewareManager.download-Downloader._enqueue_request-Downloader._download-DownloadHandlers.download_request-调用在settings中定义的DOWNLOAD_HANDLERS_BASE和DOWNLOAD_HANDLERS下载器进行下载-ExecutionEngine._handle_downloader_output#处理下载结果-Scraper.enqueue_scrape-Scraper._scrape_next-Scraper._scrape-Scraper._scrape2-Scraper.call_spider-Spider.parse或者指定的回调方法-Scraper.handle_spider_output-Scraper._process_spidermw_output-ItemPipelineManager.process_item-调用定义在settings中定义的ITEM_PIPELINES_BASE和ITEM_PIPELINES中的方法-Scraper._itemproc_finished处理异常和继续处理-Engine.start
流转过程配合着官方的数据流转图查看比较好。
在核心组件初始化里面交代了Crawler.crawl方法中调用了spider.start_requests方法来获取种子链接。
classCrawler:
defer.inlineCallbacksdefcrawl(self,*args,**kwargs):ifself.crawling:raiseRuntimeError("Crawlingalreadytakingplace")self.crawling=Truetry:#从Spiderloader找到相关的类，并实例化self.spider=self._create_spider(*args,**kwargs)#创建引擎self.engine=self._create_engine()#调用Spider类的start_requests方法start_requests=iter(self.spider.start_requests())#执行引擎的open_spider，传入实例yieldself.engine.open_spider(self.spider,start_requests)yielddefer.maybeDeferred(self.engine.start)exceptException:self.crawling=Falseifself.engineisnotNone:yieldself.engine.close()raise继续看Spider.start_requests
classSpider(object_ref):defstart_requests(self):cls=self.__class__ifnotself.start_urlsandhasattr(self,start_url):raiseAttributeError("Crawlingcouldnotstart:start_urlsnotfound""orempty(butfoundstart_urlattributeinstead,""didyoumissans?)")ifmethod_is_overridden(cls,Spider,make_requests_from_url):warnings.warn("Spider.make_requests_from_urlmethodisdeprecated;it""wontbecalledinfutureScrapyreleases.Please""overrideSpider.start_requestsmethodinstead"f"(see{cls.__module__}.{cls.__name__}).",)forurlinself.start_urls:yieldself.make_requests_from_url(url)else:forurlinself.start_urls:yieldRequest(url,dont_filter=True)defmake_requests_from_url(self,url):"""Thismethodisdeprecated."""warnings.warn("Spider.make_requests_from_urlmethodisdeprecated:""itwillberemovedandnotbecalledbythedefault""Spider.start_requestsmethodinfutureScrapyreleases.""PleaseoverrideSpider.start_requestsmethodinstead.")returnRequest(url,dont_filter=True)
Spider.start_requests最终是返回了一个Requests实例
Requests实例classRequest(object_ref):def__init__(self,url,callback=None,method=GET,headers=None,body=None,cookies=None,meta=None,encoding=utf-8,priority=0,dont_filter=False,errback=None,flags=None,cb_kwargs=None):#编码方式self._encoding=encoding#thisonehastobesetfirst#请求方法self.method=str(method).upper()#URLself._set_url(url)#请求体self._set_body(body)#优先级ifnotisinstance(priority,int):raiseTypeError(f"Requestprioritynotaninteger:{priority!r}")self.priority=priority#正常回调函数ifcallbackisnotNoneandnotcallable(callback):raiseTypeError(fcallbackmustbeacallable,got{type(callback).__name__})#异常回调函数iferrbackisnotNoneandnotcallable(errback):raiseTypeError(ferrbackmustbeacallable,got{type(errback).__name__})self.callback=callbackself.errback=errback#cookieheaderself.cookies=cookiesor{}self.headers=Headers(headersor{},encoding=encoding)#是否需要去重self.dont_filter=dont_filter#meta信息self._meta=dict(meta)ifmetaelseNoneself._cb_kwargs=dict(cb_kwargs)ifcb_kwargselseNoneself.flags=[]ifflagsisNoneelselist(flags)
Requests定义了Scrapy的请求实例。
Engine如何调度yieldself.engine.open_spider(self.spider,start_requests)
classExecutionEngine:
defer.inlineCallbacksdefopen_spider(self,spider,start_requests=(),close_if_idle=True):ifnotself.has_capacity():raiseRuntimeError(f"Nofreespiderslotwhenopening{spider.name!r}")logger.info("Spideropened",extra={spider:spider})#注册_next_request方法，循环调用nextcall=CallLaterOnce(self._next_request,spider)#初始化schedulerscheduler=self.scheduler_cls.from_crawler(self.crawler)#调用爬虫中间件，处理种子请求start_requests=yieldself.scraper.spidermw.process_start_requests(start_requests,spider)#封装slot对象slot=Slot(start_requests,close_if_idle,nextcall,scheduler)self.slot=slotself.spider=spider#调用scheduler.openyieldscheduler.open(spider)yieldself.scraper.open_spider(spider)self.crawler.stats.open_spider(spider)yieldself.signals.send_catch_log_deferred(signals.spider_opened,spider=spider)#发起调用slot.nextcall.schedule()slot.heartbeat.start(5)CallLaterOnceclassCallLaterOnce:"""Scheduleafunctiontobecalledinthenextreactorloop,butonlyifithasntbeenalreadyscheduledsincethelasttimeitran."""#在twisted的reactor中循环调度的一个方法def__init__(self,func,*a,**kw):self._func=funcself._a=aself._kw=kwself._call=Nonedefschedule(self,delay=0):#上次发起调度完成，才可以再次调度fromtwisted.internetimportreactorifself._callisNone:#将self注册到call_later中self._call=reactor.callLater(delay,self)defcancel(self):ifself._call:self._call.cancel()def__call__(self):#上面注册的是self，所以会执行__call__self._call=Nonereturnself._func(*self._a,**self._kw)
将上面注册的nextcall=CallLaterOnce(self._next_request,spider)在twisted的reactor中异步执行，只需要再次调用nextcall.schedule()方法就可以完成再次调用，这里调用的是Engine._next_request。
在这之后，调用了start_requests=yieldself.scraper.spidermw.process_start_requests(start_requests,spider)爬虫中间件的SpiderMiddlewareManager.process_start_requests方法。
调度器classScheduler:defopen(self,spider):self.spider=spider#实例化优先级队列self.mqs=self._mq()#如果定义类dqdir则实例化基于磁盘的队列self.dqs=self._dq()ifself.dqdirelseNone#调用指纹过滤器的open方法returnself.df.open()def_mq(self):"""Createanewpriorityqueueinstance,within-memorystorage"""returncreate_instance(self.pqclass,settings=None,crawler=self.crawler,downstream_queue_cls=self.mqclass,key=)def_dq(self):"""Createanewpriorityqueueinstance,withdiskstorage"""#实例化磁盘队列state=self._read_dqs_state(self.dqdir)q=create_instance(self.pqclass,settings=None,crawler=self.crawler,downstream_queue_cls=self.dqclass,key=self.dqdir,startprios=state)ifq:logger.info("Resumingcrawl(%(queuesize)drequestsscheduled)",{queuesize:len(q)},extra={spider:self.spider})returnqdef_read_dqs_state(self,dqdir):path=join(dqdir,active.json)ifnotexists(path):return()withopen(path)asf:returnjson.load(f)Scraper
classScraper:
defer.inlineCallbacksdefopen_spider(self,spider):"""Openthegivenspiderforscrapingandallocateresourcesforit"""#处理的最大并发量self.slot=Slot(self.crawler.settings.getint(SCRAPER_SLOT_MAX_ACTIVE_SIZE))#调用所有pipeline的open_spider方法yieldself.itemproc.open_spider(spider)Engie._next_request调度classExecutionEngine:def_next_request(self,spider):#此方法会循环调度slot=self.slotifnotslot:return#暂停ifself.paused:return#是否等待whilenotself._needs_backout(spider):#从scheduler中获取request#注意，第一次获取是没有的，会break出来#从而执行下面的逻辑ifnotself._next_request_from_scheduler(spider):break#如果start_requests有数据且不需要等待ifslot.start_requestsandnotself._needs_backout(spider):try:#获取下一个种子请求request=next(slot.start_requests)exceptStopIteration:slot.start_requests=NoneexceptException:slot.start_requests=Nonelogger.error(Errorwhileobtainingstartrequests,exc_info=True,extra={spider:spider})else:#调用crawl，把request放入scheduler的队列中self.crawl(request,spider)#空闲则关闭spiderifself.spider_is_idle(spider)andslot.close_if_idle:self._spider_idle(spider)def_needs_backout(self,spider):#是否等待，需要满足四个条件#1.Engine是否stop#2.slot是否close#3.downloader下载超过预设#4.scraper处理response超过预设slot=self.slotreturn(notself.runningorslot.closingorself.downloader.needs_backout()orself.scraper.slot.needs_backout())def_next_request_from_scheduler(self,spider):slot=self.slot#从scheduler拿出下一个requestrequest=slot.scheduler.next_request()ifnotrequest:return#下载d=self._download(request,spider)#注册成功、失败、出口回调方法d.addBoth(self._handle_downloader_output,request,spider)d.addErrback(lambdaf:logger.info(Errorwhilehandlingdownloaderoutput,exc_info=failure_to_exc_info(f),extra={spider:spider}))d.addBoth(lambda_:slot.remove_request(request))d.addErrback(lambdaf:logger.info(Errorwhileremovingrequestfromslot,exc_info=failure_to_exc_info(f),extra={spider:spider}))d.addBoth(lambda_:slot.nextcall.schedule())d.addErrback(lambdaf:logger.info(Errorwhileschedulingnewrequest,exc_info=failure_to_exc_info(f),extra={spider:spider}))returndEngine.crawl请求入队
classEngine:defcrawl(self,request,spider):ifspidernotinself.open_spiders:raiseRuntimeError(f"Spider{spider.name!r}notopenedwhencrawling:{request}")#request放入scheduler队列，调用nextcall的scheduleself.schedule(request,spider)#进行下一次调度self.slot.nextcall.schedule()defschedule(self,request,spider):self.signals.send_catch_log(signals.request_scheduled,request=request,spider=spider)#调用scheduler的enqueue_request，把request放入scheduler队列ifnotself.slot.scheduler.enqueue_request(request):self.signals.send_catch_log(signals.request_dropped,request=request,spider=spider)Scheduler.enqueue_request
classScheduler:defenqueue_request(self,request):#请求入队，如果不想验证，则request.dont_filter=True若请求过滤器验证重复，则返回falseifnotrequest.dont_filterandself.df.request_seen(request):self.df.log(request,self.spider)returnFalse#磁盘队列是否入队成功dqok=self._dqpush(request)ifdqok: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)returnTrue
指纹生成，没啥说的。
classRFPDupeFilter(BaseDupeFilter):defrequest_seen(self,request):#生成请求指纹fp=self.request_fingerprint(request)#请求指纹如果在集合中，则认为重复iffpinself.fingerprints:returnTrue#记录不重复的指纹self.fingerprints.add(fp)#如果有path则将指纹写入文件中ifself.file:self.file.write(fp+\n)defrequest_fingerprint(self,request):#调用utils.request的request_fingerprint方法returnrequest_fingerprint(request)
defrequest_fingerprint(request,include_headers=None,keep_fragments=False):"""生成请求指纹"""#指纹中是否包含headersifinclude_headers:include_headers=tuple(to_bytes(h.lower())forhinsorted(include_headers))cache=_fingerprint_cache.setdefault(request,{})cache_key=(include_headers,keep_fragments)ifcache_keynotincache:#使用sha1方法生成指纹fp=hashlib.sha1()fp.update(to_bytes(request.method))fp.update(to_bytes(canonicalize_url(request.url,keep_fragments=keep_fragments)))fp.update(request.bodyorb)ifinclude_headers:forhdrininclude_headers:ifhdrinrequest.headers:fp.update(hdr)forvinrequest.headers.getlist(hdr):fp.update(v)cache[cache_key]=fp.hexdigest()returncache[cache_key]Engine._download下载请求
classEngine:def_download(self,request,spider):#下载slot=self.slotslot.add_request(request)def_on_success(response):#成功回调，结果必须是Request或Responseifnotisinstance(response,(Response,Request)):raiseTypeError("Incorrecttype:expectedResponseorRequest,got"f"{type(response)}:{response!r}")#如果结果为Response则返回Responseifisinstance(response,Response):ifresponse.requestisNone:response.request=requestlogkws=self.logformatter.crawled(response.request,response,spider)iflogkwsisnotNone:logger.log(*logformatter_adapter(logkws),extra={spider:spider})self.signals.send_catch_log(signal=signals.response_received,response=response,request=response.request,spider=spider,)returnresponsedef_on_
