Membuat Background Job di Django dengan Django-Q2: Antrean, Penjadwalan, dan Pemantauan

Halo, Rekan ArtonLabs. Blog yang sedang Anda baca ini menjadwalkan artikelnya dengan alat yang dibahas di bawah, jadi ini tutorial dari pengalaman langsung.

Ada pekerjaan yang tidak boleh dilakukan di dalam permintaan HTTP: mengirim email massal, menghasilkan laporan PDF, memanggil API pihak ketiga yang lambat, atau, seperti di blog ini, membuat artikel dengan model bahasa setiap hari. Pengguna tidak mau menunggu tiga puluh detik, dan Gunicorn akan memutus permintaan yang terlalu lama. Django-Q2 adalah antrean tugas yang ringan untuk Django: tanpa Redis, cukup database yang sudah ada. Blog yang Anda baca ini memakainya untuk penjadwalan posting harian.

Kenapa Django-Q2, Bukan Celery

Celery sangat mampu, tetapi membawa Redis atau RabbitMQ, konfigurasi yang panjang, dan satu lagi layanan yang harus dipantau. Untuk aplikasi berukuran kecil sampai menengah dengan beberapa ratus tugas per hari, Django-Q2 memakai tabel database sebagai antrean, jadwalnya adalah model Django biasa yang bisa disunting di admin, dan hasil tugas tersimpan untuk ditinjau. Django-Q2 adalah kelanjutan aktif dari Django-Q lama dengan dependensi yang diperbarui. Kalau suatu hari Anda butuh puluhan ribu tugas per menit, saat itu Celery layak dipertimbangkan.

Instalasi dan Konfigurasi

pip install django-q2

Tambahkan 'django_q' ke INSTALLED_APPS (nama aplikasinya tetap django_q), lalu konfigurasi di settings.py:

Q_CLUSTER = {
    'name': 'aplikasi',
    'workers': 2,
    'timeout': 300,
    'retry': 360,
    'queue_limit': 50,
    'bulk': 10,
    'orm': 'default',
    'catch_up': False,
}

Arti kunci-kuncinya: orm menunjuk koneksi database yang dipakai sebagai antrean; workers jumlah proses pekerja; timeout batas detik sebuah tugas sebelum dihentikan; retry berapa lama antrean menunggu sebelum menawarkan kembali tugas yang belum selesai, dan harus lebih besar dari timeout; catch_up menentukan apakah jadwal yang terlewat saat cluster mati dijalankan semua begitu hidup lagi, dan untuk tugas seperti "kirim ringkasan harian" Anda hampir pasti ingin False. Lalu python manage.py migrate untuk membuat tabelnya.

Menjalankan Tugas Asinkron

# laporan/tasks.py
def buat_laporan(bulan, email):
    pdf = susun_pdf(bulan)
    kirim_email(email, pdf)
    return f"laporan {bulan} terkirim ke {email}"

# di view
from django_q.tasks import async_task
async_task('laporan.tasks.buat_laporan', '2026-08', request.user.email,
           task_name='laporan-agustus', group='laporan')

Fungsi dirujuk dengan jalur string atau langsung sebagai objek fungsi. async_task kembali seketika dengan id tugas; pekerja yang mengeksekusinya. Argumen harus bisa di-pickle, jadi kirim id objek, bukan objek model utuh; ambil ulang dari database di dalam tugas. Hasil kembalian fungsi disimpan di tabel Task dan bisa dibaca dengan result(id) atau di admin.

Penjadwalan Berkala

from django_q.models import Schedule
Schedule.objects.create(
    func='blog.tasks.buat_artikel_harian',
    name='artikel-harian',
    schedule_type=Schedule.DAILY,
    repeats=-1,
)

Jadwal dihitung dalam zona waktu Django (TIME_ZONE), jadi pastikan pengaturannya Asia/Jakarta kalau Anda ingin "harian" berarti tengah malam WIB, bukan UTC. Untuk pola yang lebih rumit, schedule_type=Schedule.CRON dengan cron="30 2 * * 1-5" menjalankan tugas pukul 02:30 pada hari kerja saja; pustaka croniter perlu dipasang untuk jenis ini. Bidang next_run bisa diisi untuk menunda jalan pertama.

Jenis jadwal yang tersedia antara lain sekali jalan, per menit dengan interval, per jam, harian, mingguan, bulanan, dan cron. repeats=-1 berarti tanpa batas. Karena jadwal adalah model biasa, ia muncul di admin Django dan bisa dihentikan atau diubah tanpa deploy ulang; itulah yang kami pakai untuk mengatur ritme posting blog ini.

Menjalankan Cluster sebagai Layanan

Tugas baru diproses hanya kalau cluster berjalan: python manage.py qcluster. Di produksi, jadikan layanan systemd terpisah dari Gunicorn:

[Unit]
Description=Django-Q cluster
After=network.target

[Service]
User=webapp
WorkingDirectory=/home/webapp/app
EnvironmentFile=/home/webapp/app/.env
ExecStart=/home/webapp/app/.venv/bin/python manage.py qcluster
Restart=always

[Install]
WantedBy=multi-user.target

Kami pernah menjalankan cluster ini dari terminal yang lupa ditutup; begitu sesi SSH putus, posting harian berhenti diam-diam selama sehari. Sejak itu ia selalu layanan systemd dengan Restart=always, dan pemantauan uptime memeriksa apakah tugas terakhir benar-benar berjalan.

Pemantauan dan Penanganan Gagal

Admin Django menampilkan tugas yang berhasil, gagal, dan jadwal beserta waktu jalan berikutnya. Untuk pemantauan otomatis, buat tugas kecil yang berjalan setiap jam dan menulis stempel waktu ke database atau berkas; pemantau luar lalu memeriksa stempel itu tidak lebih tua dari dua jam. Tugas yang gagal menyimpan traceback-nya; baca di admin, perbaiki, dan jalankan ulang. Untuk kegagalan sementara seperti API pihak ketiga yang sedang sibuk, bungkus pemanggilannya dengan percobaan ulang di dalam fungsi, karena Django-Q2 sendiri tidak mencoba ulang tugas yang gagal secara otomatis.

Jebakan yang Pernah Kami Alami

Pertama, retry lebih kecil dari timeout membuat tugas panjang dijalankan dua kali bersamaan; kami pernah mendapat dua artikel kembar karena itu, lalu menambahkan pemeriksaan duplikat di fungsinya. Kedua, mengubah kode tugas tidak berpengaruh sampai qcluster di-restart, sama seperti Gunicorn. Ketiga, tugas yang memuat pustaka berat (pandas, torch) membuat tiap pekerja memakan memori; batasi workers sesuai RAM, bukan sesuai jumlah inti. Keempat, sync: True di konfigurasi membuat tugas berjalan langsung di dalam permintaan, berguna untuk pengujian, dan bencana kalau lupa dimatikan di produksi.

Mengambil Hasil dan Mengelompokkan Tugas

from django_q.tasks import async_task, result, fetch, result_group

tid = async_task('laporan.tasks.buat_laporan', '2026-08', email)
r = result(tid, wait=0)        # None kalau belum selesai
t = fetch(tid)                 # objek Task lengkap: success, result, started, stopped

for bulan in ("2026-06", "2026-07", "2026-08"):
    async_task('laporan.tasks.buat_laporan', bulan, email, group='triwulan')
semua = result_group('triwulan', count=3)   # tunggu sampai ketiganya selesai

Argumen group berguna untuk pekerjaan yang dipecah menjadi beberapa tugas paralel; result_group mengumpulkan hasilnya. Jangan memakai wait yang lama di dalam view, karena itu mengembalikan masalah awal: permintaan HTTP yang menunggu. Pola yang lebih baik adalah view mengembalikan id tugas, dan halaman memeriksa statusnya secara berkala lewat endpoint kecil.

Hook Setelah Tugas Selesai

async_task(..., hook='laporan.tasks.setelah_selesai') memanggil fungsi lain dengan objek Task sebagai argumen begitu tugas selesai, berhasil maupun gagal. Kami memakainya untuk mengirim notifikasi dan mencatat kegagalan ke tabel sendiri, sehingga tabel bawaan Django-Q2 bisa dibersihkan secara berkala tanpa kehilangan riwayat yang penting. Tabel hasil memang tumbuh; atur save_limit di Q_CLUSTER (misalnya 500) agar hanya hasil terbaru yang disimpan.

, penjadwal, dan riwayat hasil dengan biaya satu paket dan satu layanan systemd. Mulailah dengan dua pekerja dan catch_up: False, pantau tugas pertamanya di admin, dan Anda akan jarang memikirkannya lagi.

Sampai jumpa, Rekan ArtonLabs. Kalau antrean tugas Anda punya cerita sendiri, yang berhasil maupun yang macet diam-diam, kami ingin mendengarnya.

Tags: # Scheduler # Django # Task Queue # Django-Q2 # Background Job