Golang并发安全队列实现 channel方案-Golang-PHP中文网

Golang并发安全队列实现 channel方案

P粉602998670

发布： 2025-08-19 12:30:02

原创

708人浏览过

使用channel是Go中实现并发安全队列最推荐的方式，因其天然支持多goroutine安全操作，无需额外加锁，运行时自动处理同步互斥，避免数据竞争。

golang并发安全队列实现 channel方案

在Go语言中，实现并发安全的队列最简单且推荐的方式就是使用内置的 channel。channel 本身就是为并发通信设计的，天然支持多个 goroutine 安全地进行入队和出队操作，无需额外加锁。

为什么 channel 是并发安全的？

Go 的 channel 在语言层面就保证了线程安全。多个 goroutine 同时向同一个 channel 发送或接收数据时，运行时系统会自动处理同步和互斥，不会出现数据竞争。这使得 channel 成为构建并发安全队列的理想选择。

使用 channel 实现基础并发安全队列

最简单的并发安全队列可以直接用带缓冲的 channel 实现：

type Queue struct {
    ch chan interface{}
}
<p>// NewQueue 创建一个指定容量的队列
func NewQueue(capacity int) *Queue {
return &Queue{
ch: make(chan interface{}, capacity),
}
}</p><p>// Enqueue 入队，非阻塞（如果队列满则返回错误）
func (q *Queue) Enqueue(item interface{}) bool {
select {
case q.ch <- item:
return true
default:
return false // 队列已满
}
}</p><p>// Dequeue 出队，非阻塞（如果队列空则返回 false）
func (q *Queue) Dequeue() (interface{}, bool) {
select {
case item := <-q.ch:
return item, true
default:
return nil, false // 队列为空
}
}</p><p>// TryDequeue 带超时的出队
func (q *Queue) TryDequeue(timeout time.Duration) (interface{}, bool) {
select {
case item := <-q.ch:
return item, true
case <-time.After(timeout):
return nil, false
}
}</p>

登录后复制

更灵活的无界队列（基于 channel + slice）

如果需要无界队列，可以结合 channel 和互斥锁实现一个基于 slice 的队列，channel 用于通知消费者有新数据到达：

立即学习“go语言免费学习笔记（深入）”；

ViiTor实时翻译

AI实时多语言翻译专家！强大的语音识别、AR翻译功能。

116

查看详情

type UnboundedQueue struct {
    mu   sync.Mutex
    data []interface{}
    cond *sync.Cond
    closed bool
}
<p>func NewUnboundedQueue() *UnboundedQueue {
q := &UnboundedQueue{
data: make([]interface{}, 0),
}
q.cond = sync.NewCond(&q.mu)
return q
}</p><p>func (q *UnboundedQueue) Enqueue(item interface{}) {
q.mu.Lock()
defer q.mu.Unlock()
q.data = append(q.data, item)
q.cond.Signal() // 唤醒一个等待的消费者
}</p><p>func (q *UnboundedQueue) Dequeue() (interface{}, bool) {
q.mu.Lock()
defer q.mu.Unlock()
for len(q.data) == 0 && !q.closed {
q.cond.Wait() // 等待数据
}
if len(q.data) == 0 {
return nil, false
}
item := q.data[0]
q.data = q.data[1:]
return item, true
}</p><p>func (q *UnboundedQueue) Close() {
q.mu.Lock()
defer q.mu.Unlock()
q.closed = true
q.cond.Broadcast() // 唤醒所有等待者
}</p>

登录后复制

这种方式适合生产者频繁写入、消费者可能长时间等待的场景。