Hibrit Python Eşzamanlılığı: AsyncIO + Multiprocessing

Python geliştiricileri, yüksek iş hacimli veri boru hatları oluştururken genellikle klasik bir ikilemle karşılaşır: Küresel Yorumlayıcı Kilidi (GIL). Yıllarca seçenek ikiliydi—I/O odaklı görevler için asyncio, CPU odaklı işlemler için ise multiprocessing kullanmak. Ancak modern web kazıma genellikle her ikisini de gerektirir. Ağ yanıtlarını beklerken (I/O) aynı anda ağır JSON yapılarını işleyen veya çekilen sayfalar üzerinde görüntü tanıma yapan bir kazıyıcı hayal edin.

AsyncIO ve Multiprocessing'i birleştirmek güçlü bir hibrit desen oluşturur. Bu yaklaşım, ağ gecikmeleri için asenkron verimliliği kullanırken hesaplama için GIL'yi atlar. Bu kılavuzda, bu sistemi nasıl mimarileştireceğimizi, gerekli IPC (İşlem İçi İletişim) yönetimini ele alacağımızı ve sağlam bir işçi havuzu nasıl uygulayacağımızı inceleyeceğiz.

Mimariyi Anlamak

Hibrit bir sistem oluşturmak için önce her aracın sınırlarını tanımlamamız gerekir. asyncio, olay döngüsü tabanlıdır ve binlerce eşzamanlı ağ bağlantısı için mükemmeldir. Buna karşılık, multiprocessing, kendi GIL'sine sahip ayrı süreçler oluşturarak yoğun CPU gerektiren işlemler için idealdir.

Mimari genellikle "yönetici" olarak hareket eden ana asenkron olay döngüsünü içerir. Bu yönetici istekleri oluşturur ve bunları bir kuyruk aracılığıyla işçi süreçlerine iletir. Ayrı süreçlerde çalışan işçiler ağır işi yapar ve sonuçları döndürür. Kritik olan nokta, asenkron yönetici ile işçi havuzu arasındaki iletişimin verimli olması ve darboğazlara yol açmamasıdır.

Asyncio ile Uygulama Stratejisi

İşçi süreçlerinin yaşam döngüsünü yönetmek için asyncio ve veri iletimi için multiprocessing.Queue kullanacağız. Anahtar nokta, olay döngüsünün takılmaması için engelleyici işçi mantığını bir arka plan görevinin içine sarmalamaktır.

import asyncio
import multiprocessing as mp
from aiohttp import ClientSession
import json

def cpu_intensive_parser(job):
    """Tek bir işlemde ağır CPU işini simüle eder."""
    data = job['data']
    # İşleme simüle etmek için döngü ve hashleme
    result = 0
    for char in data:
        result += hash(char)
    return result

async def fetch_task(session, url, data_queue, result_queue):
    """Veriyi asenkron olarak çeker ve işleme işçilere devreder."""
    try:
        async with session.get(url) as response:
            html = await response.text()
            
            # İşçi için bir iş oluştur
            job = {'url': url, 'data': html}
            data_queue.put(job)
            
            # Belirli işçiden sonucu bekle
            result = result_queue.get()
            return result
    except Exception as e:
        print(f"{url} çekilirken hata: {e}")
        return None

async def main():
    # IPC için kuyruklar oluştur
    data_queue = mp.Queue()
    result_queue = mp.Queue()
    
    # İşçi havuzunu tanımla
    num_workers = 4
    pool = []
    for i in range(num_workers):
        p = mp.Process(target=worker_loop, args=(data_queue, result_queue, i))
        p.start()
        pool.append(p)

    urls = [f"https://example.com/page/{i}" for i in range(10)]
    
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            task = asyncio.create_task(fetch_task(session, url, data_queue, result_queue))
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
    
    # Temizlik
    data_queue.close()
    data_queue.join_thread()
    for p in pool:
        p.terminate()

def worker_loop(data_queue, result_queue, worker_id):
    """CPU odaklı görevleri işlemek için ayrı bir süreçte çalışır."""
    while True:
        try:
            job = data_queue.get(timeout=5)
            if job is None:
                break
            
            result = cpu_intensive_parser(job)
            # Gerçek bir senaryoda, sonuçları işlere ID kullanarak eşleştirmeniz gerekebilir
            # Burada basit bir engelleyici bekleme varsayıyoruz
            result_queue.put({'worker': worker_id, 'result': result})
        except:
            break

if __name__ == "__main__":
    # Windows'ta giriş noktasını korumamız gerekir
    mp.set_start_method('spawn')
    asyncio.run(main())

İletişimi ve Ölçeklenebilirliği Optimize Etmek

Yukarıdaki örnekte, result_queue basit bir engelleyici kuyruktur. Milyonlarca öğe işleyen üretim seviyesindeki kazıyıcılar için, asyncio.Queue'un paylaşılan bellek yöneticisiyle etkileşime girmesi veya GIL'yi kesinlikle atlamak istemiyorsanız daha hafif CPU görevleri için asyncio.to_thread kullanılması gibi daha gelişmiş bir senkronizasyon mekanizmasına ihtiyacınız olabilir.

Ayrıca, asyncio.TimeoutError ve ağ hatalarını zarif bir şekilde ele aldığınızdan emin olun. Yüzlerce eşzamanlı istek için ölçeklendiğinizde kuyruk bir darboğaz haline gelebilir. Paylaşılan istatistikleri güncellerken asyncio.Lock kullanmayı veya işçi kuyruğunu arka planda asenkron olarak tüketen özel bir sonuç toplayıcı işlev kullanmayı düşünün.

Sonuç

Python'da yüksek performanslı bir kazıyıcı oluşturmak artık asyncio ile multiprocessing arasında seçim yapmayı gerektirmez. Hibrit bir desen uygulayarak, binlerce eşzamanlı ağ bağlantısını işleme yeteneği kazanırken, aynı zamanda işleme ve veri dönüştürme için mevcut tüm CPU çekirdeklerini kullanabilirsiniz. Bu yaklaşım, maksimum iş hacmi ve minimum gecikme sağlar ve modern, büyük ölçekli veri çıkarma boru hatları için standart haline gelir.

İşçi süreçlerinizi ölü kilitlere karşı her zaman test ettiğinizden ve IPC mekanizmalarınızın iş parçacığı güvenli olduğundan emin olun. Doğru mimariyle Python, web kazıma ve veri mühendisliği dünyasında hâlâ baskın bir güç olmaya devam ediyor.