GoConcurrentQueue - 高性能并发FIFO队列插件库使用指南

GoConcurrentQueue 是一个专门为 Go 语言设计的高性能并发安全 FIFO 队列库，它提供了简单易用的 API 和出色的并发性能。

安装

go get github.com/enriquebris/goconcurrentqueue

基本使用

创建队列

package main

import (
	"fmt"
	"github.com/enriquebris/goconcurrentqueue"
)

func main() {
	// 创建一个新的FIFO队列
	queue := goconcurrentqueue.NewFIFO()
	
	// 添加元素到队列
	queue.Enqueue("item1")
	queue.Enqueue(2)
	queue.Enqueue(3.14)
	
	// 获取队列长度
	fmt.Println("Queue length:", queue.GetLen()) // 输出: Queue length: 3
	
	// 从队列中取出元素
	item, err := queue.Dequeue()
	if err != nil {
		fmt.Println("Dequeue error:", err)
		return
	}
	fmt.Println("Dequeued item:", item) // 输出: Dequeued item: item1
}

高级特性

并发安全操作

package main

import (
	"fmt"
	"sync"
	"time"
	"github.com/enriquebris/goconcurrentqueue"
)

func producer(queue *goconcurrentqueue.FIFO, id int, wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 0; i < 5; i++ {
		item := fmt.Sprintf("producer-%d-item-%d", id, i)
		queue.Enqueue(item)
		fmt.Printf("Producer %d enqueued: %s\n", id, item)
		time.Sleep(time.Millisecond * 100)
	}
}

func consumer(queue *goconcurrentqueue.FIFO, id int, wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 0; i < 5; i++ {
		item, err := queue.Dequeue()
		if err != nil {
			fmt.Printf("Consumer %d error: %v\n", id, err)
			continue
		}
		fmt.Printf("Consumer %d dequeued: %v\n", id, item)
		time.Sleep(time.Millisecond * 150)
	}
}

func main() {
	queue := goconcurrentqueue.NewFIFO()
	var wg sync.WaitGroup

	// 启动3个生产者协程
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go producer(queue, i, &wg)
	}

	// 启动2个消费者协程
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go consumer(queue, i, &wg)
	}

	wg.Wait()
	fmt.Println("Final queue length:", queue.GetLen())
}

阻塞式获取元素

package main

import (
	"fmt"
	"time"
	"github.com/enriquebris/goconcurrentqueue"
)

func main() {
	queue := goconcurrentqueue.NewFIFO()

	// 启动一个消费者协程
	go func() {
		// 阻塞等待直到有元素可用
		item, err := queue.DequeueOrWaitForNextElement()
		if err != nil {
			fmt.Println("Error waiting for element:", err)
			return
		}
		fmt.Println("Received item after waiting:", item)
	}()

	// 主协程稍后添加元素
	time.Sleep(2 * time.Second)
	queue.Enqueue("delayed item")

	// 给消费者时间处理
	time.Sleep(time.Second)
}

带超时的阻塞获取

package main

import (
	"fmt"
	"time"
	"github.com/enriquebris/goconcurrentqueue"
)

func main() {
	queue := goconcurrentqueue.NewFIFO()

	// 尝试获取元素，最多等待1秒
	item, err := queue.DequeueOrWaitForNextElementWithTimeout(time.Second)
	if err != nil {
		if err == goconcurrentqueue.ErrEmptyQueue {
			fmt.Println("Queue is empty after waiting")
		} else {
			fmt.Println("Error:", err)
		}
		return
	}
	fmt.Println("Received item:", item)
}

性能优化建议

1. 批量操作：对于大量小项，考虑批量入队/出队以减少锁竞争
2. 适当队列大小：根据场景选择合适队列容量
3. 避免长时间持有锁：在队列操作中尽量减少耗时操作

与其他队列实现的比较

GoConcurrentQueue 相比标准库的 channel 和 sync.Mutex 保护的 slice 有以下优势：

• 更丰富的 API（阻塞获取、超时等）
• 动态容量（不像 channel 需要预先指定大小）
• 专门优化的并发性能

总结

GoConcurrentQueue 是一个功能强大且易于使用的并发安全队列实现，特别适合需要高性能 FIFO 队列的 Go 并发程序。它提供了丰富的特性集，包括阻塞获取、超时控制等，同时保持了优异的性能表现。