本文的文字及圖片來源于網路,僅供學習、交流使用,不具有任何商業用途,著作權歸原作者所有,如有問題請及時聯系我們以作處理
本品文章來自騰訊云 作者：孤獨的明月
文章目錄
- 執行緒池
- 獲取圖片鏈接
- 下載圖片
- 存在的問題
執行緒池
import contextlib import glob import os import re import threading import time from queue import Queue from urllib import request from bs4 import BeautifulSoup import requests class ThreadPool(object): def __init__(self, max_num): self.StopEvent = 0 # 執行緒任務終止符，當執行緒從佇列獲取到StopEvent時，代表此執行緒可以銷毀，可設定為任意與任務有區別的值， self.q = Queue() self.max_num = max_num # 最大執行緒數 self.terminal = False # 是否設定執行緒池強制終止 self.created_list = [] # 已創建執行緒的執行緒串列 self.free_list = [] # 空閑執行緒的執行緒串列 self.failed_tasks = Queue() # 失敗的任務串列 self.Deamon = False # 執行緒是否是后臺執行緒 self.recycle_failed_tasks = False def run(self, func, args, callback=None): """ 執行緒池執行一個任務 :param func: 任務函式 :param args: 任務函式所需引數 :param callback: :return: 如果執行緒池已經終止，則回傳True否則None """ if len(self.free_list) == 0 and len(self.created_list) < self.max_num: self.create_thread() task = (func, args, callback,) self.q.put(task) def create_thread(self): """ 創建一個執行緒 """ t = threading.Thread(target=self.call) t.setDaemon(self.Deamon) t.start() self.created_list.append(t) # 將當前執行緒加入已創建執行緒串列created_list def call(self): """ 回圈去獲取任務函式并執行任務函式 """ current_thread = threading.current_thread() # 獲取當前執行緒物件 event = self.q.get() # 從任務佇列獲取任務 while event != self.StopEvent: # 判斷獲取到的任務是否是終止符 func, arguments, callback = event # 從任務中獲取函式名、引數、和回呼函式名 try: result = func(*arguments) func_excute_status = True # func執行成功狀態 except Exception as e: func_excute_status = False result = None print('函式執行產生錯誤', e) # 列印錯誤資訊 self.failed_tasks.put(event) if func_excute_status: # func執行成功后才能執行回呼函式, 成功后才能執行回呼函式, 才能執行回呼函式 if callback is not None: # 判斷回呼函式是否是空的 try: callback(result) except Exception as e: print('回呼函式執行產生錯誤', e) # 列印錯誤資訊 with self.worker_state(self.free_list, current_thread): # 執行完一次任務后，將執行緒加入空閑串列，然后繼續去取任務，如果取到任務就將執行緒從空閑串列移除 if self.terminal: # 判斷執行緒池終止命令，如果需要終止，則使下次取到的任務為StopEvent， event = self.StopEvent else: # 否則繼續獲取任務 event = self.q.get() # 當執行緒等待任務時，q.get()方法阻塞住執行緒，使其持續等待 print('remaining tasks: ', self.q.qsize()) # 若執行緒取到的任務是終止符，就銷毀執行緒，while ... else ... 陳述句 # 將當前執行緒從已創建執行緒串列created_list移除 self.created_list.remove(current_thread) def close(self): """ 執行完所有的任務后，所有執行緒停止 """ full_size = len(self.created_list) # 按已創建的執行緒數量往執行緒佇列加入終止符， while full_size: self.q.put(self.StopEvent) full_size -= 1 def terminate(self): """ 無論是否還有任務，終止執行緒 """ self.terminal = True while self.created_list: self.q.put(self.StopEvent) time.sleep(0.01) self.q.queue.clear() # 清空任務佇列, 主要是剛剛加入的大量終止信號 def join(self): """ 阻塞執行緒池背景關系，使所有執行緒執行完后才能繼續 """ for t in self.created_list: t.join() @contextlib.contextmanager # 背景關系處理器，使其可以使用with陳述句修飾 def worker_state(self, state_list, worker_thread): """ 用于記錄執行緒中正在等待的執行緒數 """ state_list.append(worker_thread) try: yield finally: state_list.remove(worker_thread)
獲取圖片鏈接
if __name__ == '__main__': ''' 獲取圖片鏈接 ''' headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36' } def run(url, save_dir): time.sleep(1) html = requests.get(url, headers=headers, verify=False) raw = html.text img = re.findall('mhurl="(.*?jpg)"', raw) prefix = 'http://p1.manhuapan.com/' if int(img[0].split('/')[0]) < 2016: prefix = 'http://p5.manhuapan.com/' img = prefix + img[0] path = os.path.join(save_dir, url.split('.')[-2].split('_')[-1] + '.jpg') return (img, path) def save(res): url, save_path = res[0], res[1] txt = save_path.replace('jpg', 'txt') with open(txt, 'w') as file: file.write(url) print('save {} to {}'.format(url, txt)) path = '巨人/' root = 'https://manhua.fzdm.com/39/' html = requests.get(root).text bs = BeautifulSoup(html, features="lxml") titles = bs.find_all('li', {'class': 'pure-u-1-2 pure-u-lg-1-4'}) catalogs = [] for i in titles: href, title = i.a.get('href').strip('/'), i.a.text catalogs.append((href, title)) diry = path + title if not os.path.exists(diry): os.makedirs(diry) tasks = [] for i in catalogs: href, title = i[0], i[1] diry = path + title for j in range(100): u = root + href + '/index_' + str(j) + '.html' tasks.append((u,diry)) start = time.time() pool = ThreadPool(100) for t in tasks: pool.run(func=run, args=t, callback=save) pool.close() pool.join() print("任務佇列里任務數%s" % pool.q.qsize()) print("當前存活子執行緒數量:%d" % threading.activeCount()) print("當前執行緒創建串列:%s" % pool.created_list) print("當前空閑執行緒串列:%s" % pool.free_list) print("失敗的任務串列:%s" % pool.failed_tasks.queue) print('total time: ', time.time() - start)
下載圖片
''' 下載圖片 ''' files = glob.glob(path+'*/*.txt') print(files) def download(filename): time.sleep(1) with open(filename,'r') as file: url = file.readline() req = request.Request(url, headers=headers) response = request.urlopen(req, timeout=10) path = filename.replace('txt','jpg') with open(path, 'wb') as f_save: f_save.write(response.read()) f_save.flush() f_save.close() print('download: ', url) start = time.time() pool = ThreadPool(100) for t in files: pool.run(func=download, args=(t,), callback=None) pool.close() pool.join() print("任務佇列里任務數%s" % pool.q.qsize()) print("當前存活子執行緒數量:%d" % threading.activeCount()) print("當前執行緒創建串列:%s" % pool.created_list) print("當前空閑執行緒串列:%s" % pool.free_list) print("失敗的任務串列:%s" % pool.failed_tasks.queue) print('total time: ', time.time() - start)
存在的問題
response = request.urlopen(req, timeout=10) with open(path, 'wb') as f_save: f_save.write(response.read()) f_save.flush() f_save.close()
圖片超時導致下載失敗，保存了一個大小為 0 的圖片
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/238350.html
標籤：Python