파이썬을 사용하다보니 멀티스레드는 속도가 잘 나지 않는 다는 이야기를 많이본다.

GIL 때문에 라는 말이 많던데 정확하게는 모르겠고 파이썬에서 파이썬 스크립트를 동시에 사용할 수 없다 정도만 알 것 같다.

그래서 대안으로 멀티프로세싱을 사용하다는 것을 보았고, 아래 해당 소스를 남겨본다.

기본적으로는 각 멀티 프로세스가 처리할 함수를 만들고 인자값으로 data_queue와 result_queue를 넘겨 data_queue를 사용하고 result_queue에 결과를 남긴다.

마지막에 처리된 result_queue를 db 프로세스가 읽어 저장하는 식이다.

해당 샘플은 비동기 페이지로 되어있는 마이크로 소프트 cvss 점수를 가져오는 멀티 프로세싱 비동기 웹 크롤러이다.

import os

import sys

import datetime

import multiprocessing

import threading

import queue

import logging

import csv

import requests

from selenium.webdriver.common.by import By

from selenium.webdriver.support.wait import WebDriverWait

from selenium.webdriver.support import expected_conditions as EC

from selenium import webdriver

from selenium.common.exceptions import ElementNotVisibleException

from bs4 import BeautifulSoup

from models import CvssList, CvssScore

import database

# 기본 로깅 설정

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(relativeCreated)6d %(threadName)s %(message)s', datefmt='%H:%M:%S')

_TIME_OUT_ = 120

def read_csv():

"""

cvss목록이 있는 csv파일을 읽어 dict()의 목록으로 리턴한다.

"""

list = []

with open('doc/lists.csv', newline='') as infile:

reader = csv.DictReader(infile)

for row in reader:

list.append(row)

return list

def login_cvss(driver):

"""

로그인

"""

driver.get('https://portal.msrc.microsoft.com/ko-kr/eula')

WebDriverWait(driver, _TIME_OUT_).until(EC.presence_of_element_located((By.CSS_SELECTOR, 'div.securityeula')))

html = driver.page_source

soup = BeautifulSoup(html, 'html.parser')

driver.save_screenshot('logon.png')

if soup.select('#JelloWrapper > div.alley > div.securityeula > div > ui-view > form > input.btn.btn-primary') :

driver.find_element_by_css_selector('input[type=checkbox]').click()

driver.find_element_by_css_selector('input.btn.btn-primary').click()

return True

return None

def read_cvss_scores(driver, uid , cvecode):

"""

뭔가 복잡한 일들을 한다.

"""

def _strip(cell):

return cell.get_text().strip()

def _mk_score(cells):

product = _strip(cells[0].find('a'))

platform = _strip(cells[1])

dfltscore = _strip(cells[2])

tmpscore = _strip(cells[3])

score = CvssScore(uid, product, platform, dfltscore, tmpscore)

return score

score_list = []

try:

# cvss login

url = 'https://portal.msrc.microsoft.com/ko-KR/security-guidance/advisory/{0}'.format(cvecode)

driver.get(url)

wait = WebDriverWait(driver, _TIME_OUT_).until(EC.presence_of_element_located((By.CSS_SELECTOR, 'div.securitydetails')))

# cvss click

html = driver.page_source

soup = BeautifulSoup(html, 'html.parser')

notice = soup.select_one('#JelloWrapper > div.alley > div.securitydetails > div > ui-view > div:nth-of-type(2) > h2')

# parse scores

driver.find_element_by_css_selector('#JelloWrapper > div.alley > div.securitydetails > div > ui-view > div:nth-child(3) > ul > li:nth-child(2) > a').click()

table = soup.select_one('#AffectedProductsWithCVSS > tbody')

for tr in table.find_all('tr'):

cells = tr.find_all('td')

if len(cells) != 6:

continue

score = _mk_score(cells)

score_list.append(score)

except ElementNotVisibleException as e:

pass

except Exception as e:

logging.error(e)

raise e

return score_list

def parse_cvve(data_que, result_que):

"""

data_que의 내용을 읽어 분석 후 결과를 result_que에 저장한다.

data_queue의 Item(tuple) : (cvss코드(string), 포함된 업데이트(dict) 목록, )

result_que의 Item(tuple) : (cvss코드(string), 포함된 업데이트(모델(CvssList)) 목록, cvss점수(모델(CvssScore)) 목록, )

이를 위해 webdriver생성하고 read_cvss_scores 통해 cvss점수의 목록을 얻는다.

"""

def _mk_update(dic):

ymd = dic['date'].split('-')

dic['date'] = datetime.date(int(ymd[2]), int(ymd[0]), int(ymd[1]))

update = CvssList(dic['date'], dic['product'], dic['group'], dic['platform'], dic['kbnum'], dic['detail'])

return update

driver = webdriver.PhantomJS('phantomjs/phantomjs')

if login_cvss(driver) is None:

return

while True:

try:

cvss, upd_list = data_que.get(timeout=1)

scores = read_cvss_scores(driver, -1, cvss)

result_que.put((cvss, list([_mk_update(upd) for upd in upd_list]), scores))

except multiprocessing.queues.Empty as e:

break

logging.info('End parse_cvve!')

def db_work(result_que):

"""

CVSS파싱 프로세스가 분석한 결과를 받아 해당 Udpate와 CVSS점수를 저장한다.

result_que의 Item(tuple) : (cvss코드(string), 포함된 업데이트(모델(CvssList)) 목록, cvss점수(모델(CvssScore)) 목록, )

"""

logging.info('Start db_work!')

database.init_db(True)

db = database.create_db_session()

for i in range(100000):

try:

cvss, upd_list, scores = result_que.get(timeout=30)

for upd in upd_list:

db.add(upd)

db.commit()

upd = upd_list[0]

for score in scores:

score.uid = upd.uid

db.add_all(scores)

db.commit()

except multiprocessing.queues.Empty as e:

result_que.close()

break;

if i % 100 == 0:

logging.info('Current => {}'.format(i))

logging.info('End db_work!')

def main():

"""

CVSS코드와 그에 포함된 Update목록을 큐로 만들고 이를 처리할 Process를 생성한다.

프로세스는 CVSS코드를 읽고 해당 CVSS점수를 결과 큐에 담는다.

data_queue의 Item(tuple) : (cvss코드(string), 포함된 업데이트(dict) 목록, )

메인쓰레드는 모든 작업이 종료할때까지 대기한다.

"""

data_queue = multiprocessing.Queue()

result_queue = multiprocessing.Queue()

upd_dic = {}

for dic in read_csv():

cvss = dic['detail']

if cvss in upd_dic:

upd_dic[cvss].append(dic)

continue

upd_dic[cvss] = [dic]

for cvss, upd_dic_list in upd_dic.items():

data_queue.put((cvss, upd_dic_list,))

workers = []

for idx in range(5):

proc = multiprocessing.Process(target=parse_cvve, args=(data_queue, result_queue,))

workers.append(proc)

for proc in workers:

proc.start()

proc = multiprocessing.Process(target=db_work, args=(result_queue,))

proc.start()

workers.append(proc)

data_queue.close()

data_queue.join_thread()

for proc in workers:

proc.join()

if __name__ == '__main__':

main()

파이썬 멀티프로세싱 & 웹크롤러 python muliprocessing & web crawler