Go 怎么基于 chan 实现低延迟队列？

Question

下面是我实现的队列代码。

如果段时间添加大量任务，可能会导致 chan 缓冲用完导致添加任务失败（或者不使用select导致添加任务时阻塞），比如设置缓冲大小为 1 模拟缓冲耗尽。

问题是，如何让队列的大小不受 chan 缓冲区限制，而同时又能享受到 chan 队列的低延迟？

https://goplay.tools/snippet/DNsfIru_-5h

package queue

import (
    "fmt"
    "time"
)

var QueueChan = make(chan string, 1)

func worker(id string) {
    fmt.Printf("%q start\n", id)
    time.Sleep(time.Second * 2)
    fmt.Printf("%q done\n", id)
}

func consume() {
    for id := range QueueChan {
        worker(id)
    }
}

func Run() {
    go consume()

    AddTask("1")
    AddTask("2")
    time.Sleep(time.Second * 3)
    AddTask("3")
    AddTask("4")
    AddTask("5")
    AddTask("6")
}

func AddTask(id string) {
    select {
    case QueueChan <- id:
        fmt.Printf("add %q to queue\n", id)
    default:
        fmt.Printf("queue full\n")
    }
}

Answer 1

使用 chan 和 slice 来实现一个无界缓冲队列：

package queue

import (
    "fmt"
    "sync"
    "time"
)

type UnboundedQueue struct {
    inChan  chan string
    outChan chan string
    buffer  []string
    mu      sync.Mutex
}

func NewUnboundedQueue() *UnboundedQueue {
    q := &UnboundedQueue{
        inChan:  make(chan string),
        outChan: make(chan string),
        buffer:  make([]string, 0),
    }
    go q.run()
    return q
}

func (q *UnboundedQueue) run() {
    for {
        var out chan string
        var next string
        if len(q.buffer) > 0 {
            out = q.outChan
            next = q.buffer[0]
        }
        select {
        case val := <-q.inChan:
            q.mu.Lock()
            q.buffer = append(q.buffer, val)
            q.mu.Unlock()
        case out <- next:
            q.mu.Lock()
            q.buffer = q.buffer[1:]
            q.mu.Unlock()
        }
    }
}

func (q *UnboundedQueue) In(val string) {
    q.inChan <- val
}

func (q *UnboundedQueue) Out() string {
    return <-q.outChan
}

func worker(id string) {
    fmt.Printf("%q start\n", id)
    time.Sleep(time.Second * 2)
    fmt.Printf("%q done\n", id)
}

func consume(q *UnboundedQueue) {
    for {
        id := q.Out()
        worker(id)
    }
}

func Run() {
    q := NewUnboundedQueue()
    go consume(q)

    q.In("1")
    q.In("2")
    time.Sleep(time.Second * 3)
    q.In("3")
    q.In("4")
    q.In("5")
    q.In("6")
}

1.UnboundedQueue 结构体包含 inChan 和 outChan 通道，以及一个用于存储数据的 buffer 切片。
2.run 方法在一个单独的 goroutine 中运行，使用 select 语句处理 inChan 和 outChan 的数据传输。
3.In 方法将数据写入 inChan，Out 方法从 outChan 读取数据。
4.consume 函数从队列中读取数据并处理。

推荐第三方库

如果你的主要需求是高效、低延迟的队列处理，ZenQ 是一个很好的选择，因为它在性能和易用性方面表现出色。如果你需要更高的吞吐量和更复杂的生产者-消费者模式，4FQ 可能更适合你。如果你需要分布式消息传递，Emitter 是一个不错的选择。