Background JobsBackground Jobs

job mỗi ngàyjobs a day

300k

job mỗi ngàyjobs a day

lúc cao điểmat peak

~50/s

lúc cao điểmat peak

instance worker, mỗi máy 8 luồngworker instances, 8 slots each

6

instance worker, mỗi máy 8 luồngworker instances, 8 slots each

cron chạy trùngdouble cron runs

0

cron chạy trùngdouble cron runs

Bảng job trong PostgreSQL vừa là hàng đợi vừa là nguồn sự thật. Worker nhận job bằng SKIP LOCKED và giữ lease; Lease Reaper trả lại job có lease hết hạn. Bật “Worker chết” hoặc “Quá timeout” để xem cách hệ thống tự hồi phục.A job table in PostgreSQL is both the queue and the source of truth. Workers claim jobs with SKIP LOCKED and hold a lease; the Lease Reaper returns jobs whose lease expired. Turn on “Worker crashes” or “Timeout” to see how the system recovers.

1. 1

   Hai worker nhận cùng một jobTwo workers claim the same job

   SELECT … FOR UPDATE SKIP LOCKED trong cùng câu UPDATE đổi trạng thái sang RUNNING: worker này bỏ qua dòng worker kia đang khoá, không chờ, không trùng.SELECT … FOR UPDATE SKIP LOCKED inside the UPDATE that sets RUNNING: each worker skips rows another one holds, with no waiting and no duplicates.

   Producer-Consumer
2. 2

   Worker chết, job kẹt ở RUNNINGA worker dies and the job stays RUNNING

   Mỗi job có lease locked_until. Worker gia hạn bằng heartbeat; reaper trả job có lease hết hạn về QUEUED. Lúc ghi kết quả phải kiểm locked_by để worker cũ không ghi đè.Each job has a lease, locked_until. The worker extends it with heartbeats; a reaper returns expired leases to QUEUED. Check locked_by when writing the result so a stale worker cannot overwrite it.

   Worker Pool
3. 3

   Cron chạy trùng trên nhiều instanceCron fires on every instance

   Mỗi lần chạy cron ghi một dòng với unique(job_name, scheduled_for); instance nào INSERT được thì chạy, instance khác nhận lỗi trùng và bỏ qua. Hoặc dùng leader election (advisory lock, ShedLock).Each cron firing inserts a row with unique(job_name, scheduled_for); the instance whose INSERT succeeds runs it, the others hit the unique violation and skip. Or use leader election (an advisory lock, ShedLock).

   Semaphore
4. 4

   Job treo chiếm luồngHung jobs hold slots

   Mỗi job chạy trong context có deadline; mọi I/O nhận context đó nên bị huỷ đúng lúc. Ở Java, Future.get(timeout) rồi cancel(true) và handler phải tôn trọng interrupt.Every job runs under a context with a deadline; all I/O takes that context and is cancelled on time. In Java, Future.get(timeout) then cancel(true), and the handler must honour interrupts.

   Cancellation / Context
5. 5

   Lỗi tạm thời và lỗi vĩnh viễnTransient versus permanent failures

   Lỗi tạm thời được hẹn lại với backoff có jitter; lỗi dữ liệu hoặc hết số lần thử chuyển sang DEAD để người xem, không thử mãi.Transient errors are rescheduled with jittered backoff; bad data or exhausted attempts go to DEAD for a person to review, never retried forever.

   Retry with Exponential Backoff
6. 6

   Chạy lại gây tác dụng phụ hai lầnRe-runs repeat side effects

   Job có thể chạy hơn một lần sau crash hoặc timeout. Handler dùng idempotency key với bên ngoài và kiểm tra trạng thái trước khi ghi.A job can run more than once after a crash or a timeout. Handlers pass an idempotency key to external systems and check state before writing.

   Command

Worker nhận job theo số luồng còn trống, chạy handler trong giới hạn thời gian, rồi ghi DONE, hẹn lại hoặc DEAD.The worker claims as many jobs as it has free slots, runs the handler within a time limit, then records DONE, a retry or DEAD.

@Component
@RequiredArgsConstructor
public class JobWorker {
    private static final String CLAIM = """
        UPDATE job SET status = 'RUNNING', locked_by = ?, attempts = attempts + 1,
                       locked_until = now() + interval '60 seconds'
        WHERE id IN (SELECT id FROM job
                     WHERE status = 'QUEUED' AND run_at <= now()
                     ORDER BY priority DESC, run_at
                     LIMIT ? FOR UPDATE SKIP LOCKED)1
        RETURNING *""";

    private final JdbcTemplate jdbc;
    private final JobRepository jobs;
    private final Map<String, JobHandler> handlers;
    private final ThreadPoolTaskExecutor slots;      // core = max = 8
    private final ExecutorService handlerPool;
    private final String workerId;

    @Scheduled(fixedDelay = 1000)
    public void poll() {
        int free = slots.getMaxPoolSize() - slots.getActiveCount();2
        if (free <= 0) return;
        List<Job> batch = jdbc.query(CLAIM, JOB_MAPPER, workerId, free);
        batch.forEach(job -> slots.execute(() -> run(job)));
    }

    void run(Job job) {
        Future<?> f = handlerPool.submit(() -> handlers.get(job.type()).handle(job));
        try {
            f.get(job.timeoutSeconds(), TimeUnit.SECONDS);
            jobs.markDone(job.id(), workerId);           // WHERE locked_by = ?3
        } catch (TimeoutException e) {
            f.cancel(true);                              // interrupt the handler4
            jobs.retryOrDead(job, workerId, e);
        } catch (Exception e) {
            jobs.retryOrDead(job, workerId, e);          // run_at = now() + backoff
        }
    }
}

func (w *Worker) Run(ctx context.Context, slots int) {
	var wg sync.WaitGroup
	for i := 0; i < slots; i++ {1
		wg.Add(1)
		go func() {
			defer wg.Done()
			for ctx.Err() == nil {
				if err := w.runOne(ctx); errors.Is(err, ErrNoJob) {
					time.Sleep(time.Second)
				}
			}
		}()
	}
	wg.Wait() // graceful shutdown: running jobs finish first
}

func (w *Worker) runOne(ctx context.Context) error {
	job, err := w.store.Claim(ctx, w.id, 60*time.Second) // FOR UPDATE SKIP LOCKED
	if err != nil {
		return err
	}
	jobCtx, cancel := context.WithTimeout(ctx, job.Timeout)2
	defer cancel()
	go w.heartbeat(jobCtx, cancel, job.ID)3

	err = w.handlers[job.Type].Handle(jobCtx, job)
	switch {
	case err == nil:
		return w.store.MarkDone(ctx, job.ID, w.id)
	case job.Attempts >= job.MaxAttempts:
		return w.store.MarkDead(ctx, job.ID, w.id, err)
	default:
		delay := backoff(job.Attempts) // 10s·2^n + jitter, capped at 1h
		return w.store.Reschedule(ctx, job.ID, w.id, time.Now().Add(delay), err)
	}
}

func (w *Worker) heartbeat(ctx context.Context, cancel context.CancelFunc, id int64) {
	t := time.NewTicker(20 * time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			if err := w.store.ExtendLease(ctx, id, w.id, 60*time.Second); err != nil {
				cancel() // lease lost: stop the handler too4
				return
			}
		}
	}
}

Mỗi pattern mở bài học tương ứng trong menu Pattern.Each one opens its lesson in the Patterns menu.