爬蟲的單線程+多任務異步協程:asyncio 3.6

單線程+多任務異步協程:asyncio 3.6

• 事件循環
  • 無限循環的對象.事件循環中最終須要將一些 特殊的函數(被async關鍵字修飾的函數) 註冊在該對象中.
• 協程
  • 本質上是一個對象.能夠把協程對象(特殊的函數)註冊到事件循環中
• 任務對象
  • 就是對協程對象進一步的封裝.
• 綁定回調: task.add_done_callback(func)
  • func(task):task參數表示的就是綁定的任務對象
  • task.result():返回就是任務對象對應的特殊函數內部的返回值
  • 回調多被用做於爬蟲中的解析方法
• await
  • 在任務對象對應的特殊函數內部的實現語句中,若是出現了阻塞的操做,則必須使用await進行修飾
• 異步操做的體現
  • 當將多個協程對象(特殊的函數)註冊到事件循環中後,事件循環開啓後,則會循環執行其內部的協程對象們
  • 假如事件循環對象在執行某一個協程對象時,發生了阻塞,則事件循環對象不會等待阻塞結束,反而會執行下一個協程對象
• aiohttp:支持異步的網絡請求模塊
  • 中文文檔
    • https://www.cntofu.com/book/127/aiohttp%E6%96%87%E6%A1%A3/ClientUsage.md
  • 環境安裝 pip或者直接pycharm安裝均可以
  • 如何進行UA假裝:
    • session.get(url,headers)
  • 參數的封裝
    • session.get(url,headers,data/params)
  • 代理IP方式:
    • session.get(url,proxy="http://ip:port")

• 簡單示例

  import asyncio
  #特殊的函數：該函數調用後,函數內部的程序語句不會被執行,可是該函數調用會返回一個協程對象
  async def test():
      print('i am test()')
      print('i am test()')
      print('i am test()')
  
  #調用該特殊函數,讓其返回一個協程對象
  c = test()
  
  #建立一個事件循環對象
  loop = asyncio.get_event_loop()
  
  #將協程對象註冊到事件循環對象中,而且開啓事件循環
  loop.run_until_complete(c)
  
  print(c)

• 任務對象的使用

  import asyncio
  #特殊的函數：該函數調用後,函數內部的程序語句不會被執行,可是該函數調用會返回一個協程對象
  async def test():
      print('i am test()')
  
  #調用該特殊函數,讓其返回一個協程對象
  c = test()
  
  #將協程對象封裝到任務對象中
  task = asyncio.ensure_future(c)
  
  #建立一個事件循環對象
  loop = asyncio.get_event_loop()
  
  #將任務對象註冊到事件循環對象中,而且開啓事件循環
  loop.run_until_complete(task)

• 任務對象綁定回調函數

  import asyncio
  #特殊的函數：該函數調用後,函數內部的程序語句不會被執行,可是該函數調用會返回一個協程對象
  async def test():
      print('i am test()')
      return 'hello bobo'
  
  #任務對象的回調函數,參數task表示的就是任務對象
  def func(task):
      # print('i am task callback!')
      print(task.result()) #返回的是任務對象對應的特殊函數的返回值
  
  #調用該特殊函數,讓其返回一個協程對象
  c = test()
  
  #將協程對象封裝到任務對象中
  task = asyncio.ensure_future(c)
  
  #給任務對象綁定一個回調函數
  task.add_done_callback(func)
  
  #建立一個事件循環對象
  loop = asyncio.get_event_loop()
  
  #將任務對象註冊到事件循環對象中,而且開啓事件循環
  loop.run_until_complete(task)

• 多任務異步協程

  import asyncio
  import time
  #函數內部不能夠出現不支持異步模塊的代碼
  #該函數內部的異步操做必須使用await進行修飾
  async def request(url):
      print('正在下載:',url)
      # time.sleep(2) #time模塊是一個不支持異步的模塊
      await asyncio.sleep(2) #asyncio模塊中提供的一個支持異步的阻塞方法
      print(url,'下載完畢!')
      return url
  
  #建立一個回調函數
  def callback(task):
    #返回的是任務對象對應的特殊函數的返回值
      print(task.result())
  
  urls = [
      'www.1.com',
      'www.2.com',
      'www.3.com',
      'www.4.com',
  ]
  #記錄開始時間
  start = time.time()
  #任務列表
  tasks = []
  for url in urls:
    #調用該特殊函數,讓其返回一個協程對象
      c = request(url)
      #將協程對象封裝到任務對象中
      task = asyncio.ensure_future(c)
      # 給任務對象綁定回調
      task.add_done_callback(callback)
    #將任務對象添加到列表中
      tasks.append(task)
  
  #建立一個事件循環對象
  loop = asyncio.get_event_loop()
  #將任務對象列表註冊到事件循環對象中,而且開啓事件循環
  loop.run_until_complete(asyncio.wait(tasks))
  ##記錄結束時間
  print(time.time()-start)

• 單線程+多任務異步協程的爬蟲

  import asyncio
  import requests
  import time
  import aiohttp
  from lxml import etree
  urls = [
      'http://localhost:5000/bobo',
      'http://localhost:5000/jay',
      'http://localhost:5000/tom',
  
      'http://localhost:5000/bobo',
      'http://localhost:5000/jay',
      'http://localhost:5000/tom'
  ]
  
  # async def get_page(url):
  #     #requests模塊是一個不支持異步的模塊,解決方法就是使用一個支持異步的模塊進行請求發送
  #     page_text =  requests.get(url=url).text
  #     return page_text
  
  async def get_page(url):
      #使用aiohttp進行請求發送
      #實例化了一個發送網絡請求的對象
      async with aiohttp.ClientSession() as session:
        #該函數內部的異步操做必須使用await進行修飾
          async with await session.get(url) as response:
              #獲取響應數據(頁面源碼數據)
              page_text = await response.text()
              # print(page_text)
              return page_text
  #數據解析的操做須要在回調函數中實現
  def parse(task):
      page_text = task.result()
      tree = etree.HTML(page_text)
      parse_data = tree.xpath('//body/text()')[0]
      print(parse_data)
  
  
  start = time.time()
  tasks = []
  for url in urls:
    #調用該特殊函數,讓其返回一個協程對象
      c = get_page(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)

• 單線程+多任務異步協程的應用

  #爬取喜馬拉雅中的相聲音頻
  import requests
  import aiohttp
  import asyncio
  #通用的url模板
  url = 'https://www.ximalaya.com/revision/play/album?albumId=19366477&pageNum=%d&sort=1&pageSize=2'
  headers = {
      'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36'
  }
  #獲取了全部即將被下載的音頻鏈接
  urls = []
  for page in range(1,3):
      new_url = format(url%page)
      dic_obj = requests.get(url=new_url,headers=headers).json()
      for dic in dic_obj['data']['tracksAudioPlay']:
          audio_url = dic['src']
          urls.append(audio_url)
  
  #特殊的函數：該函數調用後,函數內部的程序語句不會被執行,可是該函數調用會返回一個協程對象
  async def get_audio_data(url):
    #使用aiohttp進行請求發送
      #實例化了一個發送網絡請求的對象
      async with aiohttp.ClientSession() as s:
        #該函數內部的異步操做必須使用await進行修飾
          async with await s.get(url=url,headers=headers) as response:
              audio_data = await response.read()  #read()返回的是二進制形式的響應數據
              return {'data':audio_data,'url':url}
  
  #任務對象的回調函數,進行數據的持久化存儲
  def saveData(task):
      dic_obj = task.result()
      name = dic_obj['url'].split('/')[-1]
      data = dic_obj['data']
      with open(name,'wb') as fp:
          fp.write(data)
      print(name+'下載完畢!')
  
  tasks = []
  for url in urls:
    #調用該特殊函數,讓其返回一個協程對象
      c = get_audio_data(url)
      #將協程對象封裝到任務對象中
      task = asyncio.ensure_future(c)
      # 給任務對象綁定回調函數
      task.add_done_callback(saveData)
      #將任務對象添加到列表中
      tasks.append(task)
  #建立一個事件循環對象
  loop = asyncio.get_event_loop()
  #將任務對象列表註冊到事件循環對象中,而且開啓事件循環
  loop.run_until_complete(asyncio.wait(tasks))