Belajar Golang dari Nol #19: Worker dan Job Terjadwal di Go

Di Belajar Golang dari Nol #18 kita sudah merapikan logging dan error handling di API produk. Sekarang API kita sudah bisa cerita sendiri lewat log ketika ada yang salah. Kalau kamu baru bergabung di tengah jalan, cek dulu daftar lengkap seri Belajar Golang supaya urutannya tidak lompat.

Kali ini kita bahas dua pola yang hampir pasti kamu butuhkan begitu API dipakai orang sungguhan: worker dan job terjadwal.

Dua jenis pekerjaan yang tidak cocok di handler

Coba perhatikan dua situasi ini.

Pertama, ada pekerjaan yang tidak boleh membuat request menunggu. Contohnya kirim email notifikasi atau generate laporan PDF. Kirim email bisa makan dua sampai lima detik. Kalau dikerjakan di dalam handler, user harus menatap loading selama itu. Padahal user tidak peduli emailnya sudah terkirim atau belum. Dia cuma mau tahu transaksinya berhasil.

Kedua, ada pekerjaan yang harus jalan rutin tanpa disuruh siapa pun. Rekap penjualan harian, bersih-bersih data lama, cek stok yang menipis. Tidak ada request yang memicu pekerjaan ini. Dia harus jalan sendiri sesuai jadwal.

Untuk masalah pertama kita pakai worker. Untuk masalah kedua kita pakai job terjadwal. Dua-duanya bisa dibangun dengan bahan yang sudah kamu pelajari di seri ini: goroutine, channel, dan context.

Worker pool: channel sebagai antrian

Di bagian 10 kamu sudah kenal goroutine dan channel. Waktu itu contohnya masih abstrak. Sekarang kita pakai untuk kasus nyata.

Idenya sederhana. Channel jadi antrian pekerjaan. Beberapa goroutine jadi pekerja yang mengambil dari antrian itu. Handler cukup memasukkan pekerjaan ke channel, lalu lanjut membalas request.

Pertama kita definisikan bentuk pekerjaannya sebagai struct.

type EmailJob struct {
	Tujuan string
	Isi    string
}

Lalu kita buat channel antrian dan beberapa worker.

jobs := make(chan EmailJob, 100)

var wg sync.WaitGroup
for i := 1; i <= 3; i++ {
	wg.Add(1)
	go func(id int) {
		defer wg.Done()
		for job := range jobs {
			slog.Info("worker ambil job", "worker", id, "tujuan", job.Tujuan)
			kirimEmail(job)
		}
	}(i)
}

Perhatikan beberapa hal. Channel diberi buffer 100 supaya handler tidak ikut menunggu saat semua worker sedang sibuk. Tiga goroutine worker melakukan range di channel yang sama. Go otomatis membagi job ke worker yang sedang kosong. Kamu tidak perlu mengatur pembagiannya sendiri.

Bagian for job := range jobs juga yang membuat shutdown jadi rapi. Loop ini baru berhenti ketika channel ditutup dan isinya habis. Jadi urutan mematikannya jelas: panggil close(jobs), worker menghabiskan sisa antrian, lalu wg.Wait() menunggu semuanya selesai. Tidak ada job yang dibuang di tengah jalan.

close(jobs)
wg.Wait()
slog.Info("semua worker selesai")

Contoh nyata: notifikasi setelah transaksi

Sekarang kita sambungkan ke API. Misalkan setiap ada produk baru dibuat, admin harus dapat email. Tanpa worker, handlernya kira-kira begini.

mux.HandleFunc("POST /produk", func(w http.ResponseWriter, r *http.Request) {
	// ... simpan produk ...
	kirimEmail(EmailJob{Tujuan: "admin@toko.com", Isi: "Produk baru masuk"})
	w.WriteHeader(http.StatusCreated)
})

Kalau kirimEmail makan dua detik, request juga makan dua detik. Coba cek dengan curl -w "%{time_total}", hasilnya sekitar 2,01 detik.

Dengan worker, handler cukup melempar job ke channel lalu langsung membalas 201.

mux.HandleFunc("POST /produk", func(w http.ResponseWriter, r *http.Request) {
	// ... simpan produk ...
	jobs <- EmailJob{Tujuan: "admin@toko.com", Isi: "Produk baru masuk"}
	w.WriteHeader(http.StatusCreated)
})

Sekarang curl yang sama selesai dalam hitungan milidetik. Email tetap terkirim, tapi di belakang layar. User tidak perlu ikut menunggu pekerjaan yang bukan urusannya.

Job terjadwal dengan time.Ticker

Pola kedua: pekerjaan yang jalan rutin. Standard library sudah menyediakan time.Ticker, yaitu channel yang berdenyut setiap interval tertentu. Gabungkan dengan goroutine dan select, jadilah scheduler sederhana.

Ingat pelajaran context di bagian 17. Setiap goroutine yang hidup lama wajib punya cara berhenti. Di sinilah ctx.Done() masuk.

func jalankanRekap(ctx context.Context, interval time.Duration) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			slog.Info("job rekap berhenti")
			return
		case <-ticker.C:
			mu.Lock()
			jumlah := len(produk)
			mu.Unlock()
			slog.Info("rekap produk", "jumlah", jumlah)
		}
	}
}

Jalankan sebagai goroutine: go jalankanRekap(ctx, time.Minute). Setiap menit dia menghitung jumlah produk dan mencatatnya lewat slog. Ketika context dibatalkan, select memilih cabang ctx.Done() dan goroutine keluar dengan rapi. Jangan lupa defer ticker.Stop() supaya resource ticker dilepas.

Jalan pada jam tertentu

Ticker cocok untuk interval. Tapi bagaimana kalau job harus jalan tiap tengah malam? Polanya sedikit berbeda: hitung durasi sampai jadwal berikutnya dengan time.Until, tidur selama itu, kerjakan, lalu ulangi.

func jalankanTengahMalam(ctx context.Context, tugas func()) {
	for {
		sekarang := time.Now()
		besok := time.Date(sekarang.Year(), sekarang.Month(), sekarang.Day()+1,
			0, 0, 0, 0, sekarang.Location())
		timer := time.NewTimer(time.Until(besok))

		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-timer.C:
			tugas()
		}
	}
}

time.Date dengan hari ditambah satu dan jam nol menghasilkan tengah malam berikutnya. time.Until menghitung sisa waktunya. Setelah tugas selesai, loop mengulang dan menghitung tengah malam berikutnya lagi.

Pola sederhana ini sering sudah cukup. Banyak orang buru-buru pasang library cron padahal jadwalnya cuma satu atau dua. Kalau kebutuhanmu masih bisa dijelaskan dalam satu kalimat, misalnya tiap jam atau tiap tengah malam, standard library sudah memadai. Lebih sedikit dependency, lebih mudah dipahami orang yang membaca kodenya nanti.

Kapan layak pakai robfig/cron

Kalau jadwalnya mulai ramai dan rumit, misalnya tiap hari kerja jam 8 pagi, tiap tanggal 1, plus tiap 15 menit di jam kantor, barulah library seperti robfig/cron layak dipakai. Dia menerima ekspresi cron yang sama dengan crontab di Linux.

c := cron.New()
c.AddFunc("0 0 * * *", rekapHarian)     // tiap tengah malam
c.AddFunc("*/15 8-17 * * 1-5", cekStok) // tiap 15 menit, jam kerja
c.Start()

Install dengan go get github.com/robfig/cron/v3. Aturannya sama seperti dependency lain di seri ini: pakai ketika kebutuhannya nyata, bukan karena terlihat keren.

Jebakan yang harus kamu tahu dari awal

Sebelum kamu pakai pola ini di production, ada tiga jebakan yang jujur harus dibahas.

Pertama, job hilang saat server restart. Antrian kita hidup di memori. Kalau server mati atau di-deploy ulang, semua job yang belum dikerjakan ikut lenyap. Untuk email notifikasi biasa, ini sering masih bisa diterima. Untuk job yang tidak boleh hilang, misalnya pemrosesan pembayaran, kamu butuh durable queue seperti Redis atau RabbitMQ. Job disimpan di luar proses aplikasi, jadi selamat dari restart. Kita tidak bedah itu sekarang, cukup tahu dulu batasnya di mana.

Kedua, dua instance server berarti job ganda. Kalau nanti aplikasimu jalan di dua server demi ketersediaan, job terjadwal ikut jalan di dua-duanya. Rekap harian terkirim dua kali. Solusinya macam-macam, dari lock di database sampai memisahkan scheduler jadi service sendiri. Yang penting sekarang: sadari bahwa pola ini mengasumsikan satu instance.

Ketiga, panic di worker mematikan goroutine diam-diam. Ini yang paling licik. Kalau kode di dalam worker panic, goroutine itu mati. Kamu punya tiga worker, satu panic, sisa dua. Panic lagi, sisa satu. Lama-lama antrian menumpuk dan tidak ada yang tahu kenapa. Ingat pelajaran bagian 18: panic yang tidak ditangani harus dicatat. Pasang recover di fungsi kerjanya.

func kirimEmail(job EmailJob) {
	defer func() {
		if r := recover(); r != nil {
			slog.Error("worker panic", "tujuan", job.Tujuan, "error", r)
		}
	}()

	time.Sleep(2 * time.Second) // simulasi kirim email
	slog.Info("email terkirim", "tujuan", job.Tujuan)
}

Dengan begini, job yang bermasalah tercatat di log dan workernya tetap hidup untuk job berikutnya.

Satu catatan lagi di level server. Untuk job yang benar-benar terpisah dari API, misalnya script backup, kamu tidak harus menjadwalkannya dari dalam Go. systemd yang kita pakai di bagian 15 untuk menjalankan aplikasi juga punya fitur timer. Kamu tulis unit timer, systemd yang menjalankan program Go kecil sesuai jadwal. Kelebihannya, jadwal terlihat oleh admin server lewat systemctl list-timers, bukan tersembunyi di dalam kode.

Latihan: API produk dengan worker dan rekap terjadwal

Sekarang kita rangkai semuanya jadi satu program utuh. API produk menerima POST, melempar job email ke worker, dan job rekap jalan tiap 30 detik. Shutdown ditangani lewat signal.NotifyContext, cara standar menangkap Ctrl+C atau sinyal stop dari systemd.

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

type Produk struct {
	ID    int    `json:"id"`
	Nama  string `json:"nama"`
	Harga int    `json:"harga"`
}

type EmailJob struct {
	Tujuan string
	Isi    string
}

var (
	mu     sync.Mutex
	produk = []Produk{{ID: 1, Nama: "Kopi Arabika", Harga: 85000}}
)

func kirimEmail(job EmailJob) {
	defer func() {
		if r := recover(); r != nil {
			slog.Error("worker panic", "tujuan", job.Tujuan, "error", r)
		}
	}()
	time.Sleep(2 * time.Second) // simulasi kirim email
	slog.Info("email terkirim", "tujuan", job.Tujuan, "isi", job.Isi)
}

func jalankanRekap(ctx context.Context, interval time.Duration) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			slog.Info("job rekap berhenti")
			return
		case <-ticker.C:
			mu.Lock()
			jumlah := len(produk)
			mu.Unlock()
			slog.Info("rekap produk", "jumlah", jumlah)
		}
	}
}

func main() {
	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil)))

	ctx, stop := signal.NotifyContext(context.Background(),
		os.Interrupt, syscall.SIGTERM)
	defer stop()

	jobs := make(chan EmailJob, 100)
	var wg sync.WaitGroup
	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for job := range jobs {
				slog.Info("worker ambil job", "worker", id, "tujuan", job.Tujuan)
				kirimEmail(job)
			}
		}(i)
	}

	go jalankanRekap(ctx, 30*time.Second)

	mux := http.NewServeMux()
	mux.HandleFunc("POST /produk", func(w http.ResponseWriter, r *http.Request) {
		var p Produk
		if err := json.NewDecoder(r.Body).Decode(&p); err != nil {
			http.Error(w, `{"error":"body tidak valid"}`, http.StatusBadRequest)
			return
		}
		mu.Lock()
		p.ID = len(produk) + 1
		produk = append(produk, p)
		mu.Unlock()

		jobs <- EmailJob{
			Tujuan: "admin@toko.com",
			Isi:    fmt.Sprintf("Produk baru: %s", p.Nama),
		}

		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusCreated)
		json.NewEncoder(w).Encode(p)
	})

	server := &http.Server{Addr: ":8080", Handler: mux}
	go func() {
		slog.Info("server jalan", "addr", ":8080")
		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			slog.Error("server gagal", "error", err)
			os.Exit(1)
		}
	}()

	<-ctx.Done()
	slog.Info("sinyal berhenti diterima")

	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	server.Shutdown(shutdownCtx)

	close(jobs)
	wg.Wait()
	slog.Info("semua worker selesai, aplikasi berhenti")
}

Jalankan dengan go run main.go, lalu kirim beberapa request.

curl -X POST localhost:8080/produk \
  -d '{"nama":"Teh Hijau","harga":45000}'

Perhatikan lognya. Respons 201 datang seketika, log worker ambil job muncul, dua detik kemudian email terkirim. Tiap 30 detik muncul rekap produk dengan jumlah terbaru. Dua pola jalan berdampingan tanpa saling ganggu.

Sekarang tekan Ctrl+C. Urutan lognya bercerita sendiri: sinyal diterima, server berhenti menerima request, job rekap berhenti, worker menghabiskan sisa antrian, baru aplikasi benar-benar mati. Itulah shutdown yang rapi. Tidak ada email yang setengah terkirim, tidak ada goroutine yang dipaksa mati.

Selanjutnya

API kita sekarang bisa mengerjakan tugas di belakang layar dan punya jadwal sendiri. Di bagian 20 kita bahas kebutuhan yang hampir pasti muncul di aplikasi nyata: Upload dan Menyimpan File di API Go.

Kalau kamu sedang membangun sistem dengan kebutuhan background job dan penjadwalan seperti ini untuk bisnismu, tim kami di Arrazy juga mengerjakan pengembangan sistem aplikasi custom dari perancangan sampai deploy.