%%{init: {"flowchart": {"diagramPadding": 40}}}%%
flowchart TD
A["20 stasiun"] --> B["Semaphore: paling banyak 10 sekaligus"]
B --> C["ambil() dengan @coba_ulang (3 kali)"]
C -- "berhasil" --> D["Data cuaca (JSON)"]
C -- "gagal 3 kali" --> E["None (dilaporkan gagal)"]
D --> F["gather: hasil berurutan"]
E --> F
Bab 35 — Async dan Concurrency: Mengerjakan Banyak Hal Sekaligus
Seri I4 — Bahasa Python Tingkat Lanjut
Semua skrip bab ini ada di folder kode/bab35/. Di versi web, Anda juga bisa mengunduh semuanya sekaligus (zip).
Bayangkan Anda mengambil data curah hujan dari 20 stasiun. Setiap permintaan ke jaringan butuh waktu menunggu, misalnya setengah detik. Dikerjakan satu per satu: 20 x 0,5 = 10 detik, dan hampir semuanya dihabiskan untuk menunggu. Seperti petani yang menyiram 20 petak dengan satu selang, menunggu air sampai di tiap petak sebelum pindah ke petak berikutnya. Padahal selama menunggu, tangannya bisa mengurus petak lain.
Itulah inti bab ini: mengatur supaya waktu menunggu tidak terbuang.
Dua jenis pekerjaan
| Jenis | Contoh | Yang jadi penghambat |
|---|---|---|
| I/O-bound (banyak menunggu) | unduh data, baca disk, tanya basis data | menunggu jaringan atau disk |
| CPU-bound (banyak menghitung) | simulasi, olah angka berat | kecepatan prosesor |
Pilihan alat bergantung pada jenis ini. Tiga alat utamanya:

asyncio: satu pekerja yang pandai bergantian
Fungsi biasa dikerjakan dari awal sampai akhir. Fungsi async def (disebut coroutine) bisa berhenti sementara di titik await dan menyerahkan giliran ke pekerjaan lain yang sedang menunggu. Satu pekerja (satu thread, disebut event loop) mengatur gilirannya.
01_coroutine_dasar.py
# Bab 35 - Dasar async: coroutine, await, dan asyncio.run
import asyncio
async def ukur_ph(kolam):
await asyncio.sleep(0.01) # menunggu tanpa memblokir program lain
return f"pH kolam {kolam} = 7.2"
# Memanggil fungsi async TIDAK menjalankannya: ia hanya membuat objek coroutine
objek = ukur_ph("A")
print(type(objek).__name__)
objek.close() # dibuang tanpa dijalankan
async def utama():
hasil = await ukur_ph("A") # await: jalankan dan tunggu hasilnya
print(hasil)
asyncio.run(utama()) # pintu masuk: jalankan event loopcoroutine
pH kolam A = 7.2
- Memanggil
ukur_ph("A")tidak menjalankannya; hanya membuat objek bertipecoroutine. (Di skrip, objek itu langsung ditutup denganclose().) Melupakanawaitadalah kesalahan pemula yang sangat umum, dan biasanya hanya menghasilkan peringatan. await ukur_ph("A")menjalankan dan menunggu hasilnya.asyncio.run(utama())adalah pintu masuk yang menyalakan event loop. Biasanya dipanggil sekali di bagian paling luar program.asyncio.sleep(0.01)adalah menunggu yang tidak memblokir. Bandingkan dengantime.sleep()biasa yang membekukan seluruh program.
TaskGroup, batas waktu, dan pembatas jumlah
03_taskgroup_timeout.py
# Bab 35 - TaskGroup, batas waktu, dan pembatas jumlah (Semaphore)
import asyncio
aktif = 0
puncak = 0
async def kerja(nomor, semaphore):
global aktif, puncak
async with semaphore: # paling banyak 2 yang masuk sekaligus
aktif += 1
puncak = max(puncak, aktif)
await asyncio.sleep(0.02)
aktif -= 1
return nomor * 10
async def lambat():
await asyncio.sleep(1)
async def utama():
semaphore = asyncio.Semaphore(2)
async with asyncio.TaskGroup() as grup:
tugas = [grup.create_task(kerja(i, semaphore)) for i in range(1, 7)]
print("hasil:", [t.result() for t in tugas])
print("paling banyak berjalan bersamaan:", puncak)
try:
async with asyncio.timeout(0.05):
await lambat()
except TimeoutError:
print("dibatalkan: melewati batas waktu")
asyncio.run(utama())hasil: [10, 20, 30, 40, 50, 60]
paling banyak berjalan bersamaan: 2
dibatalkan: melewati batas waktu
asyncio.TaskGroup(Python 3.11+) memulai sekelompok tugas dan menunggu semuanya saat blokasync withberakhir. Bila satu tugas gagal, yang lain dibatalkan: lebih aman daripadagathermentah.asyncio.Semaphore(2)membatasi hanya dua tugas yang masuk bagian dalamnya sekaligus. Buktinya: dari enam tugas,puncaktertinggi yang berjalan bersamaan adalah 2. Ini penting agar server tidak dibanjiri ratusan permintaan sekaligus.asyncio.timeout(0.05)(Python 3.11+) membatalkan pekerjaan yang melebihi batas, lalu menimbulkanTimeoutError. Pekerjaanlambat()yang butuh 1 detik dihentikan setelah 0,05 detik.
Selalu beri batas waktu (timeout) dan batas jumlah (Semaphore) pada pekerjaan jaringan. Server luar bisa lambat atau macet, dan mengirim 1.000 permintaan sekaligus bisa dianggap serangan.
Proses: untuk pekerjaan menghitung
05_multiprocessing_cpu.py
# Bab 35 - Proses terpisah untuk pekerjaan menghitung (CPU)
from concurrent.futures import ProcessPoolExecutor
def hitung_berat(n):
"""Pekerjaan CPU murni: jumlah kuadrat 1..n."""
return sum(i * i for i in range(1, n + 1))
if __name__ == "__main__": # WAJIB di Windows: tiap proses anak mengimpor berkas ini
tugas = [200_000, 300_000, 400_000, 500_000]
with ProcessPoolExecutor(max_workers=4) as kolam:
hasil = list(kolam.map(hitung_berat, tugas))
for n, h in zip(tugas, hasil):
print(n, h)200000 2666686666700000
300000 9000045000050000
400000 21333413333400000
500000 41666791666750000
ProcessPoolExecutor menjalankan fungsi di proses terpisah: tiap proses adalah program Python tersendiri dengan memorinya sendiri. Hasil dikirim kembali ke program utama, dan map menjaga urutannya (200.000, 300.000, 400.000, 500.000).
Di Windows, kode yang membuat proses baru wajib berada di bawah if __name__ == "__main__":. Alasannya: Windows memulai proses anak dengan metode spawn, yaitu menjalankan interpreter baru yang mengimpor berkas Anda dari awal. Tanpa pelindung itu, setiap proses anak akan mengulang kode pembuat proses dan menghasilkan galat RuntimeError. Dokumentasi multiprocessing menyebut spawn sebagai metode bawaan di Windows dan macOS, dan menganjurkan pelindung ini (“safe importing of main module”). Skrip di bab ini selalu memakainya bila membuat proses. Karena fungsi dan data dikirim antar-proses, keduanya harus bisa di-pickle (diserialisasi); fungsi harus didefinisikan di tingkat atas modul, bukan lambda (penulis mencoba lambda dan mendapat PicklingError).
GIL: mengapa thread tidak selalu mempercepat
Python standar punya GIL (Global Interpreter Lock), kunci yang memastikan hanya satu thread menjalankan kode Python pada satu saat dalam satu proses. Akibatnya, thread tidak mempercepat pekerjaan CPU; thread berguna untuk pekerjaan menunggu karena saat satu thread menunggu, yang lain bisa jalan. Mari buktikan di komputer ini:
06_gil_dan_cpu.py
# Bab 35 - GIL: thread tidak mempercepat pekerjaan CPU (pada build biasa), proses bisa
import sys
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
def hitung_berat(n):
return sum(i * i for i in range(n))
def ukur(fungsi):
t0 = time.perf_counter()
fungsi()
return time.perf_counter() - t0
if __name__ == "__main__":
print("GIL aktif di Python ini:", sys._is_gil_enabled())
beban = [5_000_000] * 4
urut = ukur(lambda: [hitung_berat(n) for n in beban])
with ThreadPoolExecutor(max_workers=4) as kolam:
thread = ukur(lambda: list(kolam.map(hitung_berat, beban)))
with ProcessPoolExecutor(max_workers=4) as kolam:
proses = ukur(lambda: list(kolam.map(hitung_berat, beban)))
print("4 thread mempercepat lebih dari 1,5x:", urut / thread > 1.5)
print("4 proses mempercepat lebih dari 1,3x:", urut / proses > 1.3)GIL aktif di Python ini: True
4 thread mempercepat lebih dari 1,5x: False
4 proses mempercepat lebih dari 1,3x: True
sys._is_gil_enabled()melaporkan apakah GIL aktif di Python yang menjalankan skrip:True.- Empat thread tidak mempercepat hitungan berat lebih dari 1,5 kali (
False): kuncinya membuat mereka bergantian. - Empat proses mempercepat lebih dari 1,3 kali (
True), karena tiap proses punya interpreter dan GIL sendiri sehingga benar-benar berjalan paralel. - Seperti sebelumnya, skrip hanya mencetak perbandingan, bukan detik. Hasil
True/Falseini diperiksa berkali-kali di komputer penulis; pada komputer Anda angka pastinya akan berbeda (misalnya bila inti prosesor lebih sedikit), sehingga ambang 1,3 kali bisa tidak tercapai di komputer dengan sedikit inti.
Mode free-threaded
Sejak Python 3.13 ada build opsional free-threaded dengan GIL dimatikan, sehingga thread bisa berjalan paralel di banyak inti. Menurut dokumentasi, Python 3.14 menyatakan dukungan resmi untuk mode ini (rujukan PEP 779 di bagian “What’s New”), tetapi ia bukan bawaan: harus dipilih saat memasang atau membuat build tersendiri, dan tidak semua pustaka C siap untuk itu. Cara mengeceknya: sys._is_gil_enabled() dan sysconfig.get_config_var("Py_GIL_DISABLED"). Di komputer penulis hanya build biasa (GIL aktif) yang diuji; build free-threaded belum diuji di bab ini.
Panduan memilih:
| Masalah Anda | Pilihan | Alasan |
|---|---|---|
Ratusan permintaan jaringan, pustaka mendukung async |
asyncio | ringan, satu thread, mudah dibatasi |
| Menunggu, tetapi pustaka lama memblokir | thread (ThreadPoolExecutor) |
tanpa menulis ulang jadi async |
| Hitungan berat murni (CPU) | proses (ProcessPoolExecutor) |
melewati GIL dengan banyak interpreter |
| Hanya beberapa pekerjaan, sederhana | tidak perlu semuanya | kode biasa lebih mudah dirawat |
Proyek Bersambung I4: mengunduh banyak data cuaca secara async
Puncak proyek Seri I4: mengambil data 20 stasiun cuaca/harga sekaligus. Agar aman dan stabil, server-nya tiruan: dibuat di komputer ini sendiri dengan aiohttp, di alamat lokal 127.0.0.1 pada port yang dipilih sistem. Tidak ada internet yang dipakai. Server meniru jaringan lambat (0,02 detik) dan gangguan: stasiun S07 gagal sekali (status 503), S13 gagal tiga kali.
07_proyek_unduh_async.py
# Bab 35 - Proyek bersambung I4: mengunduh banyak data cuaca dari server TIRUAN lokal
import asyncio
import logging
import sys
from functools import wraps
import aiohttp
from aiohttp import web
logging.basicConfig(stream=sys.stdout, level=logging.WARNING, format="%(levelname)s %(message)s")
log = logging.getLogger("unduh")
gangguan = {"S07": 1, "S13": 3} # stasiun tiruan yang gagal beberapa kali
# ---------- server tiruan (hanya di komputer ini, tanpa internet) ----------
async def beri_data(request):
stasiun = request.match_info["stasiun"]
await asyncio.sleep(0.02) # meniru lambatnya jaringan
if gangguan.get(stasiun, 0) > 0:
gangguan[stasiun] -= 1
return web.Response(status=503, text="sibuk")
nomor = int(stasiun[1:])
return web.json_response({"stasiun": stasiun, "curah_hujan_mm": nomor * 1.5,
"harga_gabah": 6000 + nomor * 10})
# ---------- klien async dengan coba-ulang (decorator Bab 31, versi async) ----------
def coba_ulang(maksimal=3):
def decorator(fungsi):
@wraps(fungsi)
async def pembungkus(sesi, alamat, stasiun):
for percobaan in range(1, maksimal + 1):
try:
return await fungsi(sesi, alamat, stasiun)
except aiohttp.ClientResponseError as e:
log.warning("%s percobaan %d: status %d", stasiun, percobaan, e.status)
if percobaan == maksimal:
raise
await asyncio.sleep(0.01)
return pembungkus
return decorator
@coba_ulang(maksimal=3)
async def ambil(sesi, alamat, stasiun):
async with sesi.get(f"{alamat}/data/{stasiun}") as jawaban:
jawaban.raise_for_status()
return await jawaban.json()
async def ambil_semua(sesi, alamat, daftar, batas):
semaphore = asyncio.Semaphore(batas)
async def satu(stasiun):
async with semaphore:
try:
return await ambil(sesi, alamat, stasiun)
except aiohttp.ClientResponseError:
return None
return await asyncio.gather(*(satu(s) for s in daftar))
async def utama():
aplikasi = web.Application()
aplikasi.add_routes([web.get("/data/{stasiun}", beri_data)])
runner = web.AppRunner(aplikasi)
await runner.setup()
situs = web.TCPSite(runner, "127.0.0.1", 0) # port 0: sistem memilihkan port bebas
await situs.start()
port = situs._server.sockets[0].getsockname()[1]
alamat = f"http://127.0.0.1:{port}"
daftar = [f"S{n:02d}" for n in range(1, 21)]
async with aiohttp.ClientSession() as sesi:
loop = asyncio.get_running_loop()
t0 = loop.time()
hasil = await ambil_semua(sesi, alamat, daftar, batas=10)
waktu = loop.time() - t0
await runner.cleanup()
berhasil = [h for h in hasil if h]
gagal = [s for s, h in zip(daftar, hasil) if h is None]
print("stasiun diminta:", len(daftar), "| berhasil:", len(berhasil), "| gagal:", gagal)
rata = sum(h["curah_hujan_mm"] for h in berhasil) / len(berhasil)
print("rata-rata curah hujan (mm):", round(rata, 2))
print("harga gabah tertinggi di stasiun:", max(berhasil, key=lambda h: h["harga_gabah"])["stasiun"])
print("lebih cepat dari satu per satu (20 x 0,02 detik = 0,4 detik):", waktu < 0.3)
if __name__ == "__main__":
asyncio.run(utama())WARNING S07 percobaan 1: status 503
WARNING S13 percobaan 1: status 503
WARNING S13 percobaan 2: status 503
WARNING S13 percobaan 3: status 503
stasiun diminta: 20 | berhasil: 19 | gagal: ['S13']
rata-rata curah hujan (mm): 15.55
harga gabah tertinggi di stasiun: S20
lebih cepat dari satu per satu (20 x 0,02 detik = 0,4 detik): True
Telusuri hasilnya:
- Decorator
coba_ulangdari Bab 31 dipakai lagi, kini versiasync(denganawait asyncio.sleepuntuk jeda). Ini menyambungkan Bab 31 dan 35. - Log peringatan menunjukkan: S07 gagal pada percobaan 1 lalu berhasil pada percobaan 2 (tidak ada peringatan kedua), sedangkan S13 gagal pada percobaan 1, 2, dan 3, sehingga menyerah dan hasilnya
None. - Ringkasan: 20 diminta, 19 berhasil, gagal:
['S13']. Rata-rata curah hujan dihitung hanya dari yang berhasil: 15,55 mm (angka tiruan: stasiun nomor n mengembalikan n x 1,5). Harga gabah tertinggi di stasiun S20. - Dengan
Semaphore(10), dua puluh permintaan selesai jauh lebih cepat daripada satu per satu (20 x 0,02 = 0,4 detik): perbandingannyaTrue. Skrip tidak mencetak detik. - Server dan klien berjalan dalam satu event loop yang sama, sehingga seluruh percobaan muat dalam satu skrip.
- Server dimatikan rapi dengan
runner.cleanup(); sesi klien ditutup olehasync with, yaitu context manager asinkron (kerabat async dari Bab 32).
Ini simulasi. Pada API sungguhan, ikuti batas laju (rate limit) dan aturan penggunaan penyedia data, dan jangan menembaknya dengan ratusan permintaan tanpa izin. Contoh di buku ini tidak menghubungi layanan luar mana pun.
Latihan
- Ubah
02_gather_serempak.py: tambahkan stasiun keempat dengan lama menunggu 0,01 detik. Perkirakan urutan selesai serempak lebih dulu, lalu jalankan dan cocokkan. - Ubah
Semaphore(2)di03_taskgroup_timeout.pymenjadi 3. Berapapuncakyang tercetak sekarang? Mengapa? - Ubah
batas=10di07_proyek_unduh_async.pymenjadibatas=1. Hasil ringkasan (berhasil dan gagal) tetap sama; hanya baris terakhir (perbandingan kecepatan) berubah. Mengapa? - Tantangan: tambahkan stasiun ketiga yang rusak,
S18, kegangguandengan nilai 5 (gagal lima kali), jalankan ulang, dan hitung ulang rata-rata curah hujan secara manual untuk memastikan cocok dengan keluaran.
Ringkasan Bab
- Bedakan pekerjaan menunggu (I/O) dan menghitung (CPU); pilihan alat bergantung padanya.
asyncio:async def+await+asyncio.run;gathermenjaga urutan hasil;TaskGroup,timeout, danSemaphoremengatur dan membatasi tugas.- Thread (
ThreadPoolExecutor) cocok untuk menunggu pada kode biasa; data bersama dilindungiLock. - Proses (
ProcessPoolExecutor) cocok untuk menghitung; di Windows wajibif __name__ == "__main__":. - GIL membuat thread tidak mempercepat pekerjaan CPU di Python standar; build free-threaded adalah pilihan terpisah.
- Proyek I4: 20 stasiun diunduh serempak dari server tiruan, dengan coba-ulang dan batas jumlah.
Penutup Seri I4
Selamat! Anda menyelesaikan Seri I4 dan Tingkat Intermediate jilid kedua ini. Perhatikan apa yang sekarang bisa dilakukan kode Anda dibandingkan di awal seri:
- membaca berkas besar satu baris per satu tanpa membebani memori (generator, Bab 30),
- membungkus fungsi dengan log dan coba-ulang tanpa mengubah isinya (decorator, Bab 31),
- membuka dan menutup sumber daya dengan pasti, dan menulis laporan dengan aman (context manager, Bab 32),
- berlabel tipe dan diperiksa mypy sebelum berjalan (type hint, Bab 33),
- menolak data lapangan yang buruk dengan pesan jelas dan membaca pengaturan dengan aman (Pydantic, Bab 34),
- mengunduh banyak data serempak dengan batas dan coba-ulang (async, Bab 35).
Kode sudah kuat. Seri I5 mengubahnya menjadi aplikasi yang bisa dipakai orang lain: layanan web (Bab 36), Python di browser (Bab 37), peta dan data spasial (Bab 38 sampai 40), otomasi dan antarmuka (Bab 41 dan 42), serta dasbor (Bab 43). Pydantic dan async yang baru Anda pelajari akan langsung dipakai di FastAPI.
Selanjutnya, di Bab 36: web development, yaitu Flask, FastAPI, pengantar Django, dan REST API.