Async Processing #

Dalam dunia software modern — mulai dari web application, mobile backend, hingga distributed system — async processing bukan lagi sekadar optimasi performa, melainkan kebutuhan arsitektural. Kesalahan terbesar yang dilakukan banyak tim adalah memperlakukan async sebagai sesuatu yang ditambahkan belakangan saat sistem mulai lambat. Padahal sistem besar tidak menjadi async — mereka dirancang async dari awal. Artikel ini membahas async processing dari akarnya: mengapa model synchronous punya batas yang tidak bisa ditembus dengan hardware lebih besar, tiga bentuk async yang punya karakteristik berbeda, empat prinsip dasar yang harus dipahami, serta bagaimana memutuskan dengan tepat kapan async adalah solusi dan kapan ia justru menjadi sumber masalah baru.

Apa Itu Async Processing? #

Async processing (asynchronous processing) adalah pola pemrosesan di mana suatu task tidak dieksekusi secara blocking terhadap alur utama eksekusi program. Request tidak harus menunggu proses selesai sebelum melanjutkan eksekusi berikutnya — hasilnya bisa dikembalikan di kemudian waktu, oleh worker lain, di proses lain, atau bahkan di mesin lain.

Perbedaan mendasar antara synchronous dan asynchronous:

flowchart TD
    subgraph Sync["Synchronous — Blocking"]
        S1[Request Masuk] --> S2["Proses A\nvalidasi"]
        S2 --> S3["Proses B\nsimpan ke DB"]
        S3 --> S4["Proses C\nkirim email\n⏳ tunggu SMTP"]
        S4 --> S5["Proses D\ngenerate PDF\n⏳ tunggu render"]
        S5 --> S6["Response ke User\n~3000ms"]
    end
    subgraph Async["Asynchronous — Non-Blocking"]
        A1[Request Masuk] --> A2["Proses A\nvalidasi"]
        A2 --> A3["Proses B\nsimpan ke DB"]
        A3 --> A4["Enqueue: kirim email\nEnqueue: generate PDF"]
        A4 --> A5["Response ke User\n~80ms"]
        A4 -.->|background| A6[Worker: kirim email]
        A4 -.->|background| A7[Worker: generate PDF]
    end

Dengan async processing, user mendapat response dalam milidetik, sementara proses berat diselesaikan di background tanpa menghalangi request lain.

Mengapa Sistem Synchronous Punya Batas? #

Async processing lahir dari keterbatasan nyata sistem synchronous yang tidak bisa diselesaikan hanya dengan menambah CPU atau memory.

Thread Blocking adalah Akar Masalahnya #

Dalam model synchronous tradisional, satu request mengunci satu thread selama proses berlangsung. Thread tersebut idle saat menunggu I/O — database query, HTTP call ke API eksternal, baca file dari disk — tapi tetap mengkonsumsi memory dan slot di thread pool.

sequenceDiagram
    participant User
    participant Thread as Thread Pool\n(20 thread)
    participant DB
    participant SMTP

    User->>Thread: Request registrasi (Thread 1 dialokasi)
    Thread->>DB: INSERT user (50ms tunggu)
    DB-->>Thread: OK
    Thread->>SMTP: Kirim email (800ms tunggu!)
    Note over Thread: Thread 1 idle 800ms\nTidak bisa layani request lain
    SMTP-->>Thread: OK
    Thread-->>User: Response (850ms total)
    Note over Thread: Dengan 20 thread dan 800ms/request\nhanya bisa handle ~25 request/detik

Di traffic tinggi, thread pool habis, request baru masuk antrian, latency naik berlipat, dan akhirnya sistem collapse — bukan karena kurang CPU, tapi karena thread idle menunggu I/O yang lambat.

Dampak Konkret di Sistem Produksi #

Empat Prinsip Dasar Async Processing #

1. Decoupling #

Pemanggil tidak perlu tahu bagaimana dan kapan task selesai. Ini memisahkan tanggung jawab antara yang meminta pekerjaan dan yang mengerjakan.

// ANTI-PATTERN: tight coupling — pemanggil harus tunggu sampai email terkirim
func registerUser(req RegistrationRequest) error {
    user := createUser(req)
    db.Save(user)
    
    // Request harus tunggu email selesai — coupling yang tidak perlu
    err := emailService.SendWelcomeEmail(user.Email)
    if err != nil {
        // Apakah kita rollback user hanya karena email gagal?
        return err
    }
    return nil
}

// BENAR: decoupled — registrasi dan email adalah concern terpisah
func registerUser(req RegistrationRequest) error {
    user := createUser(req)
    db.Save(user)
    
    // Publish event — tidak peduli kapan dan bagaimana email dikirim
    eventBus.Publish(UserRegisteredEvent{
        UserID: user.ID,
        Email:  user.Email,
    })
    return nil // response langsung, tanpa tunggu email
}

// Email dikirim oleh handler yang terpisah, bisa retry, bisa scale independen
func (h *EmailHandler) HandleUserRegistered(event UserRegisteredEvent) {
    emailService.SendWelcomeEmail(event.Email)
}

2. Non-Blocking Execution #

Thread utama bebas melayani request lain sementara proses berjalan di background. Ini adalah kunci skalabilitas horizontal.

flowchart LR
    subgraph Blocking["Blocking"]
        T1["Thread 1\nRequest A"] --> W1["⏳ Tunggu DB\n200ms"]
        T2["Thread 2\nRequest B"] --> W2["⏳ Tunggu DB\n200ms"]
        T3["Thread 3\nRequest C"] --> W3["⏳ Tunggu DB\n200ms"]
        T4["Thread 4\nIdle — tidak ada thread lagi!"]
    end
    subgraph NonBlocking["Non-Blocking"]
        EL["Event Loop\nSatu Thread"] --> R1["Request A\ndaftar callback"]
        EL --> R2["Request B\ndaftar callback"]
        EL --> R3["Request C\ndaftar callback"]
        EL --> R4[Request D]
        EL --> R5[Request E...]
        DB["(DB)"] -.->|result A tiba| EL
        DB -.->|result B tiba| EL
    end

3. Eventual Completion #

Hasil tidak tersedia instan — ia akan selesai “pada suatu waktu” di masa depan. Ini membutuhkan perubahan cara berpikir tentang konsistensi data: bukan strong consistency (selalu up-to-date), tapi eventual consistency (akan konsisten pada akhirnya).

// ANTI-PATTERN: mengasumsikan task async sudah selesai saat response dikirim
func uploadFile(file File) UploadResponse {
    jobID := queue.Enqueue(ResizeImageJob{File: file})
    
    return UploadResponse{
        JobID:    jobID,
        ThumbURL: "/thumbs/" + file.ID + ".jpg", // belum tentu ada!
        Status:   "completed", // SALAH — ini masih pending
    }
}

// BENAR: ekspresikan state yang jujur, sediakan cara polling atau webhook
func uploadFile(file File) UploadResponse {
    jobID := queue.Enqueue(ResizeImageJob{File: file})
    
    return UploadResponse{
        JobID:      jobID,
        Status:     "processing",             // jujur tentang state saat ini
        StatusURL:  "/jobs/" + jobID,         // client bisa poll status
        WebhookURL: "/jobs/" + jobID + "/webhook", // atau tunggu notifikasi
    }
}

4. State Awareness #

Karena proses berjalan terpisah dari request, state harus disimpan secara eksplisit. Jika worker crash, state tidak boleh ikut hilang.

// ANTI-PATTERN: state hanya di memory — hilang saat crash
var pendingJobs = make(map[string]Job) // tidak persistent!

func enqueueJob(job Job) string {
    id := uuid.New().String()
    pendingJobs[id] = job // crash = semua job hilang
    go processJob(job)
    return id
}

// BENAR: state disimpan di persistent storage sebelum diproses
func enqueueJob(job Job) (string, error) {
    // Simpan ke DB dulu — sebelum queue, sebelum proses
    id := uuid.New().String()
    if err := db.Create(&JobRecord{
        ID:        id,
        Payload:   mustMarshal(job),
        Status:    "pending",
        CreatedAt: time.Now(),
    }); err != nil {
        return "", err
    }
    
    // Baru publish ke queue — jika queue gagal, job masih ada di DB
    queue.Publish(JobMessage{ID: id})
    return id, nil
}

Tiga Bentuk Async Processing #

Async processing tidak hadir dalam satu bentuk saja. Setiap bentuk punya karakteristik, kekuatan, dan batasan yang berbeda.

Bentuk 1: Async di Level Code #

Task async dijalankan dalam proses yang sama menggunakan goroutine, coroutine, async/await, atau future/promise. Ini adalah bentuk async yang paling ringan — cocok untuk I/O-bound task yang cepat selesai.

// ANTI-PATTERN: sequential — total latency = A + B + C
func getDashboardData(userID string) DashboardData {
    profile  := fetchProfile(userID)   // 100ms
    orders   := fetchOrders(userID)    // 150ms
    notifs   := fetchNotifs(userID)    // 80ms
    // Total: 330ms — padahal bisa 150ms
    return buildDashboard(profile, orders, notifs)
}

// BENAR: concurrent goroutine — total latency = max(A, B, C)
func getDashboardData(ctx context.Context, userID string) DashboardData {
    type result struct {
        profile Profile
        orders  []Order
        notifs  []Notification
        err     error
    }

    profileCh := make(chan result, 1)
    ordersCh  := make(chan result, 1)
    notifsCh  := make(chan result, 1)

    go func() {
        p, err := fetchProfile(ctx, userID)
        profileCh <- result{profile: p, err: err}
    }()
    go func() {
        o, err := fetchOrders(ctx, userID)
        ordersCh <- result{orders: o, err: err}
    }()
    go func() {
        n, err := fetchNotifs(ctx, userID)
        notifsCh <- result{notifs: n, err: err}
    }()

    // Kumpulkan semua hasil — total ~150ms bukan 330ms
    pr := <-profileCh
    or := <-ordersCh
    nr := <-notifsCh

    return buildDashboard(pr.profile, or.orders, nr.notifs)
}

Karakteristik:

Bentuk 2: Async Berbasis Background Worker #

Task dikirim ke antrian (queue), diproses oleh worker yang berjalan terpisah — bisa proses berbeda, server berbeda, atau bahkan cloud function. Ini adalah bentuk async yang paling umum untuk operasi bisnis berat.

flowchart LR
    A[API Handler] --> B["Validasi\nRequest"]
    B --> C["Simpan ke DB\nstatus: pending"]
    C --> D[Publish ke Queue]
    D --> E["Response 202\nAccepted"]
    D -.->|async| F["Worker 1\nproses job"]
    D -.->|async| G["Worker 2\nproses job"]
    D -.->|async| H["Worker 3\nproses job"]
    F --> I["Update DB\nstatus: completed"]
    G --> I
    H --> I
    I -.-> J["Webhook / Notif\nke client"]

// ANTI-PATTERN: kirim laporan di dalam request — user tunggu menit-menit
func generateReport(w http.ResponseWriter, r *http.Request) {
    report := heavyReportGeneration(r.Context(), getParams(r)) // bisa 2 menit!
    w.Write(report.Bytes()) // HTTP timeout sebelum selesai
}

// BENAR: enqueue job, return job ID, client poll atau tunggu webhook
func generateReport(w http.ResponseWriter, r *http.Request) {
    params := getParams(r)
    
    jobID, err := jobQueue.Enqueue(ReportJob{
        UserID: currentUser(r).ID,
        Params: params,
    })
    if err != nil {
        http.Error(w, "failed to queue job", 500)
        return
    }
    
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusAccepted) // 202, bukan 200
    json.NewEncoder(w).Encode(map[string]string{
        "job_id":     jobID,
        "status":     "processing",
        "status_url": "/reports/jobs/" + jobID,
    })
}

// Worker terpisah — bisa scale independen dari API
func (w *ReportWorker) Process(job ReportJob) error {
    report, err := heavyReportGeneration(context.Background(), job.Params)
    if err != nil {
        return err // queue akan retry
    }
    
    storage.Save(job.ID, report)
    db.UpdateJobStatus(job.ID, "completed")
    notifyUser(job.UserID, job.ID)
    return nil
}

Karakteristik:

Bentuk 3: Event-Driven Async Processing #

Sistem mempublikasikan event ketika sesuatu terjadi, dan siapapun yang tertarik bisa bereaksi — tanpa publisher harus tahu siapa konsumennya. Ini adalah bentuk async yang paling loosely coupled.

flowchart LR
    P["Order Service\nPublish: OrderPaid"]
    P --> Q[("(Event Bus\nKafka / Pub/Sub)")]
    Q --> C1["Invoice Service\nbuat invoice"]
    Q --> C2["Notification Service\nkirim email + push"]
    Q --> C3["Loyalty Service\ntambah poin"]
    Q --> C4["Analytics Service\nlog ke warehouse"]

// ANTI-PATTERN: publisher harus tahu dan memanggil semua downstream langsung
func processPayment(orderID string) error {
    order := db.FindOrder(orderID)
    markOrderAsPaid(order)
    
    // Tight coupling — jika satu downstream gagal, rollback semua?
    invoiceService.CreateInvoice(order)    // coupling!
    emailService.SendReceipt(order)        // coupling!
    loyaltyService.AddPoints(order)        // coupling!
    analyticsService.LogPayment(order)    // coupling!
    return nil
}

// BENAR: publish satu event, consumer bereaksi secara independen
func processPayment(orderID string) error {
    order := db.FindOrder(orderID)
    markOrderAsPaid(order)
    
    // Satu event, banyak consumer — publisher tidak tahu siapa yang dengerin
    eventBus.Publish("order.paid", OrderPaidEvent{
        OrderID:   order.ID,
        UserID:    order.UserID,
        Amount:    order.TotalAmount,
        PaidAt:    time.Now(),
    })
    return nil
}

// Setiap service punya consumer sendiri, independen, bisa fail tanpa pengaruhi lain
func (s *InvoiceService) OnOrderPaid(event OrderPaidEvent) error {
    return s.createInvoice(event.OrderID)
}

func (s *NotificationService) OnOrderPaid(event OrderPaidEvent) error {
    return s.sendReceiptEmail(event.UserID, event.OrderID)
}

Karakteristik:

Async di Arsitektur Modern #

Async processing adalah fondasi dari hampir semua arsitektur sistem skala besar.

flowchart TD
    subgraph Microservices["Microservices"]
        MS1[Order Service] -->|event async| MS2[Payment Service]
        MS2 -->|event async| MS3[Notification Service]
        MS1 -->|event async| MS4[Inventory Service]
    end
    subgraph Serverless["Serverless"]
        TR["Trigger\nS3 / Queue / HTTP"] -->|event| FN["Function\nberjalan on-demand"]
    end
    subgraph HighTraffic["High-Traffic System"]
        CLI[Client] --> LB[Load Balancer]
        LB --> API[API Server]
        API --> Q[("(Queue\nBuffer)")]
        Q --> W1[Worker Pool]
        Q --> W2[Worker Pool]
    end

Dampak Positif dan Tantangan #

Dampak Positif #

flowchart LR
    ASYNC[Async Processing] --> SC["Scalability\nWorker scale independen"]
    ASYNC --> RS["Resilience\nFailure terisolasi, bisa retry"]
    ASYNC --> PF["Performance\nLatency user lebih rendah"]
    ASYNC --> UX["Better UX\nTidak ada UI freeze"]
    ASYNC --> DC["Decoupling\nService independen satu sama lain"]

Tantangan yang Harus Diantisipasi #

Kesalahan Umum Engineer #

// ✗ Kesalahan 1: Menganggap async = parallel
// Goroutine tanpa goroutine yang benar-benar jalan concurrent != parallel
go func() {
    result = heavyCompute() // masih CPU-bound, tidak lebih cepat
}()

// ✗ Kesalahan 2: Async untuk semua hal — termasuk yang tidak perlu
eventBus.Publish(UserNameUpdatedEvent{...})
// Validasi username tidak perlu async — hasilnya harus instan

// ✗ Kesalahan 3: Tidak menyimpan state sebelum enqueue
queue.Publish(job) // jika queue crash, job hilang
db.Save(job)       // terlambat — seharusnya DB dulu, queue kedua

// ✓ Urutan yang benar: DB dulu, queue kedua
db.Save(&JobRecord{ID: jobID, Status: "pending"})
queue.Publish(JobMessage{ID: jobID})

// ✗ Kesalahan 4: Consumer tidak idempotent — message bisa diterima 2x
func handleJob(job Job) {
    db.Create(job.Result) // duplikat jika message di-redeliver!
}
// ✓ Cek dulu sebelum proses
func handleJob(job Job) {
    if db.Exists(job.ID) { return nil }
    db.Create(job.Result)
}

// ✗ Kesalahan 5: Tidak ada visibility ke job status
queue.Publish(job) // user tidak tahu apakah berhasil atau tidak
// ✓ Simpan job status yang bisa di-query
db.Save(&Job{ID: jobID, Status: "pending"})
// expose GET /jobs/{id} untuk polling

Kapan Menggunakan Async Processing #

Gunakan Async Jika: #

Jangan Gunakan Async Jika: #

Async processing memindahkan kompleksitas dari dalam request ke di luar request. Jika sistem kamu belum punya monitoring job, retry mechanism, DLQ, dan distributed tracing yang baik — async hanya memindahkan masalah ke tempat yang lebih sulit dilihat.

Checklist Implementasi Async Processing #

DESAIN:
  □ Identifikasi operasi yang tidak perlu selesai sebelum response
  □ Tentukan bentuk async: code-level, background worker, atau event-driven
  □ Definisikan contract antara publisher dan consumer (event schema)
  □ Tentukan SLA completion time — kapan job dianggap terlalu lama?

IMPLEMENTASI:
  □ Simpan job state ke persistent storage sebelum enqueue ke queue
  □ Consumer idempotent — cek job_id / event_id sebelum proses
  □ Error handling: retry policy + DLQ untuk job yang gagal terus
  □ Expose endpoint status job (/jobs/{id}) untuk polling client

OBSERVABILITY:
  □ Correlation ID dipropagasi dari request awal ke seluruh async chain
  □ Structured logging di setiap tahap: enqueue, dequeue, complete, fail
  □ Metrics: job throughput, queue depth, consumer lag, error rate
  □ Alert jika queue depth terlalu tinggi atau consumer lag membesar

TESTING:
  □ Unit test consumer dengan event yang diinjeksi langsung
  □ Integration test dengan queue real (atau test container)
  □ Test skenario: job gagal → retry → DLQ
  □ Test skenario: message duplikat → idempotency skip
  □ Load test: validasi queue tidak bottleneck saat traffic spike

Ringkasan #

• Async Processing adalah keputusan arsitektur, bukan optimasi — sistem besar tidak menjadi async, mereka dirancang async dari awal.
• Empat prinsip dasar: decoupling (pemanggil tidak peduli kapan selesai), non-blocking (thread bebas ke request lain), eventual completion (hasil tersedia di masa depan), state awareness (state harus persist agar tidak hilang).
• Tiga bentuk async: code-level (goroutine/coroutine, dalam satu proses), background worker (queue + worker terpisah, long-running), event-driven (publish-subscribe, loosely coupled multi-consumer).
• Decoupling adalah manfaat terbesar — publisher tidak perlu tahu siapa yang mengkonsumsi event; setiap service bisa berkembang dan gagal secara independen.
• Idempotency adalah syarat mutlak — message broker menjamin at-least-once delivery; consumer harus siap menerima event yang sama lebih dari sekali.
• Simpan state ke DB sebelum enqueue ke queue — urutan ini krusial agar job tidak hilang jika queue atau worker crash.
• Eventual consistency harus dikomunikasikan ke user — jangan kembalikan status “completed” saat task masih “processing”; gunakan 202 Accepted dan sediakan endpoint polling.
• Observability wajib ada sebelum async — correlation ID, job status tracking, dan queue depth monitoring adalah prasyarat, bukan nice-to-have.
• Jangan async untuk semua hal — validasi kritis, transaksi atomic, dan CRUD sederhana lebih baik tetap synchronous.

← Sebelumnya: Reactive Programming   Berikutnya: Event-Driven →