Concurrent programming is one of the most interesting and challenging aspects of software engineering. When multiple goroutines need to share data, coordinating access becomes critical.
One elegant solution to this is the blocking queue, a data structure that changes how we think about thread coordination. In this article, we will explore what blocking queues are, how they work, and why they’re essential for building robust concurrent systems.
What is a Blocking Queue?
A blocking queue is a thread-safe queue that supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element. This blocking behavior is what makes it fundamentally different from regular queues.
In traditional queue implementations, when you try to remove an element from an empty queue, you typically get a null value or an error. Similarly, if you try to add an element to a full bounded queue, the operation either fails immediately or returns an error.
Blocking queues take a different approach. Instead of failing immediately, they pause the calling goroutine/thread until the operation can be successfully completed. This blocking behavior solves two critical problems in concurrent programming.
- it eliminates busy-waiting, where a goroutine continuously checks a condition in a loop, wasting CPU cycles.
- it provides automatic synchronization between producer and consumer goroutines without requiring explicit lock management or condition variable constructs.
The Producer-Consumer Pattern
The blocking queue is almost a de facto in a producer-consumer design pattern. In this pattern, producer goroutines generate data or tasks and place them into a shared queue. Consumer goroutines retrieve items from this queue and process them. The beauty of using a blocking queue lies in its ability to automatically coordinate these two types of goroutines.
When producers are faster than consumers, the queue fills up. At this point, the blocking queue automatically pauses producer goroutines, preventing memory overflow. When consumers are faster than producers, the queue empties out. The blocking queue then pauses consumer goroutines, preventing them from wasting CPU cycles checking for work that doesn’t exist yet.
This coordination happens transparently, without requiring us to write complex synchronization code. The blocking queue handles all the coordination internally, making the code cleaner and less error-prone.
Core Operations
A blocking queue provides several methods for adding and removing elements, each with different blocking and timeout behaviors.
The put operation inserts an element into the queue, blocking if necessary until space becomes available. This is the primary method for producers. Here’s a simple example in Go:
type BlockingQueue struct {
queue chan Task
}
func NewBlockingQueue(capacity int) *BlockingQueue {
return &BlockingQueue{
queue: make(chan Task, capacity),
}
}
// Producer goroutine
func produce(bq *BlockingQueue) {
task := createTask()
bq.queue <- task // Blocks if queue is full
} The take operation retrieves and removes the head of the queue, blocking if necessary until an element becomes available. This is the primary method for consumers.
// Consumer goroutine
func consume(bq *BlockingQueue) {
task := <-bq.queue // Blocks if the queue is empty
processTask(task)
} For scenarios where indefinite blocking isn’t desirable, timeout-based alternatives can be implemented using select statements with time.After:
func (bq *BlockingQueue) OfferWithTimeout(task Task, timeout time.Duration) bool {
select {
case bq.queue <- task:
return true
case <-time.After(timeout):
return false
}
}
func (bq *BlockingQueue) PollWithTimeout(timeout time.Duration) (Task, bool) {
select {
case task := <-bq.queue:
return task, true
case <-time.After(timeout):
return Task{}, false
}
} There are also non-blocking variants. These can be implemented using select with a default case:
func (bq *BlockingQueue) TryOffer(task Task) bool {
select {
case bq.queue <- task:
return true
default:
return false
}
}
func (bq *BlockingQueue) TryPoll() (Task, bool) {
select {
case task := <-bq.queue:
return task, true
default:
return Task{}, false
}
} Bounded vs Unbounded Queues
In Go, channels are naturally bounded when created with a capacity. A buffered channel with capacity N can hold up to N elements before blocking senders:
// Bounded queue with capacity of 100
queue := make(chan string, 100)
// Unbuffered channel - effectively a queue of size 0
queue := make(chan string) Bounded queues provide predictable memory usage. Since the buffer is allocated upfront, there are no surprises with memory allocation during runtime. This predictability is valuable in systems where memory management is critical.
The bounded nature also provides natural backpressure. When the queue is full, producers automatically block, giving consumers time to catch up. This prevents memory overflow in scenarios where producers consistently outpace consumers.
Unbuffered channels are useful for direct synchronization between goroutines, where you want to ensure the sender and receiver meet at the same time. However, for most producer-consumer scenarios, a buffered channel with reasonable capacity provides better throughput.
Building a Producer-Consumer System
Let’s build a complete producer-consumer system to see blocking queues in action. This example demonstrates a task processing system where multiple producers generate tasks and multiple consumers process them.
First, let’s define our task:
type Task struct {
ID string
Data string
} Now, let’s create a producer:
func producer(queue chan Task, producerID string, tasksToGenerate int, wg *sync.WaitGroup) {
defer wg.Done()
for i := 0; i < tasksToGenerate; i++ {
task := Task{
ID: fmt.Sprintf("%s-%d", producerID, i),
Data: fmt.Sprintf("Data from %s task %d", producerID, i),
}
queue <- task
fmt.Printf("%s produced: %s\n", producerID, task.ID)
// Simulate work
time.Sleep(100 * time.Millisecond)
}
} Next, the consumer:
func consumer(queue chan Task, consumerID string, done chan bool) {
for {
select {
case task := <-queue:
processTask(consumerID, task)
case <-done:
fmt.Printf("%s shutting down\n", consumerID)
return
}
}
}
func processTask(consumerID string, task Task) {
fmt.Printf("%s processing: %s\n", consumerID, task.ID)
// Simulate processing time
time.Sleep(200 * time.Millisecond)
fmt.Printf("%s completed: %s\n", consumerID, task.ID)
} Finally, let’s tie it all together:
func main() {
// Create a bounded blocking queue
queue := make(chan Task, 10)
// Create producer wait group
var producerWg sync.WaitGroup
// Create and start producers
numProducers := 3
for i := 0; i < numProducers; i++ {
producerWg.Add(1)
go producer(queue, fmt.Sprintf("Producer-%d", i), 5, &producerWg)
}
// Create done channels for consumers
numConsumers := 2
consumerDone := make([]chan bool, numConsumers)
for i := 0; i < numConsumers; i++ {
consumerDone[i] = make(chan bool)
go consumer(queue, fmt.Sprintf("Consumer-%d", i), consumerDone[i])
}
// Wait for all producers to finish
producerWg.Wait()
// Give consumers time to finish processing remaining tasks
time.Sleep(5 * time.Second)
// Stop consumers
for i := 0; i < numConsumers; i++ {
consumerDone[i] <- true
}
fmt.Printf("All tasks processed. Queue size: %d\n", len(queue))
} In the above example, producers automatically block when the queue is full, preventing memory overflow. Consumers automatically block when the queue is empty, avoiding busy-waiting. The system gracefully handles shutdown by allowing producers to finish and giving consumers time to drain the queue.
Integration with Worker Pools
Blocking queues are fundamental to worker pool implementations. A worker pool uses a channel to distribute tasks among a fixed number of worker goroutines:
type WorkerPool struct {
tasks chan Task
workers int
workerWg sync.WaitGroup
shutdown chan bool
}
func NewWorkerPool(workers int, queueSize int) *WorkerPool {
wp := &WorkerPool{
tasks: make(chan Task, queueSize),
workers: workers,
shutdown: make(chan bool),
}
// Start workers
for i := 0; i < workers; i++ {
wp.workerWg.Add(1)
go wp.worker(i)
}
return wp
}
func (wp *WorkerPool) worker(id int) {
defer wp.workerWg.Done()
for {
select {
case task := <-wp.tasks:
fmt.Printf("Worker %d processing task %s\n", id, task.ID)
processTask(fmt.Sprintf("Worker-%d", id), task)
case <-wp.shutdown:
return
}
}
}
func (wp *WorkerPool) Submit(task Task) {
wp.tasks <- task
}
func (wp *WorkerPool) Shutdown() {
close(wp.tasks)
close(wp.shutdown)
wp.workerWg.Wait()
} The interaction between pool sizing and the blocking queue is important. When a task is submitted to the pool, here’s what happens:
If the queue is not full, the task is added to the queue immediately. A worker goroutine will pick it up when available.
If the queue is full, the Submit call blocks until space becomes available. This provides natural backpressure.
Different queue strategies affect worker pool behavior. An unbuffered channel forces direct handoff between submitter and worker, ensuring tasks are processed immediately but reducing throughput. A large buffered channel allows burst traffic but may increase latency for individual tasks.
Pipeline Processing
Pipeline processing is another common pattern where multiple stages of processing are connected by blocking queues. Each stage has its own consumer goroutines that take work from one queue, process it, and put results into the next queue:
func pipelineStage(input, output chan string, stageName string, wg *sync.WaitGroup) {
defer wg.Done()
for data := range input {
if data == "PILL" {
output <- "PILL"
return
}
processed := processData(stageName, data)
output <- processed
}
}
func processData(stageName, input string) string {
return fmt.Sprintf("%s:%s", stageName, strings.ToUpper(input))
}
func runPipeline() {
queue1 := make(chan string, 10)
queue2 := make(chan string, 10)
queue3 := make(chan string, 10)
var wg sync.WaitGroup
wg.Add(2)
go pipelineStage(queue1, queue2, "Stage1", &wg)
go pipelineStage(queue2, queue3, "Stage2", &wg)
// Feed the pipeline
queue1 <- "data"
queue1 <- "PILL"
close(queue1)
// Retrieve results
result := <-queue3
fmt.Println(result)
wg.Wait()
} Delay Queue
For scenarios requiring delayed task execution, you can implement a delay queue using time.After:
type DelayedTask struct {
executeAt time.Time
description string
}
type DelayQueue struct {
tasks chan DelayedTask
}
func NewDelayQueue() *DelayQueue {
return &DelayQueue{
tasks: make(chan DelayedTask, 100),
}
}
func (dq *DelayQueue) Put(task DelayedTask) {
dq.tasks <- task
}
func (dq *DelayQueue) Take() DelayedTask {
task := <-dq.tasks
// Wait until the task's execution time
delay := time.Until(task.executeAt)
if delay > 0 {
time.Sleep(delay)
}
return task
} Real-world Applications
In web servers, blocking queues manage request processing. When requests arrive faster than they can be processed, they’re queued. Worker goroutines take requests from the queue and process them. This prevents server overload while maintaining responsiveness:
Batch processing systems use blocking queues to coordinate work. Records are read from a source, placed in a queue, processed by worker goroutines, and results are written to another queue for further processing or output
Log aggregation systems use blocking queues to buffer log entries. Multiple goroutines produce log entries, which are queued and written to disk or sent to a logging service by dedicated goroutines. This prevents logging from blocking application goroutines.
Common Pitfalls and Best Practices
Unbounded queues can lead to memory exhaustion. In Go, creating very large buffered channels can consume significant memory. Always use bounded channels when possible:
// Bad: Very large buffer that could exhaust memory
queue := make(chan Task, 1000000)
// Good: Reasonable buffer size
queue := make(chan Task, 1000) Goroutine leaks are a common problem when channels aren’t properly closed. Always ensure goroutines have a way to exit:
// Good: Proper cleanup with the done channel
done := make(chan bool)
go func() {
for {
select {
case task := <-queue:
process(task)
case <-done:
return
}
}
}()
// Later
done <- true Poison pills are a common pattern for signaling shutdown, but they require careful implementation. Each consumer needs to receive a poison pill, or you can close the channel:
// Signal shutdown by closing the channel
close(queue)
// Consumers will exit when the channel is drained
for task := range queue {
process(task)
} Capacity sizing is critical. Too small, and producers spend too much time blocked. Too large, and you waste memory and may increase latency. Profile your application under realistic load to find the right balance:
// Consider your use case
// High-throughput, bursty load: larger queue
queue := make(chan Task, 5000)
// Memory-constrained, steady load: smaller queue
queue := make(chan Task, 100) Always check if a channel operation succeeded when using select with default:
// Good: Check if send succeeded
select {
case queue <- task:
// Successfully queued
default:
// Queue full, handle appropriately
log.Println("Queue full, dropping task")
} When using multiple channels, be careful about deadlocks. Make sure there’s always a goroutine ready to receive when you send, or use buffered channels:
// Potential deadlock with unbuffered channels
ch1 := make(chan int)
ch2 := make(chan int)
// This could deadlock if not careful
go func() {
ch1 <- 1
<-ch2
}()
ch2 <- 2 // Might block forever
<-ch1 Monitor queue sizes in production. Growing queue sizes often indicate that consumers can’t keep up with producers. Use metrics to track queue depth:
// Monitor queue size
if len(queue) > cap(queue)*8/10 {
log.Printf("Queue nearly full: %d/%d", len(queue), cap(queue))
} Performance Considerations
Channel operations in Go are highly optimized. For most use cases, buffered channels provide excellent performance. The runtime uses efficient lock-free algorithms for channel operations when possible.
Memory allocation patterns matter. Buffered channels allocate all memory upfront, providing predictable memory usage. However, very large buffers can waste memory if the queue rarely fills up.
For single-producer, single-consumer scenarios, Go channels are already highly optimized. The runtime detects this pattern and uses specialized fast paths.
Batch operations improve throughput by reducing synchronization overhead:
func consumeBatch(queue chan Task) {
batch := make([]Task, 0, 100)
// Block for first item
task := <-queue
batch = append(batch, task)
// Drain additional available items without blocking
for len(batch) < 100 {
select {
case task := <-queue:
batch = append(batch, task)
default:
break
}
}
// Process entire batch
for _, t := range batch {
process(t)
}
} Footnote
Blocking queue is a key abstraction in concurrent programming, providing a simple and powerful mechanism for coordinating work between goroutines.
It automatically handles the complex synchronization required for producer-consumer scenarios and eliminates entire classes of concurrency bugs while making code more readable and maintainable.