好的建设网站公司,天津网站建设服务公司,网页制作与前端开发,百度视频免费高清网站Python实现基于协程的异步爬虫一、课程介绍1. 课程来源本课程核心部分来自《500 lines or less》项目#xff0c;作者是来自 MongoDB 的工程师 A. Jesse Jiryu Davis 与 Python 之父 Guido van Rossum。项目代码使用 MIT 协议#xff0c;项目文档使用 http://creativecommons…Python实现基于协程的异步爬虫一、课程介绍1. 课程来源本课程核心部分来自《500 lines or less》项目作者是来自 MongoDB 的工程师 A. Jesse Jiryu Davis 与 Python 之父 Guido van Rossum。项目代码使用 MIT 协议项目文档使用 http://creativecommons.org/licenses/by/3.0/legalcode 协议。课程内容在原文档基础上做了稍许修改增加了部分原理介绍步骤的拆解分析及源代码注释。2. 内容简介传统计算机科学往往将大量精力放在如何追求更有效率的算法上。但如今大部分涉及网络的程序它们的时间开销主要并不是在计算上而是在维持多个Socket连接上。亦或是它们的事件循环处理的不够高效导致了更多的时间开销。对于这些程序来说它们面临的挑战是如何更高效地等待大量的网络事件并进行调度。目前流行的解决方式就是使用异步I/O。本课程将探讨几种实现爬虫的方法从传统的线程池到使用协程每节课实现一个小爬虫。另外学习协程的时候我们会从原理入手以ayncio协程库为原型实现一个简单的异步编程模型。本课程实现的爬虫为爬一个整站的爬虫不会爬到站点外面去且功能较简单主要目的在于学习原理提供实现并发与异步的思路并不适合直接改写作为日常工具使用。3. 课程知识点本课程项目完成过程中我们将学习线程池实现并发爬虫回调方法实现异步爬虫协程技术的介绍一个基于协程的异步编程模型协程实现异步爬虫二、实验环境本课程使用Python 3.4所以本课程内运行py脚本都是使用python3命令。打开终端进入 Code 目录创建 crawler 文件夹, 并将其作为我们的工作目录。$ cd Code$ mkdir crawler cd crawler环保起见测试爬虫的网站在本地搭建。我们使用 Python 2.7 版本官方文档作为测试爬虫用的网站wget http://labfile.oss.aliyuncs.com/courses/574/python-doc.zipunzip python-doc.zip安装serve一个用起来很方便的静态文件服务器sudo npm install -g serve启动服务器serve python-doc如果访问不了npm的资源也可以用以下方式开启服务器ruby -run -ehttpd python-doc -p 3000访问localhost:3000查看网站三、实验原理什么是爬虫网络爬虫(又被称为网页蜘蛛网络机器人在FOAF社区中间更经常的称为网页追逐者)是一种按照一定的规则自动地抓取万维网信息的程序或者脚本。爬虫的工作流程网络爬虫基本的工作流程是从一个根URL开始抓取页面解析页面中所有的URL将还没有抓取过的URL放入工作队列中之后继续抓取工作队列中的URL重复抓取、解析将解析到的url放入工作队列的步骤直到工作队列为空为止。线程池、回调、协程我们希望通过并发执行来加快爬虫抓取页面的速度。一般的实现方式有三种线程池方式开一个线程池每当爬虫发现一个新链接就将链接放入任务队列中线程池中的线程从任务队列获取一个链接之后建立socket完成抓取页面、解析、将新连接放入工作队列的步骤。回调方式程序会有一个主循环叫做事件循环在事件循环中会不断获得事件通过在事件上注册解除回调函数来达到多任务并发执行的效果。缺点是一旦需要的回调操作变多代码就会非常散变得难以维护。协程方式同样通过事件循环执行程序利用了Python 的生成器特性生成器函数能够中途停止并在之后恢复那么原本不得不分开写的回调函数就能够写在一个生成器函数中了这也就实现了协程。四、实验一线程池实现爬虫使用socket抓取页面需要先建立连接之后发送GET类型的HTTP报文等待读入将读到的所有内容存入响应缓存。def fetch(url):sock socket.socket()sock.connect((localhost.com, 3000))request GET {} HTTP/1.0\r\nHost: localhost\r\n\r\n.format(url)sock.send(request.encode(ascii))response bchunk sock.recv(4096)while chunk:response chunkchunk sock.recv(4096)links parse_links(response)q.add(links)默认的socket连接与读写是阻塞式的在等待读入的这段时间的CPU占用是被完全浪费的。多线程默认这部分同学们都是学过的所以就粗略记几个重点没学过的同学可以直接参考廖雪峰的教程廖雪峰的官方网站-Python多线程导入线程库import threading开启一个线程的方法t 你新建的线程t.start() #开始运行线程t.join() #你的当前函数就阻塞在这一步直到线程运行完建立线程的两种方式#第一种:通过函数创建线程def 函数a():passt threading.Thread(target函数a,name自己随便取的线程名字)#第二种:继承线程类class Fetcher(threading.Thread):def __init__(self):Thread.__init__(self):#加这一步后主程序中断退出后子线程也会跟着中断退出self.daemon Truedef run(self):#线程运行的函数passt Fetcher()线程同时操作一个全局变量时会产生线程竞争所以需要锁lock threading.Lock()lock.acquire() #获得锁#..操作全局变量..lock.release() #释放锁多线程同步队列多线程同步就是多个线程竞争一个全局变量时按顺序读写一般情况下要用锁但是使用标准库里的Queue的时候它内部已经实现了锁不用程序员自己写了。导入队列类from queue import Queue创建一个队列q Queue(maxsize0)maxsize为队列大小为0默认队列大小可无穷大。队列是先进先出的数据结构q.put(item) #往队列添加一个item队列满了则阻塞q.get(item) #从队列得到一个item队列为空则阻塞还有相应的不等待的版本这里略过。队列不为空或者为空但是取得item的线程没有告知任务完成时都是处于阻塞状态q.join() #阻塞直到所有任务完成线程告知任务完成使用task_doneq.task_done() #在线程内调用实现线程池创建thread.py文件作为爬虫程序的文件。我们使用seen_urls来记录已经解析到的url地址seen_urls set([/])创建Fetcher类class Fetcher(Thread):def __init__(self, tasks):Thread.__init__(self)#tasks为任务队列self.tasks tasksself.daemon Trueself.start()def run(self):while True:url self.tasks.get()print(url)sock socket.socket()sock.connect((localhost, 3000))get GET {} HTTP/1.0\r\nHost: localhost\r\n\r\n.format(url)sock.send(get.encode(ascii))response bchunk sock.recv(4096)while chunk:response chunkchunk sock.recv(4096)#解析页面上的所有链接links self.parse_links(url, response)lock.acquire()#得到新链接加入任务队列与seen_urls中for link in links.difference(seen_urls):self.tasks.put(link)seen_urls.update(links)lock.release()#通知任务队列这个线程的任务完成了self.tasks.task_done()使用正则库与url解析库来解析抓取的页面这里图方便用了正则同学也可以用Beautifulsoup等专门用来解析页面的Python库import urllib.parseimport re在Fetcher中实现parse_links解析页面:def parse_links(self, fetched_url, response):if not response:print(error: {}.format(fetched_url))return set()if not self._is_html(response):return set()#通过href属性找到所有链接urls set(re.findall(r(?i)href[]?([^\s]),self.body(response)))links set()for url in urls:#可能找到的url是相对路径这时候就需要join一下绝对路径的话就还是会返回urlnormalized urllib.parse.urljoin(fetched_url, url)#url的信息会被分段存在parts里parts urllib.parse.urlparse(normalized)if parts.scheme not in (, http, https):continuehost, port urllib.parse.splitport(parts.netloc)if host and host.lower() not in (localhost):continue#有的页面会通过地址里的#frag后缀在页面内跳转这里去掉frag的部分defragmented, frag urllib.parse.urldefrag(parts.path)links.add(defragmented)return links#得到报文的html正文def body(self, response):body response.split(b\r\n\r\n, 1)[1]return body.decode(utf-8)def _is_html(self, response):head, body response.split(b\r\n\r\n, 1)headers dict(h.split(: ) for h in head.decode().split(\r\n)[1:])return headers.get(Content-Type, ).startswith(text/html)实现线程池类与main的部分class ThreadPool:def __init__(self, num_threads):self.tasks Queue()for _ in range(num_threads):Fetcher(self.tasks)def add_task(self, url):self.tasks.put(url)def wait_completion(self):self.tasks.join()if __name__ __main__:start time.time()#开4个线程pool ThreadPool(4)#从根地址开始抓取页面pool.add_task(/)pool.wait_completion()print({} URLs fetched in {:.1f} seconds.format(len(seen_urls),time.time() - start))运行效果这里先贴出完整代码from queue import Queuefrom threading import Thread, Lockimport urllib.parseimport socketimport reimport timeseen_urls set([/])lock Lock()class Fetcher(Thread):def __init__(self, tasks):Thread.__init__(self)self.tasks tasksself.daemon Trueself.start()def run(self):while True:url self.tasks.get()print(url)sock socket.socket()sock.connect((localhost, 3000))get GET {} HTTP/1.0\r\nHost: localhost\r\n\r\n.format(url)sock.send(get.encode(ascii))response bchunk sock.recv(4096)while chunk:response chunkchunk sock.recv(4096)links self.parse_links(url, response)lock.acquire()for link in links.difference(seen_urls):self.tasks.put(link)seen_urls.update(links)lock.release()self.tasks.task_done()def parse_links(self, fetched_url, response):if not response:print(error: {}.format(fetched_url))return set()if not self._is_html(response):return set()urls set(re.findall(r(?i)href[]?([^\s]),self.body(response)))links set()for url in urls:normalized urllib.parse.urljoin(fetched_url, url)parts urllib.parse.urlparse(normalized)if parts.scheme not in (, http, https):continuehost, port urllib.parse.splitport(parts.netloc)if host and host.lower() not in (localhost):continuedefragmented, frag urllib.parse.urldefrag(parts.path)links.add(defragmented)return linksdef body(self, response):body response.split(b\r\n\r\n, 1)[1]return body.decode(utf-8)def _is_html(self, response):head, body response.split(b\r\n\r\n, 1)headers dict(h.split(: ) for h in head.decode().split(\r\n)[1:])return headers.get(Content-Type, ).startswith(text/html)class ThreadPool:def __init__(self, num_threads):self.tasks Queue()for _ in range(num_threads):Fetcher(self.tasks)def add_task(self, url):self.tasks.put(url)def wait_completion(self):self.tasks.join()if __name__ __main__:start time.time()pool ThreadPool(4)pool.add_task(/)pool.wait_completion()print({} URLs fetched in {:.1f} seconds.format(len(seen_urls),time.time() - start))运行python3 thread.py命令查看效果(记得先开网站服务器)使用标准库中的线程池线程池直接使用multiprocessing.pool中的ThreadPool代码更改如下from multiprocessing.pool import ThreadPool#...省略中间部分...#...去掉Fetcher初始化中的self.start()#...删除自己实现的ThreadPool...if __name__ __main__:start time.time()pool ThreadPool()tasks Queue()tasks.put(/)Workers [Fetcher(tasks) for i in range(4)]pool.map_async(lambda w:w.run(), Workers)tasks.join()pool.close()print({} URLs fetched in {:.1f} seconds.format(len(seen_urls),time.time() - start))使用ThreadPool时它处理的对象可以不是线程对象实际上Fetcher的线程部分ThreadPool根本用不到。因为它自己内部已开了几个线程在等待任务输入。这里偷个懒就只把self.start()去掉了。可以把Fetcher的线程部分全去掉效果是一样的。ThreadPool活用了map函数这里它将每一个Fetcher对象分配给线程池中的一个线程线程调用了Fetcher的run函数。这里使用map_async是因为不希望它在那一步阻塞我们希望在任务队列join的地方阻塞那么到队列为空且任务全部处理完时程序就会继续执行了。运行python3 thread.py命令查看效果线程池实现的缺陷我们希望爬虫的性能能够进一步提升但是我们没办法开太多的线程因为线程的内存开销很大每创建一个线程可能需要占用50k的内存。以及还有一点网络程序的时间开销往往花在I/O上socket I/O 阻塞时的那段时间是完全被浪费了的。那么要如何解决这个问题呢下节课你就知道啦下节课见
