Task Queues in Go: Creation, Processing, and Optimization

Introduction: Why Do You Need Job Queues in Go?

In modern backend development, tasks often arise that do not require an immediate response to the user: sending emails, generating reports, processing images, synchronizing with external APIs. Performing such operations synchronously within an HTTP request is bad practice, as it slows down the response and increases server load.

Job queues solve this problem. In Go (Golang), built-in concurrency mechanisms — goroutines and channels — allow you to build reliable and performant queues without external dependencies. In this article, we will look at how to create a job queue from scratch, optimize it using a worker pool, and avoid common mistakes.

Basics: Goroutines and Channels

Go provides two key primitives for working with queues:

• Goroutines — lightweight threads that are launched using the go keyword.
• Channels — typed "pipes" for transferring data between goroutines.

The simplest queue is a goroutine that reads from a channel and processes tasks:

package main

import (    "fmt"    "time")

func worker(jobs <-chan int) {    for job := range jobs {        fmt.Printf("Processing task %d", job)        time.Sleep(1 * time.Second) // simulating work    }}

func main() {    jobs := make(chan int, 10)

    // start the worker    go worker(jobs)

    // send tasks    for i := 1; i <= 5; i++ {        jobs <- i    }

    close(jobs) // close the channel after sending all tasks    time.Sleep(2 * time.Second)}

This code creates one worker that processes tasks sequentially. But for real projects, scalability is needed — multiple workers.

Implementing a Worker Pool in Go

A worker pool is a pattern where we launch a fixed number of worker goroutines, each of which takes tasks from a common channel. This allows you to control the degree of parallelism and system load.

Basic Implementation

package main

import (    "fmt"    "sync"    "time")

func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {    defer wg.Done()    for job := range jobs {        fmt.Printf("Worker %d: task %d", id, job)        time.Sleep(500 * time.Millisecond)    }}

func main() {    const numWorkers = 3    const numJobs = 10

    jobs := make(chan int, numJobs)    var wg sync.WaitGroup

    // start workers    for i := 1; i <= numWorkers; i++ {        wg.Add(1)        go worker(i, jobs, &wg)    }

    // send tasks    for j := 1; j <= numJobs; j++ {        jobs <- j    }    close(jobs)

    wg.Wait() // wait for all workers to finish    fmt.Println("All tasks completed")}

Here we use sync.WaitGroup to wait for all workers to finish. The channel with a buffer of numJobs allows sending all tasks without blocking while the workers haven't picked them up yet.

Dynamic Task Addition

In real applications, tasks arrive asynchronously — for example, via HTTP. Here is an example with a handler:

package main

import (    "fmt"    "net/http"    "sync"    "time")

type Task struct {    ID   int    Data string}

var (    taskQueue = make(chan Task, 100)    wg        sync.WaitGroup)

func worker(id int) {    defer wg.Done()    for task := range taskQueue {        fmt.Printf("[Worker %d] Processing: %s (ID=%d)", id, task.Data, task.ID)        time.Sleep(1 * time.Second)    }}

func handler(w http.ResponseWriter, r *http.Request) {    // get task from request (simplified)    task := Task{ID: time.Now().Nanosecond(), Data: "New task"}    taskQueue <- task    w.W