Asynchronous Programming di Python dengan asyncio: Menjalankan Task secara Concurrent

Mengapa Asynchronous Programming Penting?

Saat aplikasi berinteraksi dengan sumber daya eksternal — database, API, sistem file — thread sering menunggu idle hingga respons diterima. Dalam model sinkron, setiap operasi I/O memblokir eksekusi seluruh program. Satu request HTTP yang membutuhkan 2 detik akan menghentikan semua proses lain selama 2 detik tersebut.

Asynchronous programming menyelesaikan masalah ini dengan membiarkan event loop menjalankan task lain sambil menunggu operasi I/O selesai. Alih-alih menunggu diam, program bisa mengerjakan hal lain. Kita akan mempelajari asyncio, library bawaan Python untuk menangani concurrency berbasis coroutine. Pendekatan ini memberikan throughput tinggi tanpa kompleksitas threading dan tanpa risiko race condition yang sering muncul pada multi-threaded programming.

Coroutines dan Event Loop

Coroutine adalah fungsi yang didefinisikan dengan async def dan dapat di-suspend di tengah eksekusi menggunakan await. Ketika sebuah coroutine menemui await, kontrol dikembalikan ke event loop, yang kemudian menjadwalkan coroutine lain yang siap berjalan. Mekanisme ini memungkinkan satu thread menangani ribuan koneksi I/O secara bergantian tanpa memblokir satu sama lain.

import asyncio

async def fetch_data(url: str, delay: int) -> str:
    print(f"Mulai fetch: {url}")
    await asyncio.sleep(delay)  # simulasi I/O
    print(f"Selesai fetch: {url}")
    return f"Data dari {url}"

async def main():
    result = await fetch_data("https://api.example.com/users", 2)
    print(result)

asyncio.run(main())

Output:

Mulai fetch: https://api.example.com/users
Selesai fetch: https://api.example.com/users
Data dari https://api.example.com/users

Pada contoh di atas, asyncio.sleep() mensimulasikan operasi I/O. Selama await asyncio.sleep(2) berlangsung, thread tidak diblokir — event loop dapat menjadwalkan task lain jika ada. Fungsi asyncio.run() bertindak sebagai entry point yang membuat event loop dan menjalankan coroutine hingga selesai.

Menjalankan Banyak Task secara Concurrent

Menjalankan satu coroutine pada satu waktu tidak menunjukkan kekuatan asyncio. Kita perlu asyncio.gather() untuk menjalankan beberapa coroutine secara paralel dalam event loop yang sama. Fungsi ini menerima sejumlah awaitable, menjadwalkan semuanya, dan mengembalikan hasil dalam urutan yang sama dengan input.

import asyncio

async def fetch_data(url: str, delay: int) -> str:
    print(f"Mulai fetch: {url}")
    await asyncio.sleep(delay)
    return f"Data dari {url}"

async def main():
    tasks = [
        fetch_data("https://api.example.com/users", 2),
        fetch_data("https://api.example.com/posts", 3),
        fetch_data("https://api.example.com/comments", 1),
    ]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

asyncio.run(main())

Output:

Mulai fetch: https://api.example.com/users
Mulai fetch: https://api.example.com/posts
Mulai fetch: https://api.example.com/comments
Data dari https://api.example.com/users
Data dari https://api.example.com/posts
Data dari https://api.example.com/comments

Tanpa gather(), ketiga operasi berjalan secara berurutan dengan total 6 detik (2+3+1). Dengan gather(), semua task dimulai bersamaan dan selesai dalam 3 detik — sama dengan durasi operasi terlama. Output menunjukkan bahwa print pada setiap coroutine dieksekusi secara concurrent, bukan berurutan.

Mengelola Task dengan asyncio.create_task()

asyncio.gather() cocok untuk menjalankan batch coroutine dan mengumpulkan semua hasilnya sekaligus. Tetapi terkadang kita perlu kontrol lebih granular — memulai task di satu titik, mengerjakan hal lain, lalu mengambil hasilnya nanti. asyncio.create_task() memungkinkan penjadwalan coroutine ke event loop sekaligus mengambil hasilnya di kemudian waktu.

import asyncio

async def process_item(item: str) -> str:
    await asyncio.sleep(1)
    return f"Processed: {item}"

async def main():
    task1 = asyncio.create_task(process_item("A"))
    task2 = asyncio.create_task(process_item("B"))

    # Melakukan pekerjaan lain sambil menunggu
    print("Menunggu task selesai...")

    result1 = await task1
    result2 = await task2
    print(result1, result2)

asyncio.run(main())

Output:

Menunggu task selesai...
Processed: A Processed: B

Pola ini berguna ketika kita ingin memulai background task dan melanjutkan eksekusi kode lain sebelum menunggu hasilnya. Task yang sudah dijadwalkan berjalan di background hingga kita await hasilnya. Jika task selesai sebelum di-await, hasilnya tersimpan dan bisa diakses langsung tanpa menunggu lagi.

Timeout dan Error Handling

Dalam produksi, operasi I/O bisa gagal atau hang tanpa respons. asyncio.wait_for() memberikan batas waktu agar program tidak menunggu tanpa akhir. Jika operasi melebihi durasi yang ditentukan, TimeoutError di-raise dan task dibatalkan secara otomatis.

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "Selesai"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
        print(result)
    except asyncio.TimeoutError:
        print("Operasi timeout setelah 3 detik")

asyncio.run(main())

Output:

Operasi timeout setelah 3 detik

Selain timeout, kita juga perlu menangani error pada batch task. Secara default, asyncio.gather() akan me-raise exception dari task pertama yang gagal dan membatalkan task lainnya. Untuk menghindari perilaku ini, gunakan parameter return_exceptions=True agar error di satu task tidak menggagalkan seluruh batch.

import asyncio

async def risky_task(name: str, fail: bool):
    await asyncio.sleep(1)
    if fail:
        raise ValueError(f"Error di {name}")
    return f"Berhasil: {name}"

async def main():
    results = await asyncio.gather(
        risky_task("Task-A", fail=False),
        risky_task("Task-B", fail=True),
        risky_task("Task-C", fail=False),
        return_exceptions=True,
    )
    for result in results:
        if isinstance(result, Exception):
            print(f"Gagal: {result}")
        else:
            print(result)

asyncio.run(main())

Output:

Berhasil: Task-A
Gagal: Error di Task-B
Berhasil: Task-C

Dengan return_exceptions=True, task yang sukses tetap mengembalikan hasilnya, sementara error ditangkap sebagai objek Exception dalam list hasil. Pola ini memastikan partial results tetap tersedia meskipun sebagian task gagal.

Kapan Menggunakan asyncio?

asyncio paling efektif untuk operasi I/O-bound: HTTP request, query database, baca/tulis file, dan komunikasi jaringan. Untuk operasi CPU-bound seperti komputasi berat, multiprocessing lebih cocok karena GIL (Global Interpreter Lock) mencegah thread Python menjalankan bytecode secara paralel. Menggabungkan keduanya — asyncio untuk I/O dan multiprocessing untuk CPU — memberikan arsitektur yang optimal untuk aplikasi yang membutuhkan kedua jenis concurrency.

Integrasi dengan library async seperti aiohttp untuk HTTP atau aiosqlite untuk database memungkinkan stack I/O sepenuhnya non-blocking dari endpoint hingga data layer. Dengan pola yang sudah kita bahas — gather() untuk batch, create_task() untuk kontrol granular, wait_for() untuk timeout, dan return_exceptions untuk error handling — kita memiliki fondasi yang cukup untuk membangun aplikasi async yang robust di Python.

Ingin menguasai Python secara mendalam? Kunjungi kursus dan bootcamp di Rumah Coding untuk belajar langsung dari praktisi berpengalaman dan membangun portfolio yang siap digunakan di industri.