H.Ozgan
a.k.a netologist

Python'da Eşzamanlılık (Concurrency) — Kapsamlı Rehber

10 Mart 2026 · netologist · 19 dakika, 3842 kelime · #asyncio #concurrency #gil #multiprocessing #note #python #threading

İçindekiler

  1. Giriş: Eşzamanlılık (Concurrency) ve Paralellik (Parallelism)
  2. Global Interpreter Lock (GIL)
  3. threading Modülü
  4. multiprocessing Modülü
  5. asyncio Modülü
  6. concurrent.futures Modülü
  7. Senkronizasyon Araçları
  8. Kuyruklar ve Görevler Arası İletişim
  9. Doğru Aracı Seçmek
  10. Yaygın Hatalar ve İyi Uygulamalar
  11. İleri Düzey Konular
  12. Özet Tablo

1. Giriş: Eşzamanlılık (Concurrency) ve Paralellik (Parallelism)

Eşzamanlılık (concurrency), birden fazla işi aynı anda yönetmekle ilgilidir — programı, birden fazla görevin tam olarak aynı anda çalışmasa bile ilerleme kaydedebileceği şekilde yapılandırmaktır. Paralellik (parallelism) ise birden fazla işi gerçekten aynı anda yapmakla ilgilidir — birden fazla CPU çekirdeğinde işlemleri gerçekten eşzamanlı olarak yürütmektir.

KavramTanımÖrnek
Eşzamanlılık (Concurrency)Birden fazla görev iç içe geçmiş şekilde ilerlerTek çekirdeğin görevler arasında geçiş yapması
Paralellik (Parallelism)Birden fazla görev gerçekten aynı anda çalışırBirden fazla çekirdeğin aynı anda çalışması
Sıralı (Sequential)Sırayla, tek seferde bir görevHiçbir örtüşme yok

Yararlı bir zihinsel model: eşzamanlılık bir yapıdır (işi nasıl organize ettiğinizdir), paralellik ise bir yürütme özelliğidir (donanımın işi gerçekte nasıl çalıştırdığıdır). Paralellik olmadan eşzamanlılık olabilir (örneğin tek çekirdekte çalışan asyncio), ve “çok sayıda küçük görev” anlamında fazla eşzamanlılık olmadan da paralellik olabilir (örneğin büyük bir matris çarpımının çekirdeklere bölünmesi).

Python üç ana eşzamanlılık modeli sunar:

2. Global Interpreter Lock (GIL)

GIL, Python nesnelerine erişimi koruyan bir mutex’tir; aynı süreç (process) içinde birden fazla native thread’in aynı anda Python bytecode’u çalıştırmasını engeller. Bunun nedeni, CPython’un bellek yönetiminin (referans sayımı) varsayılan olarak thread-safe olmamasıdır.

GIL neden var?

Pratik etkileri

import sys
print(sys.getswitchinterval())  # Varsayılan: 0.005 saniye
sys.setswitchinterval(0.001)     # Daha sık geçiş yap

Gelecek: Free-Threaded Python (PEP 703)

Python 3.13 itibarıyla, GIL’i tamamen kaldıran deneysel bir free-threaded (serbest thread’li) build (python3.13t) mevcuttur. Bu, (başlangıçta Sam Gross tarafından “nogil” olarak önerilen) büyük ve devam eden bir çalışmadır. Bazı ödünleşimleri vardır:

Free-threaded Python yaygınlaşıp ekosistem tarafından evrensel olarak desteklenene kadar, aşağıdaki pratik öneri (I/O için thread, CPU için process) standart yaklaşım olmaya devam etmektedir.

GIL’in etkisini gösteren örnek

import threading
import time

def cpu_bound_task(n):
    count = 0
    for i in range(n):
        count += i * i
    return count

def run_single_threaded():
    start = time.perf_counter()
    cpu_bound_task(50_000_000)
    cpu_bound_task(50_000_000)
    print(f"Tek thread: {time.perf_counter() - start:.2f}s")

def run_multi_threaded():
    start = time.perf_counter()
    t1 = threading.Thread(target=cpu_bound_task, args=(50_000_000,))
    t2 = threading.Thread(target=cpu_bound_task, args=(50_000_000,))
    t1.start(); t2.start()
    t1.join(); t2.join()
    print(f"Çoklu thread: {time.perf_counter() - start:.2f}s")

run_single_threaded()
run_multi_threaded()
# Standart CPython'da, CPU ağırlıklı işler için çoklu thread anlamlı bir şekilde daha hızlı DEĞİLDİR

3. threading Modülü

Thread’ler hafiftir ve bir süreç (process) içinde aynı bellek alanını paylaşırlar. I/O ağırlıklı işler için en uygun olanlardır: ağ istekleri, dosya I/O, veritabanı sorguları, harici kaynakları bekleme.

Temel thread oluşturma

import threading
import time

def worker(name, delay):
    print(f"Thread {name} başlıyor")
    time.sleep(delay)
    print(f"Thread {name} bitti")

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(f"Worker-{i}", 1))
    threads.append(t)
    t.start()

for t in threads:
    t.join()  # Tüm thread'lerin tamamlanmasını bekle
print("Tüm thread'ler bitti")

Thread sınıfını miras almak (subclassing)

class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(f"{self.name} çalışıyor")

t = MyThread("özel-thread")
t.start()
t.join()

Daemon thread’ler

Daemon thread’ler, normal thread’lerin aksine ana program çıkarken otomatik olarak sonlandırılır. Normal thread’ler ise tamamlanana kadar programın çıkışını engeller.

t = threading.Thread(target=worker, args=("daemon", 10), daemon=True)
t.start()
# Bu thread hâlâ çalışıyor olsa bile program çıkabilir

Thread-local veriler

threading.local(), her thread’e kendi izole depolama alanını verir — veritabanı bağlantıları gibi thread başına durum (state) için kullanışlıdır.

local_data = threading.local()

def process():
    local_data.value = threading.current_thread().name
    print(local_data.value)

for i in range(3):
    threading.Thread(target=process).start()

ThreadPoolExecutor (önizleme — Bölüm 6’da detaylı işlenmiştir)

from concurrent.futures import ThreadPoolExecutor

def fetch(url):
    # ağ çağrısını simüle et
    time.sleep(0.5)
    return f"{url} adresinden veri"

urls = ["http://a.com", "http://b.com", "http://c.com"]
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(fetch, urls))
print(results)

Yarış koşulları (race condition) ve kilitlerin önemi

counter = 0
lock = threading.Lock()

def increment_unsafe():
    global counter
    for _ in range(100_000):
        counter += 1  # Atomik DEĞİL! oku-değiştir-yaz yarış koşulu

def increment_safe():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1  # Kilit sayesinde atomik

threads = [threading.Thread(target=increment_unsafe) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)  # Yarış koşulları nedeniyle muhtemelen 400.000 DEĞİL

4. multiprocessing Modülü

Süreçlerin (process) ayrı bellek alanları vardır, bu da GIL’i tamamen atlamalarını sağlar. CPU ağırlıklı işler için idealdir: sayısal hesaplama, görüntü/video işleme, veri dönüşümü, simülasyonlar.

Temel süreç oluşturma

import multiprocessing as mp

def cpu_task(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":  # Windows/macOS'te (spawn yöntemi) ZORUNLU
    p = mp.Process(target=cpu_task, args=(10_000_000,))
    p.start()
    p.join()

Paralel eşleme (mapping) için Pool

import multiprocessing as mp

def square(x):
    return x * x

if __name__ == "__main__":
    with mp.Pool(processes=4) as pool:
        results = pool.map(square, range(20))
        print(results)

        # Diğer Pool metotları:
        result_async = pool.apply_async(square, (10,))
        print(result_async.get())

        # imap: tembel (lazy), sırayı korur
        for r in pool.imap(square, range(5)):
            print(r)

        # imap_unordered: tembel, sıra garanti değil, genellikle daha hızlı
        for r in pool.imap_unordered(square, range(5)):
            print(r)

Süreçler arasında veri paylaşımı

Süreçler belleği paylaşmadığından, açık mekanizmalara ihtiyacınız vardır:

from multiprocessing import Process, Value, Array, Manager

def increment(shared_val, shared_arr):
    with shared_val.get_lock():
        shared_val.value += 1
    shared_arr[0] += 10

if __name__ == "__main__":
    val = Value('i', 0)          # paylaşılan tamsayı
    arr = Array('i', [0, 0, 0])  # paylaşılan dizi

    processes = [Process(target=increment, args=(val, arr)) for _ in range(4)]
    for p in processes: p.start()
    for p in processes: p.join()
    print(val.value, arr[:])

    # Daha karmaşık paylaşılan nesneler için Manager (dict, list)
    with Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()
        shared_dict["key"] = "value"
        shared_list.append(1)

Süreçler arası iletişim: Pipe’lar ve Kuyruklar

from multiprocessing import Process, Pipe, Queue

def pipe_worker(conn):
    conn.send("alt süreçten merhaba")
    conn.close()

def queue_worker(q):
    q.put("kuyruk üzerinden alt süreçten merhaba")

if __name__ == "__main__":
    # Pipe: iki yönlü, iki süreç için
    parent_conn, child_conn = Pipe()
    p = Process(target=pipe_worker, args=(child_conn,))
    p.start()
    print(parent_conn.recv())
    p.join()

    # Queue: çoktan çoğa, süreç-güvenli (process-safe)
    q = Queue()
    p2 = Process(target=queue_worker, args=(q,))
    p2.start()
    print(q.get())
    p2.join()

Süreç başlatma yöntemleri (start methods)

YöntemAçıklamaVarsayılan İşletim Sistemi
forkAlt süreç, ana sürecin belleğini kopyalar (hızlı, ancak thread’lerle güvensiz olabilir)Linux, macOS (3.8 öncesi)
spawnYeni bir Python yorumlayıcısı (interpreter) başlatılır, modülü temiz bir şekilde import ederWindows, macOS (3.8+)
forkserverBir sunucu süreci yeni worker’lar fork eder, bazı fork sorunlarından kaçınırUnix’te mevcut
import multiprocessing as mp
mp.set_start_method('spawn')  # süreçler oluşturulmadan önce bir kez çağrılmalı

Multiprocessing’in maliyetleri

5. asyncio Modülü

asyncio, bir olay döngüsü (event loop) kullanarak tek iş parçacıklı, tek süreçli işbirlikçi çoklu görev (cooperative multitasking) sağlar. Thread’lere göre çok daha düşük ek yükle, çok yüksek eşzamanlılık gerektiren I/O ağırlıklı iş yüklerinde (örneğin binlerce eşzamanlı ağ bağlantısı) üstündür.

Temel kavramlar

Temel örnek

import asyncio

async def say_after(delay, message):
    await asyncio.sleep(delay)
    print(message)

async def main():
    print("başladı")
    await say_after(1, "merhaba")
    await say_after(1, "dünya")
    print("bitti")  # Toplam: ~2 saniye (sıralı await'ler)

asyncio.run(main())

Görevleri eşzamanlı çalıştırma

import asyncio

async def say_after(delay, message):
    await asyncio.sleep(delay)
    print(message)
    return message

async def main():
    # Sıralı: toplam ~2 saniye
    # await say_after(1, "merhaba")
    # await say_after(1, "dünya")

    # Eşzamanlı: toplam ~1 saniye
    task1 = asyncio.create_task(say_after(1, "merhaba"))
    task2 = asyncio.create_task(say_after(1, "dünya"))
    await task1
    await task2

asyncio.run(main())

Birden çok coroutine çalıştırmak için asyncio.gather

import asyncio

async def fetch_data(id):
    await asyncio.sleep(1)
    return f"veri-{id}"

async def main():
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3),
    )
    print(results)  # ['veri-1', 'veri-2', 'veri-3'] — hepsi ~1 saniyede

asyncio.run(main())

asyncio.TaskGroup (Python 3.11+) — yapılandırılmış eşzamanlılık

import asyncio

async def fetch_data(id):
    await asyncio.sleep(1)
    if id == 2:
        raise ValueError("id 2'de hata oluştu")
    return f"veri-{id}"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            for i in range(3):
                tg.create_task(fetch_data(i))
    except* ValueError as eg:
        print(f"Yakalanan istisna grubu: {eg.exceptions}")

asyncio.run(main())

TaskGroup, bir görev başarısız olursa kardeş görevleri otomatik olarak iptal eder ve istisnaları doğru şekilde yayar — manuel olarak bir görev listesi yönetmekten çok daha güvenlidir.

Zaman aşımı (timeout) ve iptal

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return "tamamlandı"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2)
    except asyncio.TimeoutError:
        print("İşlem zaman aşımına uğradı")

    # Python 3.11+ timeout context manager
    try:
        async with asyncio.timeout(2):
            await slow_operation()
    except TimeoutError:
        print("asyncio.timeout ile zaman aşımına uğradı")

asyncio.run(main())

Asenkron context manager’lar ve iterator’lar

class AsyncResource:
    async def __aenter__(self):
        print("kaynak alınıyor")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("kaynak serbest bırakılıyor")
        await asyncio.sleep(0.1)

async def main():
    async with AsyncResource() as res:
        print("kaynak kullanılıyor")

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.i = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.i >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.i += 1
        return self.i

async def iterate():
    async for num in AsyncCounter(3):
        print(num)

asyncio.run(main())
asyncio.run(iterate())

asyncio içinde bloklayan / CPU ağırlıklı kod çalıştırma

Bloklayan çağrılar (time.sleep veya CPU ağırlıklı döngüler gibi) tüm event loop’u dondurur ve diğer tüm coroutine’leri engeller. Bu işi devretmek için executor’ları kullanın.

import asyncio
import time

def blocking_io():
    time.sleep(2)  # bloklayan I/O simülasyonu — bu, event loop'u dondururdu
    return "bloklayan sonuç"

async def main():
    loop = asyncio.get_running_loop()
    # Bir thread pool'a devret
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

    # Ya da asyncio.to_thread kullan (Python 3.9+, daha basit API)
    result2 = await asyncio.to_thread(blocking_io)
    print(result2)

asyncio.run(main())

Asenkron generator’lar ve akış (streaming)

async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for i in async_range(5):
        print(i)

asyncio.run(main())

Üretici-tüketici (producer-consumer) desenleri için asyncio.Queue

import asyncio
import random

async def producer(queue, n):
    for i in range(n):
        item = f"öge-{i}"
        await queue.put(item)
        print(f"Üretildi: {item}")
        await asyncio.sleep(random.uniform(0.1, 0.3))
    await queue.put(None)  # tamamlandığını belirten sinyal (sentinel)

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Tüketildi: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(producer(queue, 5), consumer(queue))

asyncio.run(main())

6. concurrent.futures Modülü

Hem thread’ler hem de süreçler için yüksek seviyeli, birleşik bir arayüz sağlar ve çoğu manuel muhasebeyi (bookkeeping) soyutlar.

ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch(url):
    time.sleep(1)
    return f"{url} adresinden sonuç"

urls = [f"http://site{i}.com" for i in range(5)]

with ThreadPoolExecutor(max_workers=3) as executor:
    # Yöntem 1: map (sırayı korur, hepsi bitene kadar bloklar)
    for result in executor.map(fetch, urls):
        print(result)

    # Yöntem 2: submit + as_completed (sonuçları bitiş sırasına göre işle)
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            print(f"{url}: {data}")
        except Exception as exc:
            print(f"{url} bir istisna üretti: {exc}")

ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor
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

if __name__ == "__main__":
    numbers = list(range(100_000, 100_100))
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(is_prime, numbers))
    primes = [n for n, p in zip(numbers, results) if p]
    print(primes)

Future’larda istisna yönetimi

from concurrent.futures import ThreadPoolExecutor

def risky(x):
    if x == 3:
        raise ValueError("kötü değer")
    return x * 2

with ThreadPoolExecutor() as executor:
    futures = [executor.submit(risky, i) for i in range(5)]
    for f in futures:
        try:
            print(f.result())
        except ValueError as e:
            print(f"Yakalandı: {e}")

add_done_callback ile geri çağrılar (callback)

from concurrent.futures import ThreadPoolExecutor
import time

def task(n):
    time.sleep(1)
    return n * n

def callback(future):
    print(f"Görev şu sonuçla tamamlandı: {future.result()}")

with ThreadPoolExecutor() as executor:
    future = executor.submit(task, 5)
    future.add_done_callback(callback)

7. Senkronizasyon Araçları

Birden fazla thread (veya süreç) paylaşılan durum (shared state) üzerinde işlem yaptığında, yarış koşullarını, deadlock’ları ve veri bozulmasını önlemek için senkronizasyona ihtiyaç vardır.

Lock (mutex)

Bir kritik bölüme aynı anda yalnızca bir thread’in erişmesini sağlar.

import threading

lock = threading.Lock()
shared_resource = []

def add_item(item):
    with lock:  # lock.acquire() / lock.release() ile eşdeğerdir
        shared_resource.append(item)

RLock (yeniden girilebilir kilit)

Aynı thread’in kendini kilitlemeden (deadlock yapmadan) kilidi birden çok kez almasına izin verir — recursive (özyinelemeli) fonksiyonlar için kullanışlıdır.

import threading

rlock = threading.RLock()

def recursive_function(n):
    with rlock:
        if n > 0:
            print(n)
            recursive_function(n - 1)  # Aynı thread kilidi tekrar alır, deadlock yok

Semaphore

Bir kaynağa aynı anda erişebilecek thread sayısını sınırlar — bağlantı havuzları (connection pool), hız sınırlama (rate-limiting) için kullanışlıdır.

import threading
import time

semaphore = threading.Semaphore(3)  # aynı anda yalnızca 3 thread

def limited_resource(name):
    with semaphore:
        print(f"{name} kaynağı aldı")
        time.sleep(1)
        print(f"{name} kaynağı serbest bıraktı")

for i in range(10):
    threading.Thread(target=limited_resource, args=(f"Thread-{i}",)).start()

BoundedSemaphore

Semaphore gibidir ancak alındığından daha fazla serbest bırakılırsa hata verir — programlama hatalarını yakalar.

Event

Thread’lerin bekleyebileceği basit bir bayrak (flag), bir-çoka sinyal göndermek için kullanılır.

import threading
import time

event = threading.Event()

def waiter():
    print("Olay bekleniyor...")
    event.wait()  # event.set() çağrılana kadar bloklar
    print("Olay alındı, devam ediliyor!")

def setter():
    time.sleep(2)
    print("Olay ayarlanıyor")
    event.set()

threading.Thread(target=waiter).start()
threading.Thread(target=setter).start()

Condition

Bir kilidi bir bildirim (notification) bekleme yeteneğiyle birleştirir — üretici-tüketici (producer-consumer) desenlerinde kullanılır.

import threading

condition = threading.Condition()
items = []

def consumer():
    with condition:
        while not items:
            condition.wait()  # kilidi serbest bırakır, notify'ı bekler
        item = items.pop(0)
        print(f"Tüketildi: {item}")

def producer():
    with condition:
        items.append("öge")
        print("Öge üretildi")
        condition.notify()  # bekleyen bir thread'i uyandırır

threading.Thread(target=consumer).start()
threading.Thread(target=producer).start()

Barrier

Bir grup thread’in hepsi bariyer noktasına ulaşana kadar beklemesini sağlar.

import threading

barrier = threading.Barrier(3)

def worker(name):
    print(f"{name} bariyerde bekliyor")
    barrier.wait()  # 3 thread'in hepsi wait() çağırana kadar bloklar
    print(f"{name} bariyeri geçti")

for i in range(3):
    threading.Thread(target=worker, args=(f"Worker-{i}",)).start()

Deadlock’lar: nasıl oluşurlar ve nasıl önlenirler

Bir deadlock, iki veya daha fazla thread’in birbirini süresiz olarak beklemesi durumunda oluşur.

import threading

lock_a = threading.Lock()
lock_b = threading.Lock()

def thread_1():
    with lock_a:
        # ... bir iş yapar
        with lock_b:  # thread_2 zaten lock_b'yi tutuyor olabilir
            pass

def thread_2():
    with lock_b:
        with lock_a:  # thread_1 zaten lock_a'yı tutuyor olabilir — DEADLOCK
            pass

Önleme stratejileri:

asyncio senkronizasyon araçları

asyncio, coroutine’ler için tasarlanmış benzer araçlar sağlar: asyncio.Lock, asyncio.Event, asyncio.Condition, asyncio.Semaphore, asyncio.Barrier (3.11+). Bunlar thread-safe DEĞİLDİR ve yalnızca tek bir event loop içinde kullanılmalıdır.

import asyncio

async def main():
    lock = asyncio.Lock()
    async with lock:
        print("kritik bölüm")

asyncio.run(main())

8. Kuyruklar ve Görevler Arası İletişim

Kuyruklar, eşzamanlı çalışma birimleri arasında veri geçirmek için thread/süreç-güvenli veri yapılarıdır — genellikle kilitlere güvenmek yerine veri paylaşmanın en güvenli yoludur.

queue.Queue (threading için)

import queue
import threading

q = queue.Queue(maxsize=10)

def producer():
    for i in range(5):
        q.put(f"öge-{i}")

def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Alındı: {item}")
        q.task_done()

threading.Thread(target=producer).start()
t = threading.Thread(target=consumer)
t.start()

q.join()  # tüm ögeler işlenene kadar bekle
q.put(None)
t.join()

Kuyruk çeşitleri: Queue (FIFO), LifoQueue (yığın/stack gibi), PriorityQueue (önceliğe göre sıralı).

multiprocessing.Queue

Aynı API, ancak süreçler arasında paylaşmak güvenlidir (dahili olarak pipe’lar ve kilitler kullanır, veriyi pickle’lar).

asyncio.Queue

Kavramsal olarak aynı API, ancak coroutine’ler için tasarlanmıştır (await queue.put(), await queue.get()).

9. Doğru Aracı Seçmek

İş Yükü TürüÖnerilen AraçNeden
I/O ağırlıklı, az bağlantıthreadingBasit, iyi çalışır, I/O sırasında GIL serbest bırakılır
I/O ağırlıklı, binlerce bağlantıasyncioThread’lere göre görev başına son derece düşük ek yük
CPU ağırlıklı (ağır hesaplama)multiprocessing / ProcessPoolExecutorGIL’i aşar, birden fazla çekirdek kullanır
Karma I/O + hafif CPU işiasyncio + run_in_executor/to_threadEvent loop’u tepkisel (responsive) tutar
Veri üzerinde basit paralel “map”concurrent.futures (Thread veya Process pool)Yüksek seviyeli, daha az standart kod (boilerplate)
Worker’lar arasında paylaşılan değiştirilebilir durum gereklithreading (kilitlerle) veya multiprocessing.ManagerSüreçlerin açık paylaşım mekanizmalarına ihtiyacı vardır

Karar akış şeması (metin olarak)

  1. Görev I/O ağırlıklı mı (ağ, disk, harici sistemleri bekleme)?
    • Evet ve tüm çağrı yığınını (call stack) async kütüphaneleri kullanılabilir şekilde kontrol ediyorsanız → asyncio kullanın.
    • Evet, ancak bloklayan kütüphanelerle çalışıyorsanız (örneğin eski veritabanı sürücüleri) → threading veya ThreadPoolExecutor kullanın.
  2. Görev CPU ağırlıklı mı (hesaplama yoğun, sıkı döngüler, sayı işleme)?
    • Evet → multiprocessing veya ProcessPoolExecutor kullanın.
    • NumPy/Cython/native kütüphanelerin GIL’i serbest bırakıp bırakmadığını düşünün — eğer öyleyse, threading yine de yardımcı olabilir.
  3. Aşırı yüksek eşzamanlılığa (10.000+ eşzamanlı bağlantı) ihtiyacınız var mı?
    • Evet → asyncio esasen zorunludur; thread’ler çok pahalı hale gelir (bellek/context switching).

10. Yaygın Hatalar ve İyi Uygulamalar

Hata 1: Thread’lerin CPU ağırlıklı Python kodunu hızlandırdığına inanmak

Daha önce tartışıldığı gibi, GIL standart CPython’da bunu engeller. Bunun yerine multiprocessing kullanın.

Hata 2: multiprocessing ile if __name__ == "__main__": korumasını unutmak

Windows ve macOS’ta (spawn yöntemi), bu korumayı unutmak, alt sürecin ana modülü yeniden import etmesi nedeniyle sonsuz süreç oluşturmaya veya çökmelere neden olur.

# Windows/macOS'te YANLIŞ
import multiprocessing as mp

def worker():
    print("çalışıyor")

p = mp.Process(target=worker)
p.start()  # Bu satır oluşturulan her alt süreç tarafından tekrar çalıştırılır = kaos

# DOĞRU
if __name__ == "__main__":
    p = mp.Process(target=worker)
    p.start()
    p.join()

Hata 3: asyncio event loop’unu bloklamak

Bir coroutine içinde doğrudan time.sleep() çağırmak veya ağır CPU işi yapmak, tüm event loop’u dondurur.

# YANLIŞ
async def bad():
    time.sleep(5)  # HER ŞEYİ bloklar

# DOĞRU
async def good():
    await asyncio.sleep(5)  # Bloklamaz
    # ya da CPU ağırlıklı işi devret:
    await asyncio.to_thread(cpu_heavy_function)

Hata 4: Korunmayan paylaşılan durumdan kaynaklanan yarış koşulları

# YANLIŞ - yarış koşulu
counter = 0
def increment():
    global counter
    counter += 1  # Atomik değil

# DOĞRU
lock = threading.Lock()
def increment_safe():
    global counter
    with lock:
        counter += 1

Hata 5: Tutarsız kilit sıralamasından kaynaklanan deadlock’lar

Kod tabanı genelinde birden fazla kilidi her zaman tutarlı bir global sırayla alın.

Hata 6: Görevleri join/await etmeyi unutmak (sahipsiz kalan işler)

# Thread'ler: t.join() çağırmayı unutmak, program iş bitmeden çıkmasına
# ya da daemon yapılmadıysa zombi thread'ler kalmasına neden olabilir.

# Asyncio: bir görevi await etmeyi unutmak "fırlat ve unut" (fire and forget)
# anlamına gelir — içindeki istisnalar, bir referans tutup kontrol etmediğiniz
# ya da bunu doğru şekilde ele alan TaskGroup kullanmadığınız sürece
# sessizce yutulur.

Hata 7: Süreçler arasında mutable varsayılan argümanlar / paylaşılan nesneler kullanmak

Büyük, karmaşık veya pickle’lanamayan nesneleri multiprocessing.Process/Pool‘a geçirmek hatalara veya serileştirme ek yüküne neden olur. Worker fonksiyon argümanlarını küçük ve basit tutun (sayılar, string’ler, primitif listeler).

Hata 8: Ağır üretici yükü altında kuyruklar için maksimum boyut belirlememek

maxsize olmadan, tüketici yetişemiyorsa hızlı bir üretici sınırsız bellek büyümesine neden olabilir.

İyi uygulamalar özeti

  1. Manuel thread/süreç yönetimi yerine yüksek seviyeli API’leri (concurrent.futures, asyncio.TaskGroup) tercih edin.
  2. Temizliği garanti etmek için context manager’ları (with lock:, with ThreadPoolExecutor() as ex:) kullanın.
  3. Kritik bölümleri (kilitli kodu) olabildiğince kısa tutun.
  4. Mümkün olduğunda paylaşılan mutable durum yerine mesaj geçişini (kuyruklar) tercih edin.
  5. Future/görevlerden gelen istisnaları her zaman ele alın — bunlar programı otomatik olarak çökertmez ama sessizce kaybolabilirler.
  6. Paralelleştirmeden önce profilleyin — eşzamanlılık karmaşıklık ekler ve bazen ek yük, küçük görevler için faydadan ağır basar.
  7. Hedef işletim sisteminde test edin — fork ve spawn farkları yalnızca belirli platformlarda ortaya çıkan hatalara neden olabilir.
  8. Süresiz asılı kalmayı önlemek için kilitlerde, join’lerde ve ağ çağrılarında zaman aşımları kullanın.

11. İleri Düzey Konular

asyncio ve thread’leri birlikte kullanmak: event loop’a güvenli şekilde geri çağrı yapmak

Bir arka plan thread’inin asyncio ile etkileşime girmesi gerektiğinde, run_coroutine_threadsafe kullanın:

import asyncio
import threading

async def notify(message):
    print(f"Bildirildi: {message}")

def background_worker(loop):
    import time
    time.sleep(1)
    asyncio.run_coroutine_threadsafe(notify("thread'den tamamlandı"), loop)

async def main():
    loop = asyncio.get_running_loop()
    threading.Thread(target=background_worker, args=(loop,)).start()
    await asyncio.sleep(2)

asyncio.run(main())

Yapılandırılmış eşzamanlılık (structured concurrency) felsefesi

Modern eşzamanlı programlama, yapılandırılmış eşzamanlılığı tercih eder: eşzamanlı görevlerin, sızabilen ya da amaçlanan bağlamından daha uzun yaşayabilen “fırlat ve unut” görevleri yerine, bir kapsama (scope) bağlı net ve sınırlı yaşam sürelerine sahip olması gerekir (TaskGroup veya async with gibi). Bu, hata yönetimini ve iptali öngörülebilir hale getirir.

Multiprocessing’i asyncio ile birleştirmek

Async bir uygulama içindeki CPU ağırlıklı işler için, bir ProcessPoolExecutor‘a devredin:

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(n):
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy, 10_000_000)
        print(result)

if __name__ == "__main__":
    asyncio.run(main())

multiprocessing.shared_memory (Python 3.8+)

Süreçler arasında gerçek sıfır kopyalama (zero-copy) paylaşılan bellek bloklarına izin verir, NumPy dizileri gibi büyük veriler için pickle ek yükünden kaçınır.

from multiprocessing import shared_memory
import numpy as np

# Paylaşılan bellek bloğu oluştur
arr = np.array([1, 2, 3, 4, 5])
shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
shared_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
shared_arr[:] = arr[:]

# Başka bir süreçte, shm.name üzerinden aynı bloğa bağlan
existing_shm = shared_memory.SharedMemory(name=shm.name)
attached_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=existing_shm.buf)
print(attached_arr)

shm.close()
shm.unlink()  # paylaşılan belleği serbest bırak

Asenkron comprehension’lar

import asyncio

async def get_value(x):
    await asyncio.sleep(0.1)
    return x * 2

async def main():
    results = [await get_value(x) for x in range(5)]
    print(results)

    # Asenkron generator comprehension
    async def gen():
        for i in range(5):
            yield i

    results2 = [x async for x in gen()]
    print(results2)

asyncio.run(main())

uvloop — daha hızlı bir event loop

libuv üzerine inşa edilmiş, doğrudan değiştirilebilir (drop-in replacement) bir event loop; I/O ağırlıklı asyncio uygulamaları için önemli performans kazançları sağlar (standart kütüphanenin parçası değildir; pip ile kurulur).

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

C uzantılarında GIL serbest bırakma

NumPy gibi kütüphaneler, ağır sayısal işlemler sırasında GIL’i serbest bırakır; bu da, bu tür çağrıların hakim olduğu iş yükleri için threading’in (saf Python döngülerinin aksine) gerçek hız kazanımları sağlayabileceği anlamına gelir.

12. Özet Tablo

Özellikthreadingmultiprocessingasyncio
Paralellik (çoklu çekirdek)Hayır (GIL sınırlı)EvetHayır (tek thread’li)
En iyi kullanım alanıI/O ağırlıklı, az görevCPU ağırlıklıI/O ağırlıklı, çok görev
BellekPaylaşılanSüreç başına ayrıPaylaşılan (tek süreç)
Birim başına ek yükOrtaYüksekÇok düşük
İletişimPaylaşılan değişkenler + kilitlerPipe’lar, Kuyruklar, paylaşılan bellekPaylaşılan değişkenler (tek loop) + kuyruklar
Hata izolasyonuDüşük (paylaşılan bellek)Yüksek (ayrı süreç)Düşük (tek süreç)
Hata ayıklama zorluğuOrta-YüksekOrtaOrta (farklı zihinsel model)
Açık await/callback gerektirirHayırHayırEvet

Hızlı referans import’ları

import threading                                  # Thread, Lock, Event, Condition, Semaphore
import multiprocessing as mp                      # Process, Pool, Queue, Value, Array, Manager
import asyncio                                    # coroutine'ler, task'lar, event loop
from concurrent.futures import (
    ThreadPoolExecutor, ProcessPoolExecutor, as_completed
)
import queue                                      # Queue, LifoQueue, PriorityQueue (thread-safe)

Altın kurallar

  1. I/O ağırlıklı + az görev → threading.
  2. I/O ağırlıklı + çok görev → asyncio.
  3. CPU ağırlıklı → multiprocessing.
  4. Manuel thread/süreç yönetimi yerine basit paralel map işlemleri için concurrent.futures‘ı tercih edin.
  5. Bloklayan kodu asla doğrudan bir coroutine içinde çağırmayın — devredin.
  6. Multiprocessing giriş noktalarını her zaman if __name__ == "__main__": ile koruyun.
  7. Paylaşılan mutable durum yerine mesaj geçişini (kuyruklar) tercih edin.
  8. Optimize etmeden önce ölçün — eşzamanlılık bedavaya gelmez.