4. 异步爬虫

异步爬虫

一、基于单线程的异步爬虫

• 使用Flask搭建网站进行一部请求爬取测试

  from flask import Flask,render_template
  import time
  
  # 实例化一个app
  app = Flask(__name__)
  
  # 创建视图函数&路由地址
  @app.route('/bobo')
  def index_1():
      time.sleep(2)
      return render_template('test.html')
  
  @app.route('/jay')
  def index_2():
      time.sleep(2)
      return render_template('test.html')
  
  @app.route('/tom')
  def index_3():
      time.sleep(2)
      return render_template('test.html')
  
  if __name__ == '__main__':
      # debug-True表示开启调试模式：服务器端代码被修改后按下保存键会自动重启服务
      app.run(debug=True)
  

• 单线程的同步爬虫具体实现

  import requests
  import time
  
  urls = [
      'http://127.0.0.1:5000/bobo',
      'http://127.0.0.1:5000/jay',
      'http://127.0.0.1:5000/tom',
  ]
  
  
  def get_requests(url):
      page_text = requests.get(url=url).text
      return len(page_text)
  
  # 同步代码
  if __name__ == '__main__':
      start = time.time()
      for url in urls:
          result = get_requests(url)
          print(result)
      print('总耗时：', time.time() - start)
  
  # 执行结果
  """
      1002
      1002
      1002
      总耗时： 6.123658895492554
  """
  

• 单线程的异步爬虫具体实现

  import requests
  import time
  from multiprocessing.dummy import Pool
  
  urls = [
      'http://127.0.0.1:5000/bobo',
      'http://127.0.0.1:5000/jay',
      'http://127.0.0.1:5000/tom',
  ]
  
  def get_requests(url):
      page_text = requests.get(url=url).text
      return len(page_text)
  
  if __name__ == '__main__':
      start = time.time()
      pool = Pool(3) # 3表示开启线程的数量
      #使用get_requests作为回调函数，需要给予异步形式对urls列表中的每一列表元素进行操作
      # 保证回调函数必须要有一个参数和返回值
      result = pool.map(get_requests,urls)
      print(result)
      print('总耗时：', time.time() - start)
  # 执行结果
  """
      [1002, 1002, 1002]
      总耗时： 2.077686071395874
  """
  

• 线程池的概念

  from multiprocessing.dummy import Pool
  
  map(callbabck,alist)：可以使用callback对alist中的每一个元素进行指定形式的异步操作
  	callback:回调函数，必须要有一个形式参数和一个返回值，如需传入多个参数，只需自己构建一个列表或是字典
  	alist:必须是一个可迭代对象
  

二、基于单线程+多任务的异步爬虫

安装：pip install asyncio

1. 基础知识

• 特殊函数

  • 如果一个函数的定义被asyncio修饰后，则该函数就会变成一个特殊函数

  • 特殊之处

    • 该函数被调用后，函数内部的实现语句不会立即执行
    • 该特殊函数被调用后会返回一个协程对象

• 协程对象

  • 是一个对象，通过特殊函数的调用返回
  • 协程对象 == 特殊函数 == 一组指定的操作（可以理解为特殊函数的内部代码）
  • 协程对象 == 一组特定的操作

• 任务对象

  • 任务对象就是一个高级的协程对象。（任务对象就是对协程对象的进一步封装）

  • 任务对象 == 协程对象 == 特殊函数 == 一组指定的操作（可以理解为特殊函数的内部代码）

  • 任务对象 == 一组指定的操作

  • 如何创建一个任务对象

    • asyncio.ensure_future(协程对象)

  • 任务对象的高级之处

    """
    可以给任务对象绑定回调函数
    	任务对象.add_callback(回调函数)
    	回调函数的调用时机
    		任务被执行结束后，才可以调用回调函数
        回调函数的参数只可以有一个：表示的就是该回调函数的调用者（任务对象）
        使用回调函数的参数调用`result()`，返回的就是特殊函数的返回值结果
    """
    

• 事件循环对象

  • 是一个对象
  • 作用
    1. 可以将多个任务对象注册/装载到事件循环对象中
    2. 如果开启事件循环后，则其内部注册/装载的任务对象表示的指定操作就会基于异步执行
  • 创建方式
    • loop = asyncio.get_event_loop()
  • 注册且启动的方式
    • loop.run_until_complete(任务对象)

2. 基础知识具体代码实现

import asyncio
import requests
import time

async def get_request(url):
    print('正在请求的url：',url)
    time.time()
    print('请求结束：',url)
    return 'jason'

# 回调函数的封装
# 参数t就是该回调函数的调用者（任务对象）
def task_callback(t):
    print('i am task_callback(),参数t：',t)
    # result返回的就是特殊函数的返回值
    print('t.result()返回的是：',t.result())

if __name__ == '__main__':
    # c就是一个协程对象
    c = get_request('www.1.com')

    # 任务对象就是对协程对象的进一步封装
    task = asyncio.ensure_future(c)

    # 给task绑定一个回调函数
    task.add_done_callback(task_callback)

    # 创建事件循环对象
    loop = asyncio.get_event_loop()
    # 将任务对象注册到时间循环中且开启事件循环
    loop.run_until_complete(task)

3. 真正多任务爬虫实现（简单示例）

import asyncio
import time
import requests

# async def get_request(url):
#     print('正在请求的url：',url)
#     time.sleep(2) # 不支持异步模块的代码
#     print('请求结束：',url)
#     return 'bobo'


async def get_request(url):
    print('正在请求的url：',url)
    await asyncio.sleep(2)# 支持异步模块的代码
    print('请求结束：',url)
    return 'bobo'

urls = [
    'www.1.com',
    'www.2.com',
    'www.3.com',
]

# urls = [
#     'http://127.0.0.1:5000/bobo',
#     'http://127.0.0.1:5000/jay',
#     'http://127.0.0.1:5000/tom',
# ]


if __name__ == '__main__':
    start = time.time()
    tasks = []
    # 1.创建协程对象
    for url in urls:
        c = get_request(url)
        # 2.创建任务对象
        task = asyncio.ensure_future(c)
        tasks.append(task)
    # 创建事件循环对象
    loop = asyncio.get_event_loop()
    # loop.run_until_complete(tasks)
    # 必须使用wait方法对tasks进行封装即可
    loop.run_until_complete(asyncio.wait(tasks))
    print('总耗时：',time.time()-start)

函数介绍

1. wait方法的作用
	将任务列表中的任务对象赋予可被挂起的权限，主要任务对象被赋予了可被挂起的权限后，该任务才可以异步执行（挂起：要求当前任务对象交出CPU的使用权）
    
2. 注意事项
	在特殊函数内部不可以出现不支持异步操作的代码，否则会中断整个程序的异步效果
    
3. await关键字
	在特殊函数的内部，凡是阻塞操作前必须加上await进行修饰。await就是可以保证阻塞操作在异步过程中不会被跳过

4. 多任务异步爬虫一

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import requests
import time
import asyncio
import aiohttp
from lxml import etree

urls = [
    'http://127.0.0.1:5000/bobo',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom',
]


# async def get_request(url):
#     # requests是一个不支持异步的模块
#     page_text = requests.get(url=url).text
#     return page_text

# 解析函数的封装
def parse(t):
    # 获取到请求页面源码数据
    page_text = t.result()
    tree = etree.HTML(page_text)
    parse_text = tree.xpath('//a[@id="feng"]/text()')[0]
    print(parse_text)


async def get_request(url):
    # 实例化好了一请求对象
    async with aiohttp.ClientSession() as sess:
        # 调用get发起请求，但会一个响应对象
        # get/post(url,headers,params/data,proxy="http://ip:port")
        async with sess.get(url=url) as response:
            # text()获取了字符串类型的响应数据
            # read()获取byte类型的响应对象
            # json()获取json类型的响应对象
            page_text = await response.text()
            return page_text


if __name__ == '__main__':
    start = time.time()
    tasks = []
    for url in urls:
        c = get_request(url)
        task = asyncio.ensure_future(c)
        task.add_done_callback(parse)
        tasks.append(task)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    print('总耗时：', time.time() - start)

5. 多任务异步爬虫二（实战：爬取必应壁纸）

import aiohttp
from lxml import etree
import asyncio
import time
import requests

MAIN_URL = 'https://bing.ioliu.cn/?p={}'
HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.77 Safari/537.36'
}

# 同步爬取
# start = time.time()
# for i in range(1,100):
#     url = MAIN_URL.format(i)
#     page_text = requests.get(url=url,headers=HEADERS).text
#     tree = etree.HTML(page_text)
#     div_list = tree.xpath('//div[@class="container"]/div[@class="item"]')
#     for div in div_list:
#         path = div.xpath('.//img/@data-progressive')[0]
#         name = div.xpath(".//div[@class='description']/h3/text()")[0]
#         print(name,'===',path)
# print("总耗时：",time.time() - start)  # 29.509535789489746


# 异步爬取页面源码
async def get_response(url):
    async  with aiohttp.ClientSession() as sess:
        async with sess.get(url=url,headers=HEADERS) as response:
            page_text = await response.text()
            return page_text


# 异步解析页面源码
async def get_parse(page_text):
    tree = etree.HTML(page_text)
    div_list = tree.xpath('//div[@class="container"]/div[@class="item"]')
    for div in div_list:
        path = div.xpath('.//img/@data-progressive')[0]
        name = div.xpath(".//div[@class='description']/h3/text()")[0]
        print(name,'===',path)
    return True


# 讲爬取到的全部页面源码存在一个列表中
def parse_text(t):
    page_text = t.result()
    page_text_list.append(page_text)
    # tree = etree.HTML(page_text)
    # div_list = tree.xpath('//div[@class="container"]/div[@class="item"]')
    # for div in div_list:
    #     path = div.xpath('.//img/@data-progressive')[0]
    #     name = div.xpath(".//div[@class='description']/h3/text()")[0]
    #     print(name,'===',path)

def main():
    for i in range(1,10):
        url = MAIN_URL.format(i)
        # 生成一个协程对象
        c = get_response(url)

        # 生成一个任务对象
        task = asyncio.ensure_future(c)

        # 给任务对象绑定一个回调函数
        task.add_done_callback(parse_text)

        page_text_tasks.append(task)
    loop1 = asyncio.get_event_loop()
    # 获取网页源码异步
    loop1.run_until_complete(asyncio.wait(page_text_tasks))
	
    # 封装页面解析的特殊函数
    for page_text in page_text_list:
        c = get_parse(page_text)
        task = asyncio.ensure_future(c)
        parse_tasks.append(task)

    loop2 = asyncio.get_event_loop()
    loop2.run_until_complete(asyncio.wait(parse_tasks))

if __name__ == '__main__':
    # 获取到的页面源码列表
    page_text_list = []
    # 页面源码任务对象列表
    page_text_tasks = []
    # 页面解析任务对象
    parse_tasks = []
    start = time.time()
	main()
    print("总耗时：",time.time() - start)  # 17.57803225517273