《后端架构设计·分布式与数据一致性》11.3 延迟任务、时间轮与错过触发的补偿

// file: internal/platform/delayq/delayq.go
// tier: L2
// verify: go test ./internal/platform/delayq/ -count=1
package delayq

// Task is one delayed job. Due is an absolute tick; Seq breaks ties so both
// queue implementations fire same-deadline tasks in insertion order.
type Task struct {
	ID  string
	Due int64
	Seq uint64
}

// Heap is a binary min-heap ordered by Due then Seq. It is the baseline for
// arbitrary deadlines; push and pop are O(log n).
type Heap struct {
	items       []Task
	seq         uint64
	comparisons int
}

// Push schedules id at due.
func (h *Heap) Push(id string, due int64) {
	h.seq++
	h.items = append(h.items, Task{ID: id, Due: due, Seq: h.seq})
	h.up(len(h.items) - 1)
}

func (h *Heap) less(i, j int) bool {
	h.comparisons++
	if h.items[i].Due != h.items[j].Due {
		return h.items[i].Due < h.items[j].Due
	}
	return h.items[i].Seq < h.items[j].Seq
}

func (h *Heap) up(i int) {
	for i > 0 {
		parent := (i - 1) / 2
		if !h.less(i, parent) {
			return
		}
		h.items[i], h.items[parent] = h.items[parent], h.items[i]
		i = parent
	}
}

func (h *Heap) down(i int) {
	for {
		left := 2*i + 1
		if left >= len(h.items) {
			return
		}
		small := left
		if right := left + 1; right < len(h.items) && h.less(right, left) {
			small = right
		}
		if !h.less(small, i) {
			return
		}
		h.items[i], h.items[small] = h.items[small], h.items[i]
		i = small
	}
}

// PopDue removes every task with Due <= now and returns them in order.
func (h *Heap) PopDue(now int64) []Task {
	var out []Task
	for len(h.items) > 0 && h.items[0].Due <= now {
		out = append(out, h.items[0])
		last := len(h.items) - 1
		h.items[0] = h.items[last]
		h.items = h.items[:last]
		if len(h.items) > 0 {
			h.down(0)
		}
	}
	return out
}

// Len reports the number of queued tasks.
func (h *Heap) Len() int { return len(h.items) }

// Comparisons reports how many ordering comparisons Push and PopDue performed.
func (h *Heap) Comparisons() int { return h.comparisons }

// Wheel is a hashed timing wheel with one-tick resolution. Buckets hold tasks
// by absolute due tick; a bucket is re-scanned once per revolution, so a task
// due far in the future is inspected several times before it fires.
type Wheel struct {
	now     int64
	size    int
	buckets [][]Task
	seq     uint64
	scanned int
}

// NewWheel builds a wheel with size buckets.
func NewWheel(size int) *Wheel {
	if size < 1 {
		size = 1
	}
	return &Wheel{size: size, buckets: make([][]Task, size)}
}

// Schedule inserts id at due. Deadlines in the past are clamped to the next
// tick, because the current bucket has already been scanned.
func (w *Wheel) Schedule(id string, due int64) {
	if due <= w.now {
		due = w.now + 1
	}
	w.seq++
	slot := int(due % int64(w.size))
	w.buckets[slot] = append(w.buckets[slot], Task{ID: id, Due: due, Seq: w.seq})
}

// Advance moves the wheel one tick and returns the tasks that fired.
func (w *Wheel) Advance() []Task {
	w.now++
	slot := int(w.now % int64(w.size))
	bucket := w.buckets[slot]
	var fired, keep []Task
	for _, t := range bucket {
		w.scanned++
		if t.Due <= w.now {
			fired = append(fired, t)
			continue
		}
		keep = append(keep, t)
	}
	w.buckets[slot] = keep
	return fired
}

// Now reports the current tick.
func (w *Wheel) Now() int64 { return w.now }

// Scanned reports how many bucket entries Advance inspected.
func (w *Wheel) Scanned() int { return w.scanned }

// Misfire is the outcome of one catch-up decision.
type Misfire struct {
	Fires   int
	Skipped int
	Next    int64
}

// CatchUp decides what an interval task should do when the scheduler wakes at
// now and the next planned run was at next. At most maxFires runs are executed
// immediately; the remaining missed runs are skipped, and the schedule moves
// to the first slot strictly after now.
func CatchUp(next, now, interval int64, maxFires int) Misfire {
	if interval <= 0 || now < next {
		return Misfire{Next: next}
	}
	if maxFires < 1 {
		maxFires = 1
	}
	missed := (now-next)/interval + 1
	fires := missed
	if fires > int64(maxFires) {
		fires = int64(maxFires)
	}
	return Misfire{
		Fires:   int(fires),
		Skipped: int(missed - fires),
		Next:    next + missed*interval,
	}
}

=== RUN   TestHeapAndWheelAgree
    delayq_test.go:46: tasks=1000 heap_comparisons=16997 wheel_scanned=4416 ratio=0.26
--- PASS: TestHeapAndWheelAgree (0.00s)
=== RUN   TestWheelHandlesDelayBeyondOneRevolution
    delayq_test.go:64: due=200 size=16 scanned=13 revolutions
--- PASS: TestWheelHandlesDelayBeyondOneRevolution (0.00s)
=== RUN   TestCatchUp
    delayq_test.go:89: exactly due fires=1 skipped=0 next=110
    delayq_test.go:89: one late fires=1 skipped=0 next=110
    delayq_test.go:89: eleven missed capped at three fires=3 skipped=8 next=210
    delayq_test.go:89: cap zero means once fires=1 skipped=10 next=210
    delayq_test.go:89: large gap capped fires=100 skipped=99901 next=1000010
--- PASS: TestCatchUp (0.00s)