标签：初始 exec 数据 exe ted com lin info 处理
首先需要我们导入三个基本的模块
1 from twisted.internet import reactor 2 from twisted.web.client import getPage 3 from twisted.internet import defer
getPage的作用是自动帮我们创建socket对象，模拟浏览器给服务器发送请求，放到reactor里面，一旦下载到需要的数据，socket将发生变化，就可以被reactor监听到。（如果下载完成，自动从事件循环中移除）
defer里面的defer.Deferred()的作用是创建一个特殊的socket对象，它不会发送请求，如果放在事件循环里面，将永远不会被移除掉，需要我们自己手动移除，作用就是使事件循环一直监听，不会停止。（不会发请求，手动移除）
1、利用getPage创建socket
def response(content): print(content) def task(): url = "http://www.baidu.com" d = getPage(url) # 创建socket对象，获取返回的对象d,它其实也是一个Deferred对象 d.addCallback(response) # 当页面下载完成,自动执行函数response，来处理返回数据
但以上内容你可以理解只是创建了socket,并且给了运行结束后应该执行的函数。但还没有加到事件循环。
2.将socket添加到事件循环中
def response(content): print(content) @defer.inlineCallbacks # 添加事件循环 def task(): url = "http://www.baidu.com" d = getPage(url) d.addCallback(response) yield d
3、开始事件循环
def response(content): print(content) @defer.inlineCallbacks def task(): url = "http://www.baidu.com" d = getPage(url.encode(‘utf-8‘)) d.addCallback(response) yield d task() # 执行函数 reactor.run() # 开启事件循环。不先执行函数的话，在内部循环中该函数是不会执行的。
4、开始事件循环（自动结束）
def response(content): print(content) @defer.inlineCallbacks def task(): url = "http://www.baidu.com" d = getPage(url.encode(‘utf-8‘)) d.addCallback(response) yield d d = task() dd = defer.DeferredList([d,]) dd.addBoth(lambda _:reactor.stop()) # 监听d是否完成或者失败，如果是，调用回调函数，终止事件循环 reactor.run()
5、并发操作
_closewait = None def response(content): print(content) @defer.inlineCallbacks def task(): """ 每个爬虫的开始：stats_request :return: """ url = "http://www.baidu.com" d1 = getPage(url.encode(‘utf-8‘)) d1.addCallback(response) url = "http://www.cnblogs.com" d2 = getPage(url.encode(‘utf-8‘)) d2.addCallback(response) url = "http://www.bing.com" d3 = getPage(url.encode(‘utf-8‘)) d3.addCallback(response) global _closewait _close = defer.Deferred() yield _closewait spider1 = task() dd = defer.DeferredList([spider1,]) dd.addBoth(lambda _:reactor.stop()) reactor.run()
6、手动关闭
_closewait = None count = 0 def response(content): print(content) global count count += 1 if count == 3: # 当等于3时，说明发出去的url的响应都拿到了 _closewait.callback(None) @defer.inlineCallbacks def task(): """ 每个爬虫的开始：stats_request :return: """ url = "http://www.baidu.com" d1 = getPage(url.encode(‘utf-8‘)) d1.addCallback(response) url = "http://www.cnblogs.com" d2 = getPage(url.encode(‘utf-8‘)) d2.addCallback(response) url = "http://www.bing.com" d3 = getPage(url.encode(‘utf-8‘)) d3.addCallback(response) global _closewait _close = defer.Deferred() yield _closewait spider = task() dd = defer.DeferredList([spider, ]) dd.addBoth(lambda _:reactor.stop()) reactor.run()
（2）将socket添加到事件循环中
（3）开始事件循环 （内部发送请求，并接受响应；当所有的socekt请求完成后，终止事件循环）
在理解了上面的内容之后，那么接下来的部分将会变得轻松，基本的思想和上面差不多。
1、首先我们创建一个爬虫ChoutiSpider，如下代码所示。
有些人可能不知道start_requests是什么，其实这个跟我们用Scrapy创建的一样，这个函数在我们继承的scrapy.Spider里面，作用就是取start_url列表里面的url，并把url和规定好的回调函数parse传给Request类处理。需要注意的是这里的Request我们只是简单的接收来自start_requests传来的值，不做真正的处理，例如发送url请求什么的
from twisted.internet import defer from twisted.web.client import getPage from twisted.internet import reactor class Request(object): def __init__(self,url,callback): self.url = url self.callback = callback class ChoutiSpider(object): name = ‘chouti‘ def start_requests(self): start_url = [‘http://www.baidu.com‘,‘http://www.bing.com‘,] for url in start_url: yield Request(url,self.parse) def parse(self,response): print(response) #response是下载的页面 yield Request(‘http://www.cnblogs.com‘,callback=self.parse)
2、创建引擎Engine，用来接收需要执行的spider对象
from twisted.internet import defer from twisted.web.client import getPage from twisted.internet import reactor class Request(object): def __init__(self,url,callback): self.url = url self.callback = callback class ChoutiSpider(object): name = ‘chouti‘ def start_requests(self): start_url = [‘http://www.baidu.com‘,‘http://www.bing.com‘,] for url in start_url: yield Request(url,self.parse) def parse(self,response): print(response) #response是下载的页面 yield Request(‘http://www.cnblogs.com‘,callback=self.parse) import queue Q = queue.Queue() # 调度器 class Engine(object): def __init__(self): self._closewait = None @defer.inlineCallbacks def crawl(self,spider): start_requests = iter(spider.start_requests()) # 把生成器Request对象转换为迭代器 while True: try: request = next(start_requests) # 把Request对象放到调度器里面 Q.put(request) except StopIteration as e: break self._closewait = defer.Deferred() yield self._closewait
from twisted.internet import defer from twisted.web.client import getPage from twisted.internet import reactor class Request(object): def __init__(self,url,callback): self.url = url self.callback = callback class ChoutiSpider(object): name = ‘chouti‘ def start_requests(self): start_url = [‘http://www.baidu.com‘,‘http://www.bing.com‘,] for url in start_url: yield Request(url,self.parse) def parse(self,response): print(response) #response是下载的页面 yield Request(‘http://www.cnblogs.com‘,callback=self.parse) import queue Q = queue.Queue() class Engine(object): def __init__(self): self._closewait = None self.max = 5 # 最大并发数 self.crawlling = [] # 正在爬取的爬虫个数 def get_response_callback(self,content,request): # content就是返回的response， request就是一开始返回的迭代对象 self.crawlling.remove(request) result = request.callback(content) # 执行parse方法 import types if isinstance(result,types.GeneratorType): # 判断返回值是否是yield，如 for req in result: # 果是则加入到队列中 Q.put(req) def _next_request(self): """ 去调度器中取request对象，并发送请求 最大并发数限制 :return: """ # 什么时候终止 if Q.qsize() == 0 and len(self.crawlling) == 0: self._closewait.callback(None) # 结束defer.Deferred return if len(self.crawlling) >= self.max: return while len(self.crawlling) < self.max: try: req = Q.get(block=False) # 设置为非堵塞 self.crawlling.append(req) d = getPage(req.url.encode(‘utf-8‘)) # 发送请求 # 页面下载完成，执行get_response_callback，并且将一开始的迭代对象req作为参数传过去 d.addCallback(self.get_response_callback,req) # 未达到最大并发数，可以再去调度器中获取Request d.addCallback(lambda _:reactor.callLater(0, self._next_request)) # callLater表示多久后调用 except Exception as e: print(e) return @defer.inlineCallbacks def crawl(self,spider): # 将初始Request对象添加到调度器 start_requests = iter(spider.start_requests()) while True: try: request = next(start_requests) Q.put(request) except StopIteration as e: break # 去调度器中取request，并发送请求 # self._next_request() reactor.callLater(0, self._next_request) self._close = defer.Deferred() yield self._closewait spider = ChoutiSpider() _active = set() engine = Engine() d = engine.crawl(spider) _active.add(d) dd = defer.DeferredList(_active) dd.addBoth(lambda _:reactor.stop()) reactor.run()
上面其实就是Scrapy框架的基本实现过程，不过要想更进一步的了解，就需要模仿Scrapy的源码，把各个函数的功能独立封装起来，如下。
from twisted.internet import reactor from twisted.web.client import getPage from twisted.internet import defer from queue import Queue class Request(object): """ 用于封装用户请求相关信息 """ def __init__(self,url,callback): self.url = url self.callback = callback class HttpResponse(object): # 对返回的content和request做进一步封装，集合到response里面。然后通过回调函数parse调用response.xxx可以调用不同的功能。具体过程略 def __init__(self,content,request): self.content = content self.request = request class Scheduler(object): """ 任务调度器 """ def __init__(self): self.q = Queue() def open(self): pass def next_request(self): # 取调度器里面的值并发送 try: req = self.q.get(block=False) except Exception as e: req = None return req def enqueue_request(self,req): self.q.put(req) def size(self): return self.q.qsize() class ExecutionEngine(object): """ 引擎：所有调度 """ def __init__(self): self._closewait = None self.scheduler = None self.max = 5 self.crawlling = [] def get_response_callback(self,content,request): self.crawlling.remove(request) response = HttpResponse(content,request) result = request.callback(response) # 执行request里面的回调函数（parse） import types if isinstance(result,types.GeneratorType): for req in result: self.scheduler.enqueue_request(req) def _next_request(self): if self.scheduler.size() == 0 and len(self.crawlling) == 0: self._closewait.callback(None) return while len(self.crawlling) < self.max: req = self.scheduler.next_request() if not req: return self.crawlling.append(req) d = getPage(req.url.encode(‘utf-8‘)) d.addCallback(self.get_response_callback,req) d.addCallback(lambda _:reactor.callLater(0,self._next_request)) @defer.inlineCallbacks def open_spider(self,start_requests): self.scheduler = Scheduler() # 创建调度器对象 while True: try: req = next(start_requests) except StopIteration as e: break self.scheduler.enqueue_request(req) # 放进调度器里面 yield self.scheduler.open() # 使用那个装饰器后一定要有yield，因此传个 None 相当于yield None reactor.callLater(0,self._next_request) @defer.inlineCallbacks def start(self): self._closewait = defer.Deferred() yield self._closewait class Crawler(object): """ 用户封装调度器以及引擎的... """ def _create_engine(self): # 创建引擎对象 return ExecutionEngine() def _create_spider(self,spider_cls_path): """ :param spider_cls_path: spider.chouti.ChoutiSpider :return: """ module_path,cls_name = spider_cls_path.rsplit(‘.‘,maxsplit=1) import importlib m = importlib.import_module(module_path) cls = getattr(m,cls_name) return cls() @defer.inlineCallbacks def crawl(self,spider_cls_path): engine = self._create_engine() # 需要把任务交给引擎去做 spider = self._create_spider(spider_cls_path) # 通过路径获取spider，创建对象 start_requests = iter(spider.start_requests()) yield engine.open_spider(start_requests) # 把迭代器往调度器里面放 yield engine.start() # 相当于self._closewait方法 class CrawlerProcess(object): """ 开启事件循环 """ def __init__(self): self._active = set() def crawl(self,spider_cls_path): # 每个爬虫过来只负责创建一个crawler对象 """ :param spider_cls_path: :return: """ crawler = Crawler() d = crawler.crawl(spider_cls_path) self._active.add(d) def start(self): # 开启事件循环 dd = defer.DeferredList(self._active) dd.addBoth(lambda _:reactor.stop()) reactor.run() # 想run()起来，则要有装饰器和yield class Commond(object): def run(self): crawl_process = CrawlerProcess() spider_cls_path_list = [‘spider.chouti.ChoutiSpider‘,‘spider.cnblogs.CnblogsSpider‘,] for spider_cls_path in spider_cls_path_list: crawl_process.crawl(spider_cls_path) # 创建多个defer对象，也就是把d添加到_active里面 crawl_process.start() # 统一调用一次，相当于reactor.run() if __name__ == ‘__main__‘: cmd = Commond() cmd.run()
其实上面的代码就是从下往上依次执行，每个类之间各个完成不同的功能。
配合下面的思维导图，希望可以帮助理解。
标签：初始 exec 数据 exe ted com lin info 处理
原文地址：https://www.cnblogs.com/linyuhong/p/10164752.html
