高性能异步爬虫

目录

• 高性能异步爬虫:
  • 介绍：
    • 同步
      • 解决同步调用方案之多线程/多进程：
  • 异步IO：
  • 多任务异步操作应用到爬虫:
    • 协程操作：
  • aiohttp简介:
    • 添加请求参数
    • UA伪装:
    • 自定义cookies:
    • post请求参数:
    • 设置代理:
    • 解析数据：

高性能异步爬虫:

介绍：

爬虫的本质就是client发请求批量获取server的响应数据

分析处理：
	同步调用：即提交一个任务后就在原地等待任务结束，等到拿到任务的结果后再继续下一行代码
    
    多线程（或多进程）。多线程（或多进程）的目的是让每个连接都拥有独立的线程（或进程），这样任何一个连接的阻塞都不会影响其他的连接
    
    开启多进程或都线程的方式，我们是无法无限制地开启多进程或多线程的：在遇到要同时处理成百上千个的连接请求时，则无论多线程还是多进程都会严重占据系统资源，降低系统对外界响应效率，而且线程与进程本身也更容易进入假死状态。
    
    --> 使用“池”必须考虑其面临的响应规模，并根据响应规模调整“池”的大小

同步

import requests

def parse_page(res):
    print('解析 %s' %(len(res)))

def get_page(url):
    print('下载 %s' %url)
    response=requests.get(url)
    if response.status_code == 200:
        return response.text

urls = [
    'http://xmdx.sc.chinaz.net/Files/DownLoad/jianli/201904/jianli10231.rar',
    'http://zjlt.sc.chinaz.net/Files/DownLoad/jianli/201904/jianli10229.rar',
    'http://xmdx.sc.chinaz.net/Files/DownLoad/jianli/201904/jianli10231.rar'
]
for url in urls:
    res=get_page(url) #调用一个任务，就在原地等待任务结束拿到结果后才继续往后执行
    parse_page(res)

解决同步调用方案之多线程/多进程：

import time
def sayhello(str):           # 线程池
    print("Hello ",str)
    time.sleep(2)

name_list =['xiaozi','aa','bb','cc']
start_time = time.time()
for i in range(len(name_list)):
    sayhello(name_list[i])
print('%d second'% (time.time()-start_time))

异步IO：

单线程+异步协程实现异步IO操作

	event_loop：事件循环，相当于一个无限循环，我们可以把一些函数注册到这个事件循环上，当满足某些条件的时候，函数就会被循环执行。
	程序是按照设定的顺序从头执行到尾，运行的次数也是完全按照设定。当在编写异步程序时，必然其中有部分程序的运行耗时是比较久的，需要先让出当前程序的控制权，让其在背后运行，让另一部分的程序先运行起来
    。当背后运行的程序完成后，也需要及时通知主程序已经完成任务可以进行下一步操作，但这个过程所需的时间是不确定的，需要主程序不断的监听状态，一旦收到了任务完成的消息，就开始进行下一步。loop就是这个持续不断的监视器。

	coroutine：中文翻译叫协程，在 Python 中常指代为协程对象类型，我们可以将协程对象注册到事件循环中，它会被事件循环调用。我们可以使用 async 关键字来定义一个方法，这个方法在调用时不会立即被执行，而是返回一个协程对象。

task：任务，它是对协程对象的进一步封装，包含了任务的各个状态。

future：代表将来执行或还没有执行的任务，实际上和 task 没有本质区别。

多任务异步操作应用到爬虫:

测试：

模拟：一个慢速服务器

from flask import Flask
import time

app = Flask(__name__)


@app.route('/tiger')
def index_tiger():
    time.sleep(2)
    return 'Hello tiger'

@app.route('/jay')
def index_jay():
    time.sleep(2)
    return 'Hello jay'

@app.route('/tom')
def index_tom():
    time.sleep(2)
    return 'Hello tom'

if __name__ == '__main__':
    app.run(threaded=True)

协程操作：

import requests
import asyncio
import time

start = time.time()
urls = [
    'http://127.0.0.1:5000/tiger',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom'
]

async def get_page(url):
    print('正在下载', url)
    # 之所有没有实现异步操作，是因为requests模块是一个非异步的模块
    response = requests.get(url=url)
    print('下载完毕：', response.text)
tasks = []

for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

end = time.time()

print('总耗时:', end-start)

aiohttp简介:

aiohttp可以实现单线程并发IO操作。

环境安装
	pip install aiohttp

async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://www.baidu.com') as resposne:
            print(await resposne.text())

loop = asyncio.get_event_loop()
tasks = [fetch(),]
loop.run_until_complete(asyncio.wait(tasks))

添加请求参数

params = {'key': 'value', 'page': 10}
async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://www.baidu.com/s',params=params) as resposne:
            print(await resposne.url)

loop = asyncio.get_event_loop()
tasks = [fetch(),]
loop.run_until_complete(asyncio.wait(tasks)

UA伪装:

url = 'http://httpbin.org/user-agent'
headers = {'User-Agent': 'test_user_agent'}

async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url,headers=headers) as resposne:
            print(await resposne.text())

loop = asyncio.get_event_loop()
tasks = [fetch(),]
loop.run_until_complete(asyncio.wait(tasks))

自定义cookies:

url = 'http://httpbin.org/cookies'
cookies = {'cookies_name': 'test_cookies'}

async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url,cookies=cookies) as resposne:
            print(await resposne.text())
          

loop = asyncio.get_event_loop()
tasks = [fetch(),]
loop.run_until_complete(asyncio.wait(tasks

post请求参数:

url = 'http://httpbin.org'
payload = {'username': 'zhang', 'password': '123456'}
async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.post(url, data=payload) as resposne:
            print(await resposne.text())

loop = asyncio.get_event_loop()
tasks = [fetch(), ]
loop.run_until_complete(asyncio.wait(tasks))

设置代理:

url = "http://python.org"
async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url, proxy="http://some.proxy.com") as resposne:
        print(resposne.status)

loop = asyncio.get_event_loop()
tasks = [fetch(), ]
loop.run_until_complete(asyncio.wait(tasks))

解析数据：

import time
import asyncio
import aiohttp

# 回调函数: 主要用来解析响应数据
def callback(task):
    print('This is callback')
    # 获取响应数据
    page_text = task.result()
    print("接下来就可以在回调函数中实现数据解析")

async def get_page(url):
    async with aiohttp.ClientSession() as session:
        # 只要有耗时就会有阻塞，就得使用await进行挂起操作
        async with await session.get(url=url) as response:
            page_text = await response.text() # 二进制read()/json()
            print('响应数据', page_text)
            return page_text

start = time.time()
urls = [
    'http://127.0.0.1:5000/tiger',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
]
loop = asyncio.get_event_loop()

tasks = []
for url in urls:
    cone = get_page(url)
    task = asyncio.ensure_future(cone)
    # 给任务对象绑定回调函数用于解析响应数据
    task.add_done_callback(callback)
    tasks.append(task)

loop.run_until_complete(asyncio.wait(tasks))
print('总耗时: ', time.time()-start)