python多進程web爬蟲-提高性能利器

時間 2019-11-11

原文原文鏈接

背景介紹：html

　　小爬我最近給部門開發了一系列OA的爬蟲工具，從selenium前端模擬進化到純requests後臺post請求爬取，效率逐步提高。剛開始能維持在0.5秒/筆。惋惜當數據超過2000筆後，爬取速度逐漸變慢，最終穩定在1-1.2秒/筆。（此處有較大的坑，原則上在萬行數據這個量級上，速度不該該有肉眼可見的衰減幅度的，後期再來填坑）這個速度，咱們部門領導表示「滿意」。可是我我的不滿意這種「從無到有」、「慢總好過純手工」論調。好多不懂的人老是調侃一句：「能夠了，比人手工的速度仍是快些的，畢竟是自動爬取，無人值守！」也有開發人員看了以後直搖頭：「你這麼個爬法，我還不如直接抽出時間修改底層代碼，讓平臺原生支持」，一副對爬蟲的無限鄙夷的神情。前端

　　因爲爬取速度比較慢，爲了提高用戶體驗，我還用selenium載入JS腳原本渲染了一個進度條，效果以下：node

　　體驗效果是提高了很多，至少用戶不用等待的時候過度焦躁。同事見了面也老是跟我打趣：「能夠了，又進步了，這不從黑黑的dos界面進化到web、JS了嗎？「，然而我本身清楚自身不足，本質上這個進度條並不會真正讓爬蟲效率提高。我知道，最終小爬我得去熟悉和掌握多進程、多線程、協程等編程知識，並用到爬蟲中，以提高效率。python

　　看了崔慶才的多線程、多進程文章，對比過python多線程、多進程等技術的優劣，這裏不作贅述。傳送門：https://cuiqingcai.com/3363.htmlweb

小爬的工具中用到的多線程模塊是mulitiprocessing模塊，而後要用到GIL進程鎖Lock。沒有進程鎖的話，數據相互寫入的時候會產生排列不規則和亂碼。PS：加上進程鎖後，程序速度會有必定下滑，可是存儲的數據則很是工整有序。編程

工具的主要思路和步驟：json

————————————————————————————————————————————————————————————————————cookie

一、模擬登錄，拿到網頁session會話，存儲會話中的cookies；session

二、後臺構造data，發送post請求（攜帶步驟1提到的cookies），獲得json文件，提取出須要爬取的url網址列表，並獲得程序的總計算量；多線程

三、利用多進程模塊multiprocessing中的Pool（進程池）以及map（可迭代）、partial（偏函數）方法，進行傳參，構造多進程，爬取步驟2提到的urls列表，直到爬取完畢，同時保存到txt或者csv文件中；

四、利用tkinter模塊構造簡單GUI，提升可視化程度，並利用pyinstaller打包爲exe文件（此步驟待後續再完善）。

須要注意的要點：

一、多進程的代碼須要都寫到」if __name__ == '__main__':「語句下方，不然每一個進程都會執行頭部文件。若是頭部代碼涉及登錄窗口，則很差意思，程序會瞬間彈出跟進程數匹配的登錄窗口，電腦會崩潰的，親！！！

模擬個錯誤範例讓諸位引覺得戒，不要踩坑，請看下圖：

或者這樣看更加壯觀（恐怖）：

二、雖然進程Lock很是有必要，可是鎖不是隨便加，最好是在涉及IO（數據保存）的代碼段添加，以避免嚴重拖慢程序速度，白白犧牲多進程帶來的性能提高。

三、這個Lock要設置成全局變量，能夠在各個進程間通訊和傳遞；

四、須要將步驟1拿到的cookies（這個就至關於拿到用戶權限）掛到步驟3每一個進程process的get請求中，作成全局變量，而pool.map(函數名，urlList)方法雖然通俗易懂，但不能直接接收多個可迭代的參數，此時須要利用partial偏函數的知識來傳遞多個參數；

五、進程的數量原則上沒有特別的限制，可是計算機的CPU性能和內存大小都會限制網頁爬取和解析的速度。因此，進程數並不是越大越好。在小爬個人我的電腦上，當進程數開到15時，CPU就基本維持在90%的利用率，再提高進程數到100，速度的提高都不明顯，反而

電腦會因CPU滿載而致使大機率卡死。因此，務必根據實際狀況選擇程序的進程數。多進程的python再任務管理器中是這樣的：

下面爲小爬上面的工具提到的詳細的代碼，供參考：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2019/1/25 17:15
# @Author  : New June
# @Desc    :
# @Software: PyCharm
from multiprocessing import Process, Lock, Queue,Pool
import time,re,datetime,csv,loginInfo,requests,json
from requests.exceptions import ConnectionError
from lxml import etree
from math import ceil
from bs4 import BeautifulSoup
from functools import  partial
def get_urls(cookies):
    bpmDefNames=['採購訂單結算']
    startDate="2018-09-01"
    endDate="2018-12-01"
    urls=[]
    data_search={
        'page':1,
        'rows':10,
        'condition':
        """[\
            {"column":"BPM_DEF_NAME","exp":"in","value":"""+str(bpmDefNames)+"""},\
            {"column":"DELETE_STATUS","exp":"=","value":0},\
            {"column":"TO_CHAR(TO_DATE(CREATE_DATE,'YYYY-MM-DD HH24:MI:SS'),'YYYY-MM-DD')","exp":">=","value":"%s"},\
            {"column":"TO_CHAR(TO_DATE(CREATE_DATE,'YYYY-MM-DD HH24:MI:SS'),'YYYY-MM-DD')","exp":"<=","value":"%s"},\
            {"column":"CHECK_TYPE","exp":"like","value":"2"},\
            {"column":"LOCKED_STATUS","exp":"=","value":0},\
            {"column":"DELETE_STATUS","orderType":"default","orderKey":"","direction":"ASC"}\
        ]"""%(startDate,endDate),
        'additionalParams':'{}'
    }   
    s=requests.session()
    s.cookies=cookies  #登錄，而後拿到session中的cookies，供後續使用
    try:
        response=s.post(url=url1,data=data_search)
        if response.status_code==200:
            pageContent = response.json()
    except requests.ConnectionError as e:
        print('Error',e.args)
    maxNum=ceil(pageContent['total']/400)  #maxNux即表單翻頁的總頁數       
    for i in range(1,maxNum+1):
        data_search['page']=i
        data_search['rows']=400
        response=s.post(url=url2,data=data_search)
        pageContent = response.json()
        listlen=len(pageContent['rows']) 
        for num in range(listlen): 
            dataId = pageContent['rows'][num]['dataId']  #每一個表單的dataID號
            urls.append(url3+dataId)
    #print("urlLength:",len(urls))
    return urls

def get_content(settlement_href,cookies):
    '''獲取源碼解析網頁並保存到txt'''
    operatorName1=""
    operatorName2="" #做業人員 變量初始化
    operateTime1=""
    operateTime2=""
    s=requests.session()
    s.cookies=cookies
    res = s.get(url=settlement_href,timeout=60)
    #print(res.encoding)
    soup=BeautifulSoup(res.content,"lxml") 
    haf=str(soup.select('script')[6])   #獲得字段，再進行後續提取
    flowHiComments=re.search('.*?flowHiComments\":(.*?),\"flowHiNodeIds.*?',haf,re.S) 
    flowHiComments=json.loads(flowHiComments.group(1))  #獲得頁面評論信息 
    applyerName=re.search('.*?applyerName\":\"(.*?)\".*?',haf,re.S).group(1) #申請人姓名 
    applyerId=soup.find(id="afPersonId")['value']  #申請人編號   
    afFormNumber=soup.find(id="afFormNumber")['value']  #申請單號    
    totalAmount=str(soup.find(id="totalMoneyPlustax")["value"])#含稅金額
    supplierName=soup.find(id="supplierName")["value"]#供應商名稱 
    settlementCategoryName=soup.find(id="settlementCategoryName")["value"].strip()#結算類別 
    companyCode=soup.find(id="companyCode")["value"]#公司代碼
    flowHiComments=json.loads(re.search('.*?flowHiComments\":(.*?),\"flowHiNodeIds.*?',str(soup.select('script')[6]),re.S).group(1))
    for comment in range(len(flowHiComments)-1,-1,-1): #從審批流後往前數
        if operatorName1=="" and flowHiComments[comment]['subOperateType']=='submit' and flowHiComments[comment]['nodeName']== '角色B':
            operatorName1=flowHiComments[comment]['operaterName']
            operateTime1=flowHiComments[comment]['operateTime']
            continue
        elif operatorName2=="" and flowHiComments[comment]['subOperateType']=='submit'and flowHiComments[comment]['nodeName']==  "角色A":
            operatorName2 = flowHiComments[comment]['operaterName']
            operateTime2 = flowHiComments[comment]['operateTime']
            break
    lock.acquire()　#添加鎖
    with open("abc.txt",'a',encoding='utf-8') as f:
        f.write("%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n"%(afFormNumber,companyCode,settlementCategoryName,applyerName,applyerId,totalAmount,supplierName,operatorName1,operateTime1,operatorName2,operateTime2))
    lock.release()#釋放鎖
def init(l):
    global lock #定義lock爲全局變量，供全部進程用
    lock = l
    # print("申請人：%s;申請人ID：%s；申請單號：%s；含稅金額：%s；供應商名稱：%s\n"%(applyerName,applyerId,afFormNumber,totalAmount,supplierName))

if __name__ == '__main__':
    """登錄，拿到登錄後的session"""
    loginData={'redirect':'','username':loginInfo.username,'password':loginInfo.password}
    session=requests.session()
    session.post(login_url,loginData)
    cookies=session.cookies
    totalStartTime = datetime.datetime.now()  #全部業務的起始時間
    urls=get_urls(cookies)
    with open("abc.txt","w",encoding="utf-8") as t:
        t.write("表單號\t公司碼\t結算類別\t申請人\t申請人ID\t含稅金額\t供應商名稱\t角色A\t角色A提交時間\t角色B\t角色B提交時間\n")
    lock = Lock()
    pool = Pool(processes=20,initializer=init, initargs=(lock,))  #設定進程數爲20
    pool.map(partial(get_content,cookies=cookies),urls)  #利用偏函數傳遞多個參數給get_content函數
    pool.close()
    pool.join()
    endtime=datetime.datetime.now()
    print("time consuming:%d seconds"%(endtime-totalStartTime).seconds)