Task Queues in Go: A Practical Guide

Introduction: Why Do You Need Task Queues in Go?

In modern development, there is often a need to process tasks asynchronously without blocking the main application thread. Task queues are a classic pattern that allows you to distribute work among multiple handlers (workers), manage load, and ensure fault tolerance.

The Go language (Golang) provides built-in mechanisms for implementing task queues: goroutines and channels. In this article, we will practically examine how to build a reliable task queue using the Go standard library, without resorting to external message brokers (although RabbitMQ or Redis can be useful in production).

Main Part

1. Basic Worker Pool Pattern

A Worker Pool is a set of a fixed number of goroutines (workers) that listen to a single channel and process incoming tasks. This allows you to control the degree of parallelism and avoid system overload.

Let's look at a simple example: we create a pool of 3 workers that process integers (simulating work).

package main\n\nimport (\n    "fmt"\n    "sync"\n    "time"\n)\n\nfunc worker(id int, jobs <-chan int, wg *sync.WaitGroup) {\n    defer wg.Done()\n    for job := range jobs {\n        fmt.Printf("Worker %d started task %d\\n", id, job)\n        time.Sleep(time.Second) // simulating work\n        fmt.Printf("Worker %d finished task %d\\n", id, job)\n    }\n}\n\nfunc main() {\n    const numJobs = 5\n    const numWorkers = 3\n\n    jobs := make(chan int, numJobs)\n    var wg sync.WaitGroup\n\n    // Start workers\n    for w := 1; w <= numWorkers; w++ {\n        wg.Add(1)\n        go worker(w, jobs, &wg)\n    }\n\n    // Send tasks\n    for j := 1; j <= numJobs; j++ {\n        jobs <- j\n    }\n    close(jobs) // close the channel so workers can finish\n\n    wg.Wait()\n    fmt.Println("All tasks completed")\n}

Code Breakdown:

\n
• jobs — a buffered channel to which we send tasks.
\n
• worker — a function that reads from the channel and processes each task.
\n
• sync.WaitGroup — waits for all workers to finish.
\n
• After sending all tasks, the channel is closed (close(jobs)), which causes the for range loop in the workers to terminate.
\n

This pattern is the foundation for more complex solutions.

2. Queue with Priorities and Timeouts

In real-world applications, tasks may have different priorities or need to be completed within a certain time. Let's implement a queue where priority tasks are processed first, and also add a context with a timeout to cancel long-running tasks.

package main\n\nimport (\n    "context"\n    "fmt"\n    "time"\n)\n\ntype Task struct {\n    ID       int\n    Priority int // the lower the number, the higher the priority\n    Data     string\n}\n\nfunc processTask(ctx context.Context, task Task) error {\n    select {\n    case <-ctx.Done():\n        return ctx.Err()\n    case <-time.After(time.Duration(task.Priority) * 500 * time.Millisecond):\n        fmt.Printf("Processed task %d: %s (priority %d)\\n", task.ID, task.Data, task.Priority)\n        return nil\n    }\n}\n\nfunc main() {\n    tasks := []Task{\n        {ID: 1, Priority: 3, Data: "Low priority"},\n        {ID: 2, Priority: 1, Data: "High priority"},\n        {ID: 3, Priority: 2, Data: "Medium priority"},\n    }\n\n    // Sort by priority (highest first)\n    // In a real project, use sorting or a heap (container/heap)\n    for i := 0; i < len(tasks); i++ {\n        for j := i + 1; j < len(tasks); j++ {\n            if tasks[j].Priority < tasks[i].Priority {\n                tasks[i], tasks[j] = tasks[j], tasks[i]\n            }\n