Go语言实现延迟队列

package main
import (  "fmt"  "time")
// Message is a structure that holds the contents of the message.type Message struct {  ID     int  Body   string  Delay  time.Duration}
// delayMessageQueue holds the queue of messagestype delayMessageQueue struct {  queue []Message}
func (d *delayMessageQueue) send(msg Message) {  go func() {    timer := time.NewTimer(msg.Delay)    <-timer.C    d.queue = append(d.queue, msg)    fmt.Printf("Message %d sent\n", msg.ID)  }()}
func (d *delayMessageQueue) receive() {  for {    if len(d.queue) > 0 {      msg := d.queue[0]      fmt.Printf("Message %d received \n", msg.ID)      d.queue = d.queue[1:] // Dequeue    }  end}
func main() {  dmq := &delayMessageQueue{}  dmq.send(Message{ID: 1, Body: "Hello, World", Delay: 2 * time.Second})  dmq.send(Message{ID: 2, Body: "Hello, Go", Delay: 1 * time.Second})
  // Keep the main function alive to let goroutines finish  time.Sleep(5 * time.Second)
  dmq.receive()}

package main
import (  "container/heap"  "fmt"  "sync"  "time")
// 优先队列内部维护的队列元素type Item struct {  value    string  priority int64 // 延迟任务到期时间(用时间戳表示)  index    int   // 队列元素在堆中的索引}
// 优先队列：底层用最小堆实现type PriorityQueue []*Item
func (pq PriorityQueue) Len() int { return len(pq) }
// 注意这里比较的是任务的优先级，为了让离当前时间最近的任务在堆顶，我们让比较结果颠倒func (pq PriorityQueue) Less(i, j int) bool {  return pq[i].priority < pq[j].priority}
func (pq PriorityQueue) Swap(i, j int) {  pq[i], pq[j] = pq[j], pq[i]  pq[i].index = i  pq[j].index = j}
// 插入元素func (pq *PriorityQueue) Push(x interface{}) {  n := len(*pq)  item := x.(*Item)  item.index = n  *pq = append(*pq, item)}
// 删除元素func (pq *PriorityQueue) Pop() interface{} {  old := *pq  n := len(old)  item := old[n-1]  old[n-1] = nil  // avoid memory leak  item.index = -1 // for safety  *pq = old[0 : n-1]  return item}
var mutex sync.Mutex // 并发控制互斥锁
// 设定好比cmpTime更晚的任务执行时间并加入堆func addTask(pq *PriorityQueue, cmpTime int64, diff int64) {  mutex.Lock() // 加锁  taskTime := cmpTime + diff*int64(time.Second)  item := &Item{    value:    fmt.Sprintf("任务%d", taskTime),    priority: taskTime,  }  heap.Push(pq, item)  mutex.Unlock() // 解锁}
// 从堆上取出时间最早的任务执行func doTask(pq *PriorityQueue) {  for {    mutex.Lock()    if len(*pq) == 0 {      mutex.Unlock()      continue // 堆空则不处理    }    // 堆非空，取出最早任务项    item := heap.Pop(pq).(*Item)    now := time.Now().Unix()    if item.priority-now > 0 {      // 未到执行时间，任务重新入堆      heap.Push(pq, item)    } else {      // 执行任务      fmt.Printf("%s 执行\n", item.value)    }    mutex.Unlock()    // 防止doTask过于紧密，每次循环停顿1秒    time.Sleep(1 * time.Second)  }}
func main() {  pq := make(PriorityQueue, 0)  heap.Init(&pq)
  // 启动执行任务goroutine  go doTask(&pq)
  now := time.Now().Unix()  // 预设3个延迟任务，延迟时间分别为8s, 2s, 5s  addTask(&pq, now, 8)  addTask(&pq, now, 2)  addTask(&pq, now, 5)
  // 保持主进程  time.Sleep(15 * time.Second)}