数据结构-堆与优先队列

无序数组:  插入 O(1)      取最小 O(n)
有序数组:  插入 O(n)      取最小 O(1)

有序数组:  1 3 4 7 8 9 12      ← 完全有序,信息过剩
堆:              1
                / \
               3   4           ← 只保证"父 < 子",兄弟之间不排序
              / \  / \
             7  8 9  12

小顶堆:                     大顶堆:
        1                          9
       / \                        / \
      3   4                      7   8
     / \  / \                   / \  / \
    7  8 9  12                 3  1 4  2

  根 = 最小值                  根 = 最大值

        1                下标:  0  1  2  3  4  5  6
       / \               数组: [1, 3, 4, 7, 8, 9, 12]
      3   4
     / \  / \            对下标 i (从 0 开始):
    7  8 9  12             左孩子 = 2i + 1
                           右孩子 = 2i + 2
                           父亲   = (i - 1) / 2

插入 2 到这个小顶堆:

        1                    1                      1
       / \                  / \                    / \
      3   4      →         3   4        →         2   4
     / \  / \             / \  / \               / \  / \
    7  8 9  12           7  8 9  12             7  3 9  12
                        /                      /
                       2                      8
                    ← 放末尾                ← 2 < 3,交换
                       2 < 8,交换             2 > 1,停止

// 小顶堆的上浮
func siftUp(h []int, i int) {
    for i > 0 {
        parent := (i - 1) / 2
        if h[parent] <= h[i] {
            break                  // 已经满足堆序,停
        }
        h[parent], h[i] = h[i], h[parent]
        i = parent
    }
}

从小顶堆取出最小值 1:

        1                  12                    3                    3
       / \                / \                   / \                  / \
      3   4      →       3   4        →       12   4       →        7   4
     / \  / \           / \  /                / \  /                / \  /
    7  8 9  12         7  8 9                7  8 9               12  8 9

   取走根 1          末尾的 12 搬到根      12 和较小的孩子       12 和较小的孩子
                     (长度减一)          3 交换               7 交换,没孩子了,停

// 小顶堆的下沉,n 是堆的有效长度
func siftDown(h []int, i, n int) {
    for {
        left, right := 2*i+1, 2*i+2
        smallest := i

        if left < n && h[left] < h[smallest] {
            smallest = left
        }
        if right < n && h[right] < h[smallest] {
            smallest = right      // 关键:挑两个孩子里更小的
        }
        if smallest == i {
            break                 // 自己就是最小,停
        }
        h[i], h[smallest] = h[smallest], h[i]
        i = smallest
    }
}

type Interface interface {
    sort.Interface          // Len() int, Less(i, j int) bool, Swap(i, j int)
    Push(x any)             // 把元素加到末尾
    Pop() any               // 从末尾摘掉元素并返回
}

实现的接口Push(x any)，只负责"追加到末尾",不维护堆序，所以不是真的上浮，heap.Push(h, x) 才是完成的流程。

你实现的 pq.Pop()，只负责"摘掉末尾元素"，不回自动下沉，heap.Pop(h) 才是完整的流程。

// waitFor 保存"要加入的数据"和"该在什么时候加入"
type waitFor[T any] struct {
	data    T
	readyAt time.Time
	// index in the priority queue (heap)
	index int          // ← 记住自己在堆里的位置,后面 heap.Fix 要用
}

// 底层就是一个切片,没有任何多余包装
type waitForPriorityQueue[T any] []*waitFor[T]

func (pq waitForPriorityQueue[T]) Len() int {
	return len(pq)
}

// 比较规则:readyAt 早的排前面 → 小顶堆,根是"最近要到期的"
func (pq waitForPriorityQueue[T]) Less(i, j int) bool {
	return pq[i].readyAt.Before(pq[j].readyAt)
}

// 注意这里:交换元素的同时,必须同步更新它们记录的 index
func (pq waitForPriorityQueue[T]) Swap(i, j int) {
	pq[i], pq[j] = pq[j], pq[i]
	pq[i].index = i
	pq[j].index = j
}

// Push adds an item to the queue. Push should not be called directly; instead, // 注释中直接把上述的坑写在这里
// use `heap.Push`.
func (pq *waitForPriorityQueue[T]) Push(x interface{}) {
	n := len(*pq)
	item := x.(*waitFor[T])
	item.index = n
	*pq = append(*pq, item)      // 只是追加,不管堆序
}

// Pop removes an item from the queue. Pop should not be called directly;
// instead, use `heap.Pop`.
func (pq *waitForPriorityQueue[T]) Pop() interface{} {
	n := len(*pq)
	item := (*pq)[n-1]
	item.index = -1              // 标记"已不在堆中"
	*pq = (*pq)[0:(n - 1)]
	return item
}

// Peek returns the item at the beginning of the queue, without removing the
// item or otherwise mutating the queue. It is safe to call directly.
func (pq waitForPriorityQueue[T]) Peek() interface{} {
	return pq[0]                 // 根 = 最近到期的那个
}

// insert adds the entry to the priority queue, or updates the readyAt if it already exists in the queue
func insert[T comparable](q *waitForPriorityQueue[T], knownEntries map[T]*waitFor[T], entry *waitFor[T]) {
	// if the entry already exists, update the time only if it would cause the item to be queued sooner
	existing, exists := knownEntries[entry.data]
	if exists {
		if existing.readyAt.After(entry.readyAt) {
			existing.readyAt = entry.readyAt    // 改成更早的时间
			heap.Fix(q, existing.index)         // 就地修堆序,O(log n)
		}
		return                                  // 不重复入堆
	}

	heap.Push(q, entry)
	knownEntries[entry.data] = entry
}

for {
    now := q.clock.Now()

    // 1) 先把所有已经到期的都取出来,塞进真正的工作队列
    for waitingForQueue.Len() > 0 {
        entry := waitingForQueue.Peek().(*waitFor[T])
        if entry.readyAt.After(now) {
            break                                   // 根都没到期,后面的更不可能
        }
        entry = heap.Pop(waitingForQueue).(*waitFor[T])
        q.Add(entry.data)
        delete(waitingEntryByData, entry.data)
    }

    // 2) 按"根的到期时间"设一个定时器,而不是轮询
    nextReadyAt := never
    if waitingForQueue.Len() > 0 {
        entry := waitingForQueue.Peek().(*waitFor[T])
        nextReadyAtTimer = q.clock.NewTimer(entry.readyAt.Sub(now))
        nextReadyAt = nextReadyAtTimer.C()
    }

    select {
    case <-q.stopCh:
        return
    case <-q.heartbeat.C():        // 兜底心跳,防止永久阻塞
    case <-nextReadyAt:            // 最近的那个到期了,回到循环顶部
    case waitEntry := <-q.waitingForAddCh:
        // 有新的延时请求进来
    }
}

type heapData struct {
	// items is a map from key of the objects to the objects and their index.
	items map[string]*heapItem     // key → (对象, 在堆里的下标)
	// queue keeps the keys of objects stored in "items".
	queue []string                 // 堆本身,存的是 key 不是对象

	keyFunc  KeyFunc               // 怎么从对象算出 key
	lessFunc LessFunc              // 怎么比较两个对象   ← 优先级规则注入
}

func (h *heapData) Less(i, j int) bool {
	if i > len(h.queue) || j > len(h.queue) {
		return false
	}
	itemi, ok := h.items[h.queue[i]]
	if !ok {
		return false
	}
	itemj, ok := h.items[h.queue[j]]
	if !ok {
		return false
	}
	return h.lessFunc(itemi.obj, itemj.obj)    // 委托给用户传入的函数
}

func (h *Heap) Pop() (interface{}, error) {
	h.lock.Lock()
	defer h.lock.Unlock()
	for len(h.data.queue) == 0 {
		// 队列空 → 阻塞等待,直到有人 Add 或者 Close
		if h.closed {
			return nil, fmt.Errorf("heap is closed")
		}
		h.cond.Wait()
	}
	obj := heap.Pop(h.data)
	// ...
}

type Typed[t comparable] struct {
	queue []t                  // 就是个切片,FIFO

	dirty      sets.Set[t]     // 需要处理的
	processing sets.Set[t]     // 正在处理的
}