Menguasai Pengaturcaraan Serentak Python: Tingkatkan Prestasi dengan Teknik Lanjutan-Tutorial Python-php.cn

Mastering Python

Keupayaan pengaturcaraan serentak Python telah berkembang dengan ketara, menawarkan alat berkuasa pembangun untuk menulis kod selari yang cekap. Saya telah meluangkan banyak masa untuk meneroka teknik lanjutan ini dan saya teruja untuk berkongsi pandangan saya dengan anda.

Pengaturcaraan tak segerak dengan asyncio ialah penukar permainan untuk tugas terikat I/O. Ia membolehkan kami menulis kod tidak menyekat yang boleh mengendalikan berbilang operasi serentak tanpa overhed threading. Berikut ialah contoh mudah bagaimana kita boleh menggunakan asyncio untuk mengambil data daripada berbilang URL serentak:

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = ['http://example.com', 'http://example.org', 'http://example.net']
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, result in zip(urls, results):
            print(f"Content length of {url}: {len(result)}")

asyncio.run(main())

Salin selepas log masuk

Kod ini menunjukkan cara kami boleh membuat berbilang coroutine untuk mengambil data daripada URL berbeza secara serentak. Fungsi asyncio.gather() membolehkan kami menunggu sehingga semua coroutine selesai dan mengumpul keputusan mereka.

Walaupun asyncio sangat baik untuk tugas terikat I/O, ia tidak sesuai untuk operasi terikat CPU. Bagi mereka, kami beralih kepada modul concurrent.futures, yang menyediakan kedua-dua ThreadPoolExecutor dan ProcessPoolExecutor. ThreadPoolExecutor sesuai untuk tugas terikat I/O yang tidak melepaskan GIL, manakala ProcessPoolExecutor sesuai untuk tugas terikat CPU.

Berikut ialah contoh menggunakan ThreadPoolExecutor untuk memuat turun berbilang fail serentak:

import concurrent.futures
import requests

def download_file(url):
    response = requests.get(url)
    filename = url.split('/')[-1]
    with open(filename, 'wb') as f:
        f.write(response.content)
    return f"Downloaded {filename}"

urls = [
    'https://example.com/file1.pdf',
    'https://example.com/file2.pdf',
    'https://example.com/file3.pdf'
]

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    future_to_url = {executor.submit(download_file, url): url for url in urls}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print(f"{url} generated an exception: {exc}")
        else:
            print(data)

Salin selepas log masuk

Kod ini mencipta kumpulan benang dengan tiga pekerja dan menyerahkan tugas muat turun untuk setiap URL. Fungsi as_completed() membolehkan kami memproses keputusan apabila ia tersedia, dan bukannya menunggu semua tugasan selesai.

Untuk tugasan terikat CPU, kami boleh menggunakan ProcessPoolExecutor untuk memanfaatkan berbilang teras CPU. Berikut ialah contoh yang mengira nombor perdana secara selari:

import concurrent.futures
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

def find_primes(start, end):
    return [n for n in range(start, end) if is_prime(n)]

ranges = [(1, 25000), (25001, 50000), (50001, 75000), (75001, 100000)]

with concurrent.futures.ProcessPoolExecutor() as executor:
    results = executor.map(lambda r: find_primes(*r), ranges)

all_primes = [prime for sublist in results for prime in sublist]
print(f"Found {len(all_primes)} prime numbers")

Salin selepas log masuk

Kod ini membahagikan tugas mencari nombor perdana kepada empat julat dan memprosesnya secara selari menggunakan proses Python yang berasingan. Fungsi map() menggunakan fungsi find_primes() kami pada setiap julat dan mengumpul hasilnya.

Apabila bekerja dengan berbilang proses, kami selalunya perlu berkongsi data antara mereka. Modul multiprocessing menyediakan beberapa pilihan untuk ini, termasuk memori dan baris gilir yang dikongsi. Berikut ialah contoh menggunakan tatasusunan memori yang dikongsi:

from multiprocessing import Process, Array
import numpy as np

def worker(shared_array, start, end):
    for i in range(start, end):
        shared_array[i] = i * i

if __name__ == '__main__':
    size = 10000000
    shared_array = Array('d', size)

    # Create 4 processes
    processes = []
    chunk_size = size // 4
    for i in range(4):
        start = i * chunk_size
        end = start + chunk_size if i < 3 else size
        p = Process(target=worker, args=(shared_array, start, end))
        processes.append(p)
        p.start()

    # Wait for all processes to finish
    for p in processes:
        p.join()

    # Convert shared array to numpy array for easy manipulation
    np_array = np.frombuffer(shared_array.get_obj())
    print(f"Sum of squares: {np_array.sum()}")

Salin selepas log masuk

Kod ini mencipta tatasusunan memori dikongsi dan menggunakan empat proses untuk mengira kuasa dua nombor secara selari. Tatasusunan yang dikongsi membolehkan semua proses menulis ke ruang memori yang sama, mengelakkan keperluan untuk komunikasi antara proses.

Walaupun teknik ini hebat, teknik ini datang dengan cabaran tersendiri. Keadaan perlumbaan, kebuntuan dan penukaran konteks yang berlebihan semuanya boleh memberi kesan kepada prestasi dan ketepatan. Adalah penting untuk mereka bentuk kod serentak anda dengan teliti dan menggunakan primitif penyegerakan yang sesuai apabila perlu.

Sebagai contoh, apabila berbilang rangkaian atau proses perlu mengakses sumber yang dikongsi, kami boleh menggunakan Kunci untuk memastikan keselamatan benang:

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = ['http://example.com', 'http://example.org', 'http://example.net']
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, result in zip(urls, results):
            print(f"Content length of {url}: {len(result)}")

asyncio.run(main())

Salin selepas log masuk

Kod ini menunjukkan cara menggunakan Kunci untuk melindungi kaunter kongsi daripada keadaan perlumbaan apabila berbilang rangkaian menambahnya secara serentak.

Satu lagi teknik lanjutan ialah penggunaan semaphore untuk mengawal akses kepada sumber terhad. Berikut ialah contoh yang mengehadkan bilangan sambungan rangkaian serentak:

import concurrent.futures
import requests

def download_file(url):
    response = requests.get(url)
    filename = url.split('/')[-1]
    with open(filename, 'wb') as f:
        f.write(response.content)
    return f"Downloaded {filename}"

urls = [
    'https://example.com/file1.pdf',
    'https://example.com/file2.pdf',
    'https://example.com/file3.pdf'
]

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    future_to_url = {executor.submit(download_file, url): url for url in urls}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print(f"{url} generated an exception: {exc}")
        else:
            print(data)

Salin selepas log masuk

Kod ini menggunakan semafor untuk mengehadkan bilangan sambungan rangkaian serentak kepada 10, menghalang rangkaian atau pelayan yang melampau.

Apabila bekerja dengan kod serentak, adalah penting juga untuk mengendalikan pengecualian dengan betul. Modul asyncio menyediakan fungsi asyncio.gather() dengan parameter return_exceptions yang boleh berguna untuk ini:

import concurrent.futures
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

def find_primes(start, end):
    return [n for n in range(start, end) if is_prime(n)]

ranges = [(1, 25000), (25001, 50000), (50001, 75000), (75001, 100000)]

with concurrent.futures.ProcessPoolExecutor() as executor:
    results = executor.map(lambda r: find_primes(*r), ranges)

all_primes = [prime for sublist in results for prime in sublist]
print(f"Found {len(all_primes)} prime numbers")

Salin selepas log masuk

Kod ini menunjukkan cara mengendalikan pengecualian dalam tugasan serentak tanpa menghentikan pelaksanaan tugas lain.

Apabila kami mendalami pengaturcaraan serentak, kami menghadapi konsep yang lebih maju seperti gelung acara dan rantaian coroutine. Berikut ialah contoh yang menunjukkan cara merantai coroutine:

from multiprocessing import Process, Array
import numpy as np

def worker(shared_array, start, end):
    for i in range(start, end):
        shared_array[i] = i * i

if __name__ == '__main__':
    size = 10000000
    shared_array = Array('d', size)

    # Create 4 processes
    processes = []
    chunk_size = size // 4
    for i in range(4):
        start = i * chunk_size
        end = start + chunk_size if i < 3 else size
        p = Process(target=worker, args=(shared_array, start, end))
        processes.append(p)
        p.start()

    # Wait for all processes to finish
    for p in processes:
        p.join()

    # Convert shared array to numpy array for easy manipulation
    np_array = np.frombuffer(shared_array.get_obj())
    print(f"Sum of squares: {np_array.sum()}")

Salin selepas log masuk

Kod ini merantai tiga coroutine (fetch_data, process_data dan save_result) untuk membuat saluran paip bagi setiap URL. Fungsi asyncio.gather() kemudian menjalankan saluran paip ini secara serentak.

Apabila bekerja dengan tugasan yang berjalan lama, selalunya perlu untuk melaksanakan mekanisme pembatalan dan tamat masa. Berikut ialah contoh yang menunjukkan kedua-duanya:

from threading import Lock, Thread

class Counter:
    def __init__(self):
        self.count = 0
        self.lock = Lock()

    def increment(self):
        with self.lock:
            self.count += 1

def worker(counter, num_increments):
    for _ in range(num_increments):
        counter.increment()

counter = Counter()
threads = []
for _ in range(10):
    t = Thread(target=worker, args=(counter, 100000))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Final count: {counter.count}")

Salin selepas log masuk

Kod ini memulakan lima tugasan yang berjalan lama tetapi menetapkan tamat masa 5 saat untuk semua tugasan diselesaikan. Jika tamat masa dicapai, ia membatalkan semua tugasan yang tinggal.

Kesimpulannya, keupayaan pengaturcaraan serentak Python menawarkan pelbagai alat dan teknik untuk menulis kod selari yang cekap. Daripada pengaturcaraan asynchronous dengan asyncio kepada multiprocessing untuk tugas terikat CPU, teknik lanjutan ini boleh meningkatkan prestasi aplikasi kami dengan ketara. Walau bagaimanapun, adalah penting untuk memahami konsep asas, memilih alat yang sesuai untuk setiap tugasan dan mengurus sumber yang dikongsi dengan teliti dan keadaan perlumbaan yang berpotensi. Dengan amalan dan reka bentuk yang teliti, kami boleh memanfaatkan kuasa penuh pengaturcaraan serentak dalam Python untuk membina aplikasi yang pantas, berskala dan responsif.

Ciptaan Kami

Pastikan anda melihat ciptaan kami:

Kami berada di Medium

Atas ialah kandungan terperinci Menguasai Pengaturcaraan Serentak Python: Tingkatkan Prestasi dengan Teknik Lanjutan. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!