Asynchronous Programming

Tahap ingestion kadang harus memanggil ratusan endpoint API sekaligus. Membuat ratusan utas untuk pekerjaan itu terasa mahal: setiap utas membawa stack sendiri dan penjadwalannya makin berat seiring jumlahnya naik. asyncio menawarkan model lain — satu utas dengan banyak coroutine yang menyerahkan kendali secara sukarela di setiap titik await. Kita bahas dasar async/await, lalu asyncio.gather untuk menjalankan banyak tugas bersamaan, dan terakhir membandingkannya dengan threading.

Coroutine dan await #

Fungsi yang didefinisikan dengan async def menghasilkan objek coroutine saat dipanggil — isinya belum berjalan. Isi coroutine hanya dieksekusi oleh event loop, dan kata kunci await menandai titik di mana coroutine bersedia menunggu tanpa menghalangi utas.

import asyncio


async def salam(nama):
    print(f"mulai   : menyapa {nama}")
    await asyncio.sleep(0.1)  # simulasi menunggu respons jaringan
    print(f"selesai : menyapa {nama}")
    return f"halo, {nama}"


hasil = asyncio.run(salam("Budi"))
print("nilai balikan:", hasil)

Output yang Dihasilkan:

mulai   : menyapa Budi
selesai : menyapa Budi
nilai balikan: halo, Budi

Mari kita telaah baris-baris kode di atas:

1. Baris 4 (async def salam): pemanggilan salam("Budi") tidak menjalankan isi fungsi; pemanggilan hanya membuat objek coroutine siap dijadwalkan.
2. Baris 6 (await asyncio.sleep(0.1)): coroutine berhenti di titik ini dan memberi tahu event loop bahwa utas boleh dipakai untuk hal lain. Setelah 0,1 detik berlalu, eksekusi dilanjutkan di baris berikutnya.
3. Baris 10 (asyncio.run): fungsi ini membuat event loop, menjalankan coroutine sampai selesai, mengembalikan nilai balikannya, lalu menutup loop. asyncio.run adalah pintu masuk tunggal program async.

asyncio.gather untuk Banyak Tugas #

Sekarang kita terapkan pola tersebut pada kasus ingestion: mengambil data dari tiga endpoint api.example.com yang masing-masing merespons dengan latensi berbeda. Versi sekuensial menunggu satu demi satu; versi gather menunggu semuanya bersamaan.

import asyncio
import time

ENDPOINTS = [
    ("api.example.com/v1/pelanggan", 0.30),
    ("api.example.com/v1/produk", 0.20),
    ("api.example.com/v1/transaksi", 0.25),
]


async def fetch(endpoint, jeda):
    # simulasi HTTP GET: di dunia nyata ini aiohttp/httpx async
    print(f"GET {endpoint}")
    await asyncio.sleep(jeda)
    baris = 100 + (len(endpoint) % 5) * 25
    return f"{endpoint} -> 200 OK ({baris} baris)"


async def ingestion_sekuensial():
    hasil = []
    for endpoint, jeda in ENDPOINTS:
        hasil.append(await fetch(endpoint, jeda))
    return hasil


async def ingestion_gather():
    tugas = [fetch(endpoint, jeda) for endpoint, jeda in ENDPOINTS]
    return await asyncio.gather(*tugas)


async def main():
    mulai = time.perf_counter()
    hasil_seq = await ingestion_sekuensial()
    durasi_seq = time.perf_counter() - mulai
    print(f"-- sekuensial {durasi_seq:.2f} detik --")

    mulai = time.perf_counter()
    hasil_gen = await ingestion_gather()
    durasi_gen = time.perf_counter() - mulai
    print(f"-- gather     {durasi_gen:.2f} detik --")

    print("hasil identik:", hasil_seq == hasil_gen)
    for baris in hasil_gen:
        print("data:", baris)
    print("percepatan:", round(durasi_seq / durasi_gen, 1), "x")


asyncio.run(main())

Output yang Dihasilkan:

GET api.example.com/v1/pelanggan
GET api.example.com/v1/produk
GET api.example.com/v1/transaksi
-- sekuensial 0.75 detik --
GET api.example.com/v1/pelanggan
GET api.example.com/v1/produk
GET api.example.com/v1/transaksi
-- gather     0.30 detik --
hasil identik: True
data: api.example.com/v1/pelanggan -> 200 OK (175 baris)
data: api.example.com/v1/produk -> 200 OK (100 baris)
data: api.example.com/v1/transaksi -> 200 OK (175 baris)
percepatan: 2.5 x

Mari kita telaah baris-baris kode di atas:

1. Baris 11-16 (fetch): coroutine tiruan untuk HTTP GET. Baris GET ... tercetak begitu tugas mulai; setelah itu coroutine menunggu selama jeda. Ukuran data pada baris 15 dihitung dari panjang nama endpoint agar output uji dapat diulang identik.
2. Baris 19-23 (ingestion_sekuensial): await di dalam perulangan memaksa setiap panggilan selesai sebelum panggilan berikutnya dimulai, sehingga durasinya adalah jumlah semua jeda: 0,30 + 0,20 + 0,25 = 0,75 detik.
3. Baris 26-28 (ingestion_gather): list comprehension membuat tiga objek coroutine tanpa menjalankannya. asyncio.gather lalu menjadwalkan ketiganya sekaligus — terlihat dari tiga baris GET yang tampil berurutan tanpa jeda — dan total durasinya menjadi jeda terpanjang saja: 0,30 detik.
4. Baris 31-45 (main): kedua jalur diukur dengan time.perf_counter. Baris hasil identik: True menegaskan kedua strategi menghasilkan data yang sama persis.

Async vs Threading #

Keduanya menyelesaikan masalah yang sama — beban I/O — dengan mekanisme yang berbeda. Perbandingan berikut membantu memilih.

Untuk ingestion sehari-hari dari belasan sampai ratusan sumber, pool utas dari Chapter Concurrent Processing biasanya sudah memadai dan lebih sederhana. Async menjadi unggulan ketika jumlah koneksi bersamaan naik ke ribuan — biaya utas dan penjadwalannya mulai terasa, sementara coroutine tetap ringan.

Rujukan resmi untuk pendalaman: dokumentasi asyncio dan halaman menjalankan tugas bersamaan . Dengan konkurensi dan asinkron di tangan, kita siap menghadapi data yang volumenya jauh melampaui memori satu mesin — lanjut ke Chapter Pemrosesan Data Skala Besar.