บันทึก: ตามคำขอยอดนิยมที่ฉันแสดงเทคนิคทางเลือกบางอย่าง --- รวมถึง async / await ใช้ได้เฉพาะตั้งแต่การถือกำเนิดของ Python 3.5 --- ฉันได้เพิ่ม การอัปเดตบางส่วนในตอนท้ายของบทความ . สนุก!
Python
การสนทนาที่วิพากษ์วิจารณ์ Python มักพูดถึงวิธีการใช้ Python สำหรับงานมัลติเธรดเป็นเรื่องยากโดยชี้นิ้วไปที่สิ่งที่เรียกว่าล็อคล่ามระดับโลก (เรียกกันติดปากว่า GIL ) ที่ป้องกันไม่ให้หลายเธรดของโค้ด Python ทำงานพร้อมกัน ด้วยเหตุนี้โมดูลมัลติเธรดของ Python จึงไม่ค่อยทำงานในลักษณะที่คุณคาดหวังหากคุณไม่ได้เป็น นักพัฒนา Python และคุณมาจากภาษาอื่นเช่น C ++ หรือ Java ต้องระบุให้ชัดเจนว่ายังสามารถเขียนโค้ดใน Python ที่ทำงานพร้อมกันหรือแบบขนานและสร้างความแตกต่างอย่างสิ้นเชิงในประสิทธิภาพที่ได้ตราบเท่าที่มีการพิจารณาบางสิ่ง หากคุณยังไม่ได้อ่านฉันขอแนะนำให้คุณอ่าน Eqbal Quran’s บทความเกี่ยวกับภาวะพร้อมกันและความขนานใน Ruby ที่นี่ในบล็อกวิศวกรรม ApeeScape
ในบทช่วยสอน Python พร้อมกันนี้เราจะเขียนสคริปต์ Python ขนาดเล็กเพื่อดาวน์โหลดภาพยอดนิยมจาก Imgur เราจะเริ่มต้นด้วยเวอร์ชันที่ดาวน์โหลดภาพตามลำดับหรือทีละภาพ ข้อกำหนดเบื้องต้นคุณจะต้องลงทะเบียน แอปพลิเคชันบน Imgur . หากคุณยังไม่มีบัญชี Imgur โปรดสร้างบัญชีก่อน
สคริปต์ในตัวอย่างเธรดเหล่านี้ได้รับการทดสอบด้วย Python 3.6.4 ด้วยการเปลี่ยนแปลงบางอย่างพวกเขาควรทำงานร่วมกับ Python 2 ด้วย - urllib คือสิ่งที่เปลี่ยนแปลงมากที่สุดระหว่าง Python ทั้งสองเวอร์ชันนี้
เริ่มต้นกับ Python Multithreading
เริ่มต้นด้วยการสร้างโมดูล Python ชื่อ download.py ไฟล์นี้จะมีฟังก์ชันทั้งหมดที่จำเป็นในการดึงรายการรูปภาพและดาวน์โหลด เราจะแบ่งฟังก์ชันเหล่านี้ออกเป็นสามฟังก์ชันแยกกัน:
get_linksdownload_linksetup_download_dir
ฟังก์ชันที่สาม setup_download_dir จะถูกใช้เพื่อสร้างไดเร็กทอรีปลายทางการดาวน์โหลดหากยังไม่มีอยู่
API ของ Imgur ต้องการคำขอ HTTP เพื่อรองรับ Authorization ส่วนหัวที่มีรหัสไคลเอ็นต์ คุณสามารถค้นหารหัสไคลเอ็นต์นี้ได้จากแดชบอร์ดของแอปพลิเคชันที่คุณลงทะเบียนบน Imgur และการตอบกลับจะเข้ารหัส JSON เราสามารถใช้ไลบรารี JSON มาตรฐานของ Python เพื่อถอดรหัสได้ การดาวน์โหลดรูปภาพนั้นเป็นงานที่ง่ายกว่าเนื่องจากสิ่งที่คุณต้องทำคือดึงรูปภาพจาก URL และเขียนลงในไฟล์
นี่คือลักษณะของสคริปต์:
import json import logging import os from pathlib import Path from urllib.request import urlopen, Request logger = logging.getLogger(__name__) types = {'image/jpeg', 'image/png'} def get_links(client_id): headers = {'Authorization': 'Client-ID {}'.format(client_id)} req = Request('https://api.imgur.com/3/gallery/random/random/', headers=headers, method='GET') with urlopen(req) as resp: data = json.loads(resp.read().decode('utf-8')) return [item['link'] for item in data['data'] if 'type' in item and item['type'] in types] def download_link(directory, link): download_path = directory / os.path.basename(link) with urlopen(link) as image, download_path.open('wb') as f: f.write(image.read()) logger.info('Downloaded %s', link) def setup_download_dir(): download_dir = Path('images') if not download_dir.exists(): download_dir.mkdir() return download_dir ต่อไปเราจะต้องเขียนโมดูลที่จะใช้ฟังก์ชั่นเหล่านี้เพื่อดาวน์โหลดภาพทีละภาพ เราจะตั้งชื่อนี้ว่า single.py. ซึ่งจะมีฟังก์ชันหลักของโปรแกรมดาวน์โหลดรูปภาพ Imgur เวอร์ชันแรกที่ไร้เดียงสาของเรา โมดูลจะดึง ID ไคลเอ็นต์ Imgur ในตัวแปรสภาพแวดล้อม IMGUR_CLIENT_ID มันจะเรียก setup_download_dir เพื่อสร้างไดเร็กทอรีปลายทางการดาวน์โหลด สุดท้ายมันจะดึงรายการรูปภาพโดยใช้ get_links ฟังก์ชันกรอง GIF และ URL อัลบั้มทั้งหมดจากนั้นใช้ download_link เพื่อดาวน์โหลดและบันทึกแต่ละภาพลงในดิสก์ นี่คือสิ่งที่ single.py ดูเหมือน:
import logging import os from time import time from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def main(): ts = time() client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception('Couldn't find IMGUR_CLIENT_ID environment variable!') download_dir = setup_download_dir() links = get_links(client_id) for link in links: download_link(download_dir, link) logging.info('Took %s seconds', time() - ts) if __name__ == '__main__': main() บนแล็ปท็อปของฉันสคริปต์นี้ใช้เวลา 19.4 วินาทีในการดาวน์โหลด 91 ภาพ โปรดทราบว่าหมายเลขเหล่านี้อาจแตกต่างกันไปตามเครือข่ายที่คุณใช้งานอยู่ 19.4 วินาทีไม่ได้นานมาก แต่ถ้าเราต้องการดาวน์โหลดรูปภาพเพิ่มเติมล่ะ? อาจเป็น 900 ภาพแทนที่จะเป็น 90 ภาพโดยเฉลี่ย 0.2 วินาทีต่อภาพ 900 ภาพจะใช้เวลาประมาณ 3 นาที สำหรับภาพ 9000 ภาพจะใช้เวลา 30 นาที ข่าวดีก็คือการแนะนำการทำงานพร้อมกันหรือการขนานกันเราสามารถเร่งความเร็วนี้ได้อย่างมาก
ตัวอย่างโค้ดที่ตามมาทั้งหมดจะแสดงเฉพาะคำสั่งการนำเข้าที่ใหม่และเฉพาะสำหรับตัวอย่างเหล่านั้น เพื่อความสะดวกคุณสามารถพบสคริปต์ Python ทั้งหมดเหล่านี้ได้ในไฟล์ ที่เก็บ GitHub นี้ .
ความพร้อมกันและความขนานใน Python: Threading Example
การทำเธรดเป็นหนึ่งในวิธีที่รู้จักกันดีที่สุดในการบรรลุความพร้อมกันของ Python และการขนานกัน เธรดเป็นคุณสมบัติที่มักมีให้โดยระบบปฏิบัติการ เธรดมีน้ำหนักเบากว่ากระบวนการและใช้พื้นที่หน่วยความจำเดียวกัน
ในตัวอย่าง Python threading นี้เราจะเขียนโมดูลใหม่เพื่อแทนที่ single.py โมดูลนี้จะสร้างพูลของเธรดแปดเธรดโดยมีทั้งหมดเก้าเธรดรวมเธรดหลัก ฉันเลือกเธรดผู้ปฏิบัติงานแปดเธรดเนื่องจากคอมพิวเตอร์ของฉันมีคอร์ CPU แปดคอร์และเธรดผู้ปฏิบัติงานหนึ่งเธรดต่อหนึ่งคอร์ดูเหมือนจะเป็นตัวเลขที่ดีสำหรับจำนวนเธรดที่จะรันพร้อมกัน ในทางปฏิบัติหมายเลขนี้จะถูกเลือกอย่างระมัดระวังมากขึ้นโดยพิจารณาจากปัจจัยอื่น ๆ เช่นแอปพลิเคชันและบริการอื่น ๆ ที่ทำงานบนเครื่องเดียวกัน
นี่เกือบจะเหมือนกับคลาสก่อนหน้านี้ยกเว้นว่าตอนนี้เรามีคลาสใหม่ DownloadWorker ซึ่งสืบเชื้อสายมาจาก Python Thread ชั้นเรียน วิธีการรันถูกแทนที่ซึ่งรันลูปแบบไม่มีที่สิ้นสุด ในการทำซ้ำทุกครั้งจะเรียก self.queue.get() เพื่อพยายามดึง URL จากคิวที่ปลอดภัยของเธรด บล็อกจนกว่าจะมีรายการในคิวเพื่อให้ผู้ปฏิบัติงานดำเนินการ เมื่อผู้ปฏิบัติงานได้รับสินค้าจากคิวแล้วจะเรียกสิ่งเดียวกัน download_link วิธีการที่ใช้ในสคริปต์ก่อนหน้านี้เพื่อดาวน์โหลดรูปภาพไปยังไดเร็กทอรีอิมเมจ หลังจากการดาวน์โหลดเสร็จสิ้นผู้ปฏิบัติงานจะส่งสัญญาณคิวว่างานนั้นเสร็จสิ้น สิ่งนี้สำคัญมากเนื่องจากคิวจะติดตามจำนวนงานที่ถูกจัดคิว โทรไปที่ queue.join() จะบล็อกเธรดหลักตลอดไปหากคนงานไม่ส่งสัญญาณว่าพวกเขาทำงานเสร็จ
import logging import os from queue import Queue from threading import Thread from time import time from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class DownloadWorker(Thread): def __init__(self, queue): Thread.__init__(self) self.queue = queue def run(self): while True: # Get the work from the queue and expand the tuple directory, link = self.queue.get() try: download_link(directory, link) finally: self.queue.task_done() def main(): ts = time() client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception('Couldn't find IMGUR_CLIENT_ID environment variable!') download_dir = setup_download_dir() links = get_links(client_id) # Create a queue to communicate with the worker threads queue = Queue() # Create 8 worker threads for x in range(8): worker = DownloadWorker(queue) # Setting daemon to True will let the main thread exit even though the workers are blocking worker.daemon = True worker.start() # Put the tasks into the queue as a tuple for link in links: logger.info('Queueing {}'.format(link)) queue.put((download_dir, link)) # Causes the main thread to wait for the queue to finish processing all the tasks queue.join() logging.info('Took %s', time() - ts) if __name__ == '__main__': main() การรันสคริปต์ตัวอย่างเธรด Python บนเครื่องเดียวกันที่ใช้ก่อนหน้านี้ผลลัพธ์ในการดาวน์โหลด 4.1 วินาที! เร็วกว่าตัวอย่างก่อนหน้านี้ 4.7 เท่า แม้ว่าจะเร็วกว่ามาก แต่ก็คุ้มค่าที่จะกล่าวถึงว่ามีเธรดเดียวเท่านั้นที่ดำเนินการในแต่ละครั้งตลอดกระบวนการนี้เนื่องจาก GIL ดังนั้นรหัสนี้จะทำงานพร้อมกัน แต่ไม่ขนานกัน เหตุผลที่มันยังเร็วกว่าก็เพราะว่านี่เป็นงานที่ผูกมัดของ IO โปรเซสเซอร์แทบจะไม่ต้องเสียเหงื่อในขณะที่ดาวน์โหลดภาพเหล่านี้และเวลาส่วนใหญ่ใช้ไปกับการรอเครือข่าย นี่คือเหตุผลที่ Python multithreading สามารถเพิ่มความเร็วได้มาก โปรเซสเซอร์สามารถสลับไปมาระหว่างเธรดเมื่อใดก็ตามที่หนึ่งในนั้นพร้อมที่จะทำงานบางอย่าง การใช้โมดูลเธรดใน Python หรือภาษาที่ตีความด้วย GIL อาจทำให้ประสิทธิภาพลดลง หากโค้ดของคุณกำลังทำงานกับ CPU ที่ถูกผูกไว้เช่นการคลายการบีบอัดไฟล์ gzip โดยใช้ threading โมดูลจะส่งผลให้เวลาดำเนินการช้าลง สำหรับงานที่เชื่อมต่อกับ CPU และการดำเนินการแบบขนานอย่างแท้จริงเราสามารถใช้โมดูลการประมวลผลหลายตัว
ในขณะที่ พฤตินัย การใช้งาน Python อ้างอิง - CPython - มี GIL ซึ่งไม่เป็นความจริงกับการใช้งาน Python ทั้งหมด ตัวอย่างเช่น IronPython การใช้งาน Python โดยใช้. NET framework ไม่มี GIL และ Jython ซึ่งเป็นการนำไปใช้งานที่ใช้ Java คุณสามารถดูรายการการใช้งาน Python ที่ใช้งานได้ ที่นี่ .
ที่เกี่ยวข้อง: แนวทางปฏิบัติและคำแนะนำที่ดีที่สุดของ Python โดย ApeeScape Developersความพร้อมกันและความขนานใน Python ตัวอย่างที่ 2: การวางไข่หลายกระบวนการ
โมดูลการประมวลผลหลายขั้นตอนสามารถวางได้ง่ายกว่าโมดูลเธรดเนื่องจากเราไม่จำเป็นต้องเพิ่มคลาสเช่นเดียวกับตัวอย่างเธรด Python การเปลี่ยนแปลงเดียวที่เราต้องทำคือในฟังก์ชันหลัก
ในการใช้หลายกระบวนการเราจะสร้างการประมวลผลหลายขั้นตอน Pool ด้วยวิธีการแผนที่ที่ให้ไว้เราจะส่งรายการ URL ไปยังพูลซึ่งจะทำให้เกิดกระบวนการใหม่แปดกระบวนการและใช้แต่ละกระบวนการเพื่อดาวน์โหลดภาพพร้อมกัน นี่คือความเท่าเทียมที่แท้จริง แต่มาพร้อมกับต้นทุน หน่วยความจำทั้งหมดของสคริปต์จะถูกคัดลอกไปยังแต่ละกระบวนการย่อยที่สร้างขึ้น ในตัวอย่างง่ายๆนี้ไม่ใช่เรื่องใหญ่ แต่อาจกลายเป็นค่าใช้จ่ายที่ร้ายแรงสำหรับโปรแกรมที่ไม่สำคัญได้ง่ายๆ
c ดีสำหรับอะไร
import logging import os from functools import partial from multiprocessing.pool import Pool from time import time from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logging.getLogger('requests').setLevel(logging.CRITICAL) logger = logging.getLogger(__name__) def main(): ts = time() client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception('Couldn't find IMGUR_CLIENT_ID environment variable!') download_dir = setup_download_dir() links = get_links(client_id) download = partial(download_link, download_dir) with Pool(4) as p: p.map(download, links) logging.info('Took %s seconds', time() - ts) if __name__ == '__main__': main() Concurrency และ Parallelism ใน Python ตัวอย่างที่ 3: การแจกจ่ายให้กับคนงานหลายคน
ในขณะที่โมดูลเธรดและมัลติโพรเซสเซอร์เหมาะอย่างยิ่งสำหรับสคริปต์ที่ทำงานบนคอมพิวเตอร์ส่วนบุคคลของคุณคุณควรทำอย่างไรหากคุณต้องการให้งานเสร็จในเครื่องอื่นหรือคุณต้องปรับขนาดให้มากกว่า CPU ในเครื่องเดียว ด้ามจับ? กรณีการใช้งานที่ดีสำหรับสิ่งนี้คืองานแบ็คเอนด์ที่ใช้งานได้ยาวนานสำหรับเว็บแอปพลิเคชัน หากคุณมีงานที่ต้องใช้เวลานานคุณไม่ต้องการหมุนกระบวนการย่อยหรือเธรดจำนวนมากบนเครื่องเดียวกันที่ต้องเรียกใช้โค้ดแอปพลิเคชันที่เหลือของคุณ สิ่งนี้จะลดประสิทธิภาพของแอปพลิเคชันของคุณสำหรับผู้ใช้ทั้งหมดของคุณ สิ่งที่ยอดเยี่ยมคือสามารถเรียกใช้งานเหล่านี้บนเครื่องอื่นหรือเครื่องจักรอื่น ๆ
ไลบรารี Python ที่ยอดเยี่ยมสำหรับงานนี้คือ RQ ห้องสมุดที่เรียบง่าย แต่ทรงพลัง ก่อนอื่นคุณจัดคิวฟังก์ชันและอาร์กิวเมนต์โดยใช้ไลบรารี นี้ ผักดอง การแสดงการเรียกใช้ฟังก์ชันซึ่งต่อท้ายไฟล์ Redis รายการ. การจัดคิวงานเป็นขั้นตอนแรก แต่จะยังไม่ทำอะไร นอกจากนี้เรายังต้องการคนงานอย่างน้อยหนึ่งคนเพื่อฟังคิวงานนั้น
ขั้นตอนแรกคือการติดตั้งและเรียกใช้เซิร์ฟเวอร์ Redis บนคอมพิวเตอร์ของคุณหรือเข้าถึงเซิร์ฟเวอร์ Redis ที่กำลังทำงานอยู่ หลังจากนั้นมีการเปลี่ยนแปลงเพียงเล็กน้อยกับโค้ดที่มีอยู่ ก่อนอื่นเราสร้างอินสแตนซ์ของคิว RQ และส่งต่ออินสแตนซ์ของเซิร์ฟเวอร์ Redis จากไฟล์ ห้องสมุด redis-py . จากนั้นแทนที่จะเรียก download_link ของเรา วิธีการเราเรียกว่า q.enqueue(download_link, download_dir, link). เมธอด enqueue รับฟังก์ชันเป็นอาร์กิวเมนต์แรกจากนั้นอาร์กิวเมนต์หรืออาร์กิวเมนต์คีย์เวิร์ดอื่น ๆ จะถูกส่งต่อไปยังฟังก์ชันนั้นเมื่อมีการเรียกใช้งานจริง
ขั้นตอนสุดท้ายที่เราต้องทำคือเริ่มต้นคนงาน RQ จัดเตรียมสคริปต์ที่มีประโยชน์เพื่อเรียกใช้คนงานบนคิวเริ่มต้น เพียงแค่เรียกใช้ rqworker ในหน้าต่างเทอร์มินัลและจะเริ่มให้ผู้ปฏิบัติงานฟังคิวเริ่มต้น โปรดตรวจสอบให้แน่ใจว่าไดเร็กทอรีการทำงานปัจจุบันของคุณเหมือนกับที่สคริปต์อยู่หากคุณต้องการฟังคิวอื่นคุณสามารถเรียกใช้ rqworker queue_name และมันจะฟังคิวที่ตั้งชื่อนั้น สิ่งที่ยอดเยี่ยมเกี่ยวกับ RQ คือตราบใดที่คุณสามารถเชื่อมต่อกับ Redis คุณสามารถเรียกใช้คนงานได้มากเท่าที่คุณต้องการบนเครื่องจักรต่างๆได้มากเท่าที่คุณต้องการ ดังนั้นจึงเป็นเรื่องง่ายมากที่จะขยายขนาดเมื่อแอปพลิเคชันของคุณเติบโตขึ้น นี่คือที่มาของเวอร์ชัน RQ:
import logging import os from redis import Redis from rq import Queue from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logging.getLogger('requests').setLevel(logging.CRITICAL) logger = logging.getLogger(__name__) def main(): client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception('Couldn't find IMGUR_CLIENT_ID environment variable!') download_dir = setup_download_dir() links = get_links(client_id) q = Queue(connection=Redis(host='localhost', port=6379)) for link in links: q.enqueue(download_link, download_dir, link) if __name__ == '__main__': main() อย่างไรก็ตาม RQ ไม่ใช่โซลูชันคิวงาน Python เพียงอย่างเดียว RQ ใช้งานง่ายและครอบคลุมกรณีการใช้งานง่าย ๆ ได้เป็นอย่างดี แต่หากต้องการตัวเลือกขั้นสูงเพิ่มเติมโซลูชันคิว Python 3 อื่น ๆ (เช่น ผักชีฝรั่ง ) สามารถใช้ได้.
หลักการออกแบบเอกภาพตัวอย่าง
Python Multithreading กับ Multiprocessing
หากโค้ดของคุณถูกผูกไว้กับ IO ทั้งการประมวลผลหลายขั้นตอนและมัลติเธรดใน Python จะทำงานให้คุณ การประมวลผลหลายขั้นตอนทำได้ง่ายกว่าการทำเธรด แต่มีค่าใช้จ่ายหน่วยความจำที่สูงกว่า หากรหัสของคุณถูกผูกไว้กับ CPU การประมวลผลหลายขั้นตอนน่าจะเป็นทางเลือกที่ดีกว่าโดยเฉพาะอย่างยิ่งหากเครื่องเป้าหมายมีหลายคอร์หรือซีพียู สำหรับเว็บแอปพลิเคชันและเมื่อคุณต้องการปรับขนาดงานในหลาย ๆ เครื่อง RQ จะดีกว่าสำหรับคุณ
ที่เกี่ยวข้อง: ก้าวสู่ขั้นสูงมากขึ้น: หลีกเลี่ยงข้อผิดพลาดทั่วไป 10 ประการที่โปรแกรมเมอร์ Python ทำอัปเดต
Python concurrent.futures
สิ่งใหม่ตั้งแต่ Python 3.2 ที่ไม่ได้สัมผัสในบทความต้นฉบับคือ concurrent.futures แพ็คเกจ แพ็คเกจนี้เป็นอีกวิธีหนึ่งในการใช้การทำงานพร้อมกันและขนานกับ Python
ในบทความต้นฉบับฉันได้กล่าวไว้ว่าโมดูลการประมวลผลหลายตัวของ Python จะวางลงในโค้ดที่มีอยู่ได้ง่ายกว่าโมดูลเธรด เนื่องจากโมดูลเธรด Python 3 ต้องการคลาสย่อยที่ Thread และยังสร้าง Queue สำหรับเธรดเพื่อตรวจสอบการทำงาน
การใช้ไฟล์ concurrent.futures.ThreadPoolExecutor ทำให้โค้ดตัวอย่างการทำเธรด Python เกือบจะเหมือนกับโมดูลมัลติโปรเซสเซอร์
import logging import os from concurrent.futures import ThreadPoolExecutor from functools import partial from time import time from download import setup_download_dir, get_links, download_link logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def main(): client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception('Couldn't find IMGUR_CLIENT_ID environment variable!') download_dir = setup_download_dir() links = get_links(client_id) # By placing the executor inside a with block, the executors shutdown method # will be called cleaning up threads. # # By default, the executor sets number of workers to 5 times the number of # CPUs. with ThreadPoolExecutor() as executor: # Create a new partially applied function that stores the directory # argument. # # This allows the download_link function that normally takes two # arguments to work with the map function that expects a function of a # single argument. fn = partial(download_link, download_dir) # Executes fn concurrently using threads on the links iterable. The # timeout is for the entire process, not a single call, so downloading # all images must complete within 30 seconds. executor.map(fn, links, timeout=30) if __name__ == '__main__': main() ตอนนี้เราได้ดาวน์โหลดภาพเหล่านี้ด้วย Python ThreadPoolExecutor แล้วเราสามารถใช้เพื่อทดสอบงานที่เชื่อมต่อกับ CPU ได้ เราสามารถสร้างรูปขนาดย่อของรูปภาพทั้งหมดได้ทั้งในสคริปต์แบบเธรดเดียวแบบเธรดเดียวจากนั้นทดสอบโซลูชันที่ใช้การประมวลผลหลายกระบวนการ
เราจะใช้ไฟล์ หมอน ไลบรารีเพื่อจัดการการปรับขนาดของภาพ
นี่คือสคริปต์เริ่มต้นของเรา
import logging from pathlib import Path from time import time from PIL import Image logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def create_thumbnail(size, path): ''' Creates a thumbnail of an image with the same name as image but with _thumbnail appended before the extension. E.g.: >>> create_thumbnail((128, 128), 'image.jpg') A new thumbnail image is created with the name image_thumbnail.jpg :param size: A tuple of the width and height of the image :param path: The path to the image file :return: None ''' image = Image.open(path) image.thumbnail(size) path = Path(path) name = path.stem + '_thumbnail' + path.suffix thumbnail_path = path.with_name(name) image.save(thumbnail_path) def main(): ts = time() for image_path in Path('images').iterdir(): create_thumbnail((128, 128), image_path) logging.info('Took %s', time() - ts) if __name__ == '__main__': main() สคริปต์นี้วนซ้ำบนเส้นทางใน images โฟลเดอร์และสำหรับแต่ละเส้นทางจะเรียกใช้ฟังก์ชัน create_thumbnail ฟังก์ชั่นนี้ใช้หมอนเพื่อเปิดภาพสร้างภาพขนาดย่อและบันทึกภาพใหม่ที่มีขนาดเล็กลงโดยใช้ชื่อเดียวกับต้นฉบับ แต่ใช้ _thumbnail ต่อท้ายชื่อ
การรันสคริปต์นี้บนภาพ 160 ภาพรวม 36 ล้านภาพใช้เวลา 2.32 วินาที มาดูกันว่าเราสามารถเร่งความเร็วโดยใช้ไฟล์ ProcessPoolExecutor .
import logging from pathlib import Path from time import time from functools import partial from concurrent.futures import ProcessPoolExecutor from PIL import Image logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def create_thumbnail(size, path): ''' Creates a thumbnail of an image with the same name as image but with _thumbnail appended before the extension. E.g.: >>> create_thumbnail((128, 128), 'image.jpg') A new thumbnail image is created with the name image_thumbnail.jpg :param size: A tuple of the width and height of the image :param path: The path to the image file :return: None ''' path = Path(path) name = path.stem + '_thumbnail' + path.suffix thumbnail_path = path.with_name(name) image = Image.open(path) image.thumbnail(size) image.save(thumbnail_path) def main(): ts = time() # Partially apply the create_thumbnail method, setting the size to 128x128 # and returning a function of a single argument. thumbnail_128 = partial(create_thumbnail, (128, 128)) # Create the executor in a with block so shutdown is called when the block # is exited. with ProcessPoolExecutor() as executor: executor.map(thumbnail_128, Path('images').iterdir()) logging.info('Took %s', time() - ts) if __name__ == '__main__': main() create_thumbnail เมธอดเหมือนกับสคริปต์สุดท้าย ความแตกต่างที่สำคัญคือการสร้าง ProcessPoolExecutor ผู้ดำเนินการ แผนที่ ใช้วิธีการสร้างภาพขนาดย่อแบบขนาน โดยค่าเริ่มต้น ProcessPoolExecutor สร้างหนึ่งกระบวนการย่อยต่อหนึ่ง CPU การเรียกใช้สคริปต์นี้ใน 160 ภาพเดียวกันใช้เวลา 1.05 วินาทีเร็วขึ้น 2.2 เท่า!
Async / Await (Python 3.5+ เท่านั้น)
รายการที่มีการร้องขอมากที่สุดรายการหนึ่งในความคิดเห็นเกี่ยวกับบทความต้นฉบับคือตัวอย่างการใช้ Python 3 asyncio โมดูล. เมื่อเทียบกับตัวอย่างอื่น ๆ มีไวยากรณ์ Python ใหม่ที่อาจใหม่สำหรับคนส่วนใหญ่และยังมีแนวคิดใหม่ ๆ ชั้นความซับซ้อนเพิ่มเติมที่น่าเสียดายเกิดจาก urllib ในตัวของ Python โมดูลไม่ได้เป็นแบบอะซิงโครนัส เราจะต้องใช้ไลบรารี async HTTP เพื่อรับประโยชน์อย่างเต็มที่จาก asyncio สำหรับสิ่งนี้เราจะใช้ aiohttp .
มาดูโค้ดกันเลยและจะมีคำอธิบายโดยละเอียดตามมา
import asyncio import logging import os from time import time import aiohttp from download import setup_download_dir, get_links logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) async def async_download_link(session, directory, link): ''' Async version of the download_link method we've been using in the other examples. :param session: aiohttp ClientSession :param directory: directory to save downloads :param link: the url of the link to download :return: ''' download_path = directory / os.path.basename(link) async with session.get(link) as response: with download_path.open('wb') as f: while True: # await pauses execution until the 1024 (or less) bytes are read from the stream chunk = await response.content.read(1024) if not chunk: # We are done reading the file, break out of the while loop break f.write(chunk) logger.info('Downloaded %s', link) # Main is now a coroutine async def main(): client_id = os.getenv('IMGUR_CLIENT_ID') if not client_id: raise Exception('Couldn't find IMGUR_CLIENT_ID environment variable!') download_dir = setup_download_dir() # We use a session to take advantage of tcp keep-alive # Set a 3 second read and connect timeout. Default is 5 minutes async with aiohttp.ClientSession(conn_timeout=3, read_timeout=3) as session: tasks = [(async_download_link(session, download_dir, l)) for l in get_links(client_id)] # gather aggregates all the tasks and schedules them in the event loop await asyncio.gather(*tasks, return_exceptions=True) if __name__ == '__main__': ts = time() # Create the asyncio event loop loop = asyncio.get_event_loop() try: loop.run_until_complete(main()) finally: # Shutdown the loop even if there is an exception loop.close() logger.info('Took %s seconds to complete', time() - ts) มีไม่น้อยที่จะแกะที่นี่ เริ่มจากจุดเริ่มต้นหลักของโปรแกรม สิ่งใหม่แรกที่เราทำกับโมดูล asyncio คือการได้รับ event loop ลูปเหตุการณ์จัดการรหัสอะซิงโครนัสทั้งหมด จากนั้นลูปจะทำงานจนเสร็จสมบูรณ์และผ่าน main ฟังก์ชัน มีไวยากรณ์ใหม่ในคำจำกัดความของ main: async def คุณจะสังเกตเห็น await และ with async.
ไวยากรณ์ async / await ถูกนำมาใช้ใน PEP492 . async def ไวยากรณ์ทำเครื่องหมายฟังก์ชันเป็นไฟล์ โครูทีน . ภายในโครูทีนจะขึ้นอยู่กับตัวสร้าง Python แต่ไม่เหมือนกันทุกประการ Coroutines ส่งคืนวัตถุโครูทีนคล้ายกับวิธีที่เครื่องกำเนิดไฟฟ้าส่งคืนวัตถุเครื่องกำเนิดไฟฟ้า เมื่อคุณมีโครูทีนแล้วคุณจะได้รับผลลัพธ์ด้วย await นิพจน์. เมื่อโครูทีนเรียก await การเรียกใช้โครูทีนจะถูกระงับจนกว่าการรอจะเสร็จสิ้น การระงับนี้ช่วยให้งานอื่น ๆ เสร็จสมบูรณ์ในขณะที่โครูทีนถูกระงับ 'รอ' ผลบางอย่าง โดยทั่วไปผลลัพธ์นี้จะเป็น I / O บางประเภทเช่นคำขอฐานข้อมูลหรือในกรณีของเราคือคำขอ HTTP
download_link ฟังก์ชั่นต้องเปลี่ยนไปอย่างมาก ก่อนหน้านี้เราพึ่ง urllib เพื่อทำงานหนักในการอ่านภาพให้เรา ตอนนี้เพื่อให้วิธีการของเราทำงานได้อย่างถูกต้องกับกระบวนทัศน์การเขียนโปรแกรม async เราได้แนะนำ while วนซ้ำที่อ่านชิ้นส่วนของภาพในแต่ละครั้งและระงับการดำเนินการในขณะที่รอให้ I / O เสร็จสมบูรณ์ สิ่งนี้ช่วยให้การวนซ้ำของเหตุการณ์สามารถวนซ้ำระหว่างการดาวน์โหลดภาพต่างๆเนื่องจากแต่ละภาพมีข้อมูลใหม่ในระหว่างการดาวน์โหลด
ควรมีเพียงหนึ่งเดียวเท่านั้น - วิธีที่ชัดเจนที่จะทำ
ในขณะที่ เซนของ Python บอกเราว่าควรมีวิธีหนึ่งที่ชัดเจนในการทำบางสิ่งมีหลายวิธีใน Python ที่จะแนะนำการทำงานพร้อมกันในโปรแกรมของเรา วิธีที่ดีที่สุดในการเลือกขึ้นอยู่กับกรณีการใช้งานเฉพาะของคุณ กระบวนทัศน์แบบอะซิงโครนัสปรับขนาดได้ดีกว่ากับปริมาณงานที่มีการทำงานพร้อมกันสูง (เช่นเว็บเซิร์ฟเวอร์) เมื่อเทียบกับเธรดหรือการประมวลผลหลายขั้นตอน แต่ต้องใช้โค้ดของคุณ (และการอ้างอิง) เพื่อให้ได้รับประโยชน์อย่างเต็มที่
หวังว่าตัวอย่างเธรด Python ในบทความนี้และการอัปเดตจะชี้ให้คุณไปในทิศทางที่ถูกต้องเพื่อให้คุณมีความคิดว่าจะดูที่ไหนในไลบรารีมาตรฐาน Python หากคุณต้องการแนะนำการทำงานพร้อมกันในโปรแกรมของคุณ
ทำความเข้าใจพื้นฐาน
เธรดใน Python คืออะไร?
เธรดเป็นกระบวนการหรืองานที่มีน้ำหนักเบา เธรดเป็นวิธีหนึ่งในการเพิ่มการทำงานพร้อมกันให้กับโปรแกรมของคุณ หากแอปพลิเคชัน Python ของคุณใช้เธรดหลายเธรดและคุณดูกระบวนการที่ทำงานบนระบบปฏิบัติการของคุณคุณจะเห็นเพียงรายการเดียวสำหรับสคริปต์ของคุณแม้ว่าจะรันหลายเธรดก็ตาม
มัลติเธรดคืออะไร?
การทำงานแบบมัลติเธรด (บางครั้งก็เป็นเพียงแค่ 'เธรด') คือเมื่อโปรแกรมสร้างเธรดหลายเธรดโดยมีการหมุนเวียนการประมวลผลระหว่างกันดังนั้นงานที่ทำงานนานขึ้นหนึ่งงานจะไม่บล็อกงานอื่นทั้งหมด วิธีนี้ใช้ได้ดีกับงานที่สามารถแบ่งย่อยออกเป็นงานย่อย ๆ ซึ่งแต่ละงานจะถูกกำหนดให้กับเธรดเพื่อให้เสร็จสมบูรณ์
อะไรคือความแตกต่างระหว่างเธรด Python และการประมวลผลหลายขั้นตอน?
ด้วยเธรดการทำงานพร้อมกันสามารถทำได้โดยใช้เธรดหลายเธรด แต่เนื่องจาก GIL สามารถรันเธรดได้ครั้งละหนึ่งเธรดเท่านั้น ในการประมวลผลหลายขั้นตอนกระบวนการดั้งเดิมจะแยกกระบวนการออกเป็นกระบวนการย่อยหลาย ๆ กระบวนการโดยผ่าน GIL แต่ละโปรเซสลูกจะมีสำเนาหน่วยความจำของโปรแกรมทั้งหมด
Python multithreading และ multithrocessing เกี่ยวข้องกันอย่างไร?
ทั้งมัลติเธรดและการประมวลผลหลายขั้นตอนอนุญาตให้โค้ด Python ทำงานพร้อมกัน การประมวลผลหลายขั้นตอนเท่านั้นที่จะทำให้โค้ดของคุณขนานกันได้อย่างแท้จริง อย่างไรก็ตามหากรหัสของคุณเป็นแบบ IO-heavy (เช่นคำขอ HTTP) การใช้งานมัลติเธรดอาจทำให้โค้ดของคุณเร็วขึ้น
