Golang并发队列实现与使用详解

各位小伙伴们，大家好呀！看看今天我又给各位带来了什么文章？本文标题是《Golang并发队列实现与使用教程》，很明显是关于Golang的文章哈哈哈，其中内容主要会涉及到等等，如果能帮到你，觉得很不错的话，欢迎各位多多点评和分享！

Golang中通过sync.Mutex和sync.Cond结合container/list实现并发队列，确保多Goroutine下安全存取。

Golang中实现并发队列，核心在于确保多个Goroutine在同时存取数据时，不会发生竞态条件导致数据损坏或逻辑错误。我们通常会借助Go语言自带的并发原语，如sync.Mutex结合标准库的container/list，或者更Go风格的chan（通道）来实现。无论是哪种方式，目标都是提供一个线程安全的数据结构，让生产者和消费者可以安全地进行入队和出队操作。

一个基于sync.Mutex和container/list的并发队列实现，能提供更灵活的容量控制和阻塞/非阻塞行为，特别是当我们需要一个能明确判断队列空满状态，并能根据需要进行等待或立即返回的队列时。

package main

import (
    "container/list"
    "fmt"
    "sync"
    "time"
)

// ConcurrentQueue 定义了一个并发安全的队列
type ConcurrentQueue struct {
    queue *list.List      // 实际存储数据的链表
    mutex sync.Mutex      // 保护队列的互斥锁
    cond  *sync.Cond      // 条件变量，用于等待和通知
    cap   int             // 队列容量，0表示无限制
}

// NewConcurrentQueue 创建一个新的并发队列
// capacity为0表示无限制容量，否则为固定容量
func NewConcurrentQueue(capacity int) *ConcurrentQueue {
    q := &ConcurrentQueue{
        queue: list.New(),
        cap:   capacity,
    }
    q.cond = sync.NewCond(&q.mutex) // 条件变量需要一个Locker
    return q
}

// Enqueue 将元素加入队列
func (q *ConcurrentQueue) Enqueue(item interface{}) error {
    q.mutex.Lock()
    defer q.mutex.Unlock()

    // 如果有容量限制，且队列已满，则等待
    for q.cap > 0 && q.queue.Len() >= q.cap {
        q.cond.Wait() // 释放锁并等待，被唤醒后重新获取锁
    }

    q.queue.PushBack(item)
    q.cond.Signal() // 通知一个等待的消费者
    return nil
}

// Dequeue 从队列中取出元素
func (q *ConcurrentQueue) Dequeue() (interface{}, error) {
    q.mutex.Lock()
    defer q.mutex.Unlock()

    // 如果队列为空，则等待
    for q.queue.Len() == 0 {
        q.cond.Wait() // 释放锁并等待，被唤醒后重新获取锁
    }

    element := q.queue.Front()
    q.queue.Remove(element)
    q.cond.Signal() // 通知一个等待的生产者（如果队列曾满而等待）
    return element.Value, nil
}

// TryEnqueue 尝试将元素加入队列，如果队列满则立即返回错误
func (q *ConcurrentQueue) TryEnqueue(item interface{}) error {
    q.mutex.Lock()
    defer q.mutex.Unlock()

    if q.cap > 0 && q.queue.Len() >= q.cap {
        return fmt.Errorf("queue is full")
    }

    q.queue.PushBack(item)
    q.cond.Signal()
    return nil
}

// TryDequeue 尝试从队列中取出元素，如果队列空则立即返回错误
func (q *ConcurrentQueue) TryDequeue() (interface{}, error) {
    q.mutex.Lock()
    defer q.mutex.Unlock()

    if q.queue.Len() == 0 {
        return nil, fmt.Errorf("queue is empty")
    }

    element := q.queue.Front()
    q.queue.Remove(element)
    q.cond.Signal()
    return element.Value, nil
}

// Len 返回队列当前长度
func (q *ConcurrentQueue) Len() int {
    q.mutex.Lock()
    defer q.mutex.Unlock()
    return q.queue.Len()
}

// IsEmpty 判断队列是否为空
func (q *ConcurrentQueue) IsEmpty() bool {
    return q.Len() == 0
}

func main() {
    queue := NewConcurrentQueue(5) // 创建一个容量为5的并发队列
    var wg sync.WaitGroup

    // 生产者
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            item := fmt.Sprintf("数据-%d", id)
            if err := queue.Enqueue(item); err != nil {
                fmt.Printf("生产者%d: 尝试入队 %s 失败: %v\n", id, item, err)
            } else {
                fmt.Printf("生产者%d: 入队 %s, 当前队列长度: %d\n", id, item, queue.Len())
            }
        }(i)
    }

    // 消费者
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            // 稍微延迟一下，让生产者有机会先生产一些数据
            time.Sleep(time.Millisecond * 50)
            item, err := queue.Dequeue()
            if err != nil {
                fmt.Printf("消费者%d: 尝试出队失败: %v\n", id, err)
            } else {
                fmt.Printf("消费者%d: 出队 %v, 当前队列长度: %d\n",

今天关于《Golang并发队列实现与使用详解》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！