Celery, Redis Queue & Pekerjaan Latar Belakang — Belajar FastAPI

⚖️ BackgroundTasks vs Celery

	BackgroundTasks	Celery / RQ / ARQ
Tempat berjalan	Proses web yang sama	Proses pekerja terpisah
Selamat bila proses mati?	Tidak	Ya
Retry otomatis	Tidak	Ya
Penjadwalan berkala	Tidak	Ya
Penskalaan terpisah	Tidak	Ya
Pemantauan	Tidak ada	Ada
Kerumitan	Nol	Perlu broker & pekerja

💡 Panduan memilih

Pakai BackgroundTasks untuk pekerjaan singkat yang boleh gagal diam-diam: mencatat audit, membersihkan berkas sementara, mengirim event analitik. Pakai Celery begitu pekerjaan itu tidak boleh hilang — email verifikasi, pembuatan invoice, pemrosesan pembayaran, pembuatan laporan.

🚫 Jebakan BackgroundTasks

@app.post("/daftar")
async def daftar(data: PenggunaBaru, tugas: BackgroundTasks):
    pengguna = await simpan(data)
    tugas.add_task(kirim_email_verifikasi, pengguna.email)   # ⚠️
    return {"id": pengguna.id}

# Bila pod di-restart, di-scale down, atau crash tepat setelah
# respons terkirim, email itu TIDAK PERNAH dikirim -- dan tidak ada
# jejak apa pun bahwa ia gagal. Pengguna tidak bisa masuk.

⚙️ Menyiapkan Celery

celery_app.py

from celery import Celery

celery = Celery(
    "aplikasi",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery.conf.update(
    task_serializer="json",
    accept_content=["json"],
    timezone="Asia/Jakarta",

    # Baru di-ACK SETELAH selesai, bukan saat diterima.
    # Bila pekerja mati di tengah, task dikerjakan pekerja lain.
    task_acks_late=True,
    task_reject_on_worker_lost=True,

    # Ambil satu per satu -- penting untuk task yang durasinya beragam,
    # supaya tidak ada pekerja yang menimbun task lalu menganggur.
    worker_prefetch_multiplier=1,

    task_time_limit=300,        # dibunuh paksa
    task_soft_time_limit=270,   # exception dilempar dulu agar bisa bersih-bersih
    result_expires=3600,
)

⚠️ task_acks_late=True mengubah jaminan pengiriman

Dengan pengaturan bawaan, task di-ACK begitu diterima pekerja — bila pekerja mati sedetik kemudian, task itu hilang. Dengan acks_late, ACK baru dikirim setelah task selesai, sehingga task yang terputus akan dikerjakan ulang. Konsekuensinya: jaminan berubah menjadi at-least-once, dan task Anda wajib idempoten.

📝 Menulis Task

tasks.py

@celery.task(
    bind=True,
    autoretry_for=(httpx.HTTPError, ConnectionError),
    retry_backoff=True,        # 1s, 2s, 4s, 8s...
    retry_backoff_max=600,
    retry_jitter=True,         # sebar agar tidak serempak
    max_retries=5,
)
def kirim_email_verifikasi(self, pengguna_id: int):
    with sesi_db() as db:
        pengguna = db.get(Pengguna, pengguna_id)
        if pengguna is None:
            return "pengguna sudah dihapus"      # jangan retry
        if pengguna.email_terkirim_pada:
            return "sudah pernah dikirim"        # ← IDEMPOTEN

        layanan_email.kirim(pengguna.email, buat_token(pengguna))
        pengguna.email_terkirim_pada = datetime.now(timezone.utc)
        db.commit()
    return "terkirim"

🚫 Jangan kirim objek besar sebagai argumen

# ❌ Seluruh objek diserialisasi ke broker.
#    Datanya juga sudah basi saat task akhirnya dijalankan.
proses_pesanan.delay(objek_pesanan)

# ✅ Kirim ID saja, ambil data terbaru di dalam task.
proses_pesanan.delay(pesanan_id)

Argumen task disimpan di broker sebagai JSON. Objek besar membebani Redis, memperlambat antrean, dan yang terpenting: data yang dikirim adalah potret saat task dibuat, bukan saat ia dijalankan — yang bisa berselang beberapa menit.

🔗 Memanggil dari FastAPI

@app.post("/daftar", status_code=202)
async def daftar(data: PenggunaBaru, db: SesiDB):
    pengguna = await simpan(db, data)

    hasil = kirim_email_verifikasi.delay(pengguna.id)

    return {
        "id": pengguna.id,
        "task_id": hasil.id,       # klien bisa memantau statusnya
    }

@app.get("/task/{task_id}")
async def status_task(task_id: str):
    hasil = celery.AsyncResult(task_id)
    return {
        "status": hasil.status,     # PENDING, STARTED, SUCCESS, FAILURE, RETRY
        "hasil": hasil.result if hasil.ready() else None,
    }

⚠️ Jangan pernah memanggil .get() di dalam endpoint

hasil.get() memblokir sampai task selesai. Melakukannya di dalam endpoint menghapus seluruh manfaat pemrosesan asinkron — permintaan HTTP kembali menunggu, dan worker web Anda tertahan. Kembalikan 202 Accepted beserta task_id, lalu biarkan klien memeriksa statusnya.

⏰ Penjadwalan Berkala

from celery.schedules import crontab

celery.conf.beat_schedule = {
    "bersihkan-sesi": {
        "task": "tasks.bersihkan_sesi_kedaluwarsa",
        "schedule": crontab(hour=3, minute=0),
    },
    "laporan-harian": {
        "task": "tasks.kirim_laporan_harian",
        "schedule": crontab(hour=7, minute=0, day_of_week="1-5"),
    },
    "sinkron-stok": {
        "task": "tasks.sinkron_stok",
        "schedule": 300.0,        # tiap 5 menit
    },
}

🚫 Hanya boleh ada SATU proses beat

Celery Beat adalah penjadwal. Menjalankan dua proses beat berarti setiap tugas terjadwal dieksekusi dua kali — laporan terkirim ganda, tagihan dibuat dua kali. Di Kubernetes, jalankan beat sebagai Deployment dengan replicas: 1, atau gunakan penjadwal yang mendukung penguncian terdistribusi.

🧵 Alternatif yang Lebih Ringan: ARQ

# ARQ dibangun untuk asyncio, jauh lebih sederhana daripada Celery
from arq import create_pool
from arq.connections import RedisSettings

async def kirim_email(ctx, pengguna_id: int):
    ...

class WorkerSettings:
    functions = [kirim_email]
    redis_settings = RedisSettings(host="localhost")
    max_tries = 5

# Dari FastAPI:
await app.state.antrian.enqueue_job("kirim_email", pengguna_id)

💡 Kapan memilih ARQ

Celery berakar pada kode sinkron dan dukungan asyncio-nya terasa ditempelkan belakangan. Bila seluruh aplikasi Anda sudah async dan kebutuhannya sederhana, ARQ memberi integrasi asyncio yang jauh lebih alami dengan permukaan konfigurasi yang jauh lebih kecil. Celery tetap unggul untuk alur kerja rumit: chain, group, chord, dan routing antrean berjenjang.

📊 Yang Harus Dipantau

Panjang antrean. Naik terus berarti pekerja tidak cukup — beri alert pada ambang tertentu.
Lama menunggu di antrean. Selisih antara task dibuat dan mulai dikerjakan.
Tingkat kegagalan per jenis task. Lonjakan menandakan dependensi bermasalah.
Task yang mati mendadak — biasanya karena OOM atau melewati batas waktu.
Antrean surat mati. Task yang gagal melewati batas retry harus masuk ke sini untuk diperiksa manusia, bukan diulang selamanya.