Building an Asynchronous Queue in Go: Design, Demo, and Benchmark

Why an asynchronous queue?

In high‑concurrency systems, a sudden surge of requests can exceed the system's processing capacity, causing overload or database crashes. The article uses a flash‑sale scenario where only 300 items are available but millions of users attempt to purchase simultaneously, illustrating the need to limit concurrent access and discard excess requests to keep the service stable.

Logical architecture

The solution follows a classic producer‑consumer model. A configurable queue holds incoming jobs, and a pool of worker goroutines consumes them. The design is similar to the author’s earlier EasyRedis series, emphasizing reusability.

Demo with Gin

A single‑machine demo uses the Gin framework. Three HTTP endpoints are provided: /cache – returns statistics such as remaining stock, sold‑out users, queue overflow count, successful purchases, and total calls. /add – adds 20 units to the stock. /ping – represents a purchase request. It first checks stock, pushes a job to the queue, and blocks until the job finishes. If the queue is full, the request is rejected with a “too many users, retry” message.

The demo stores stock locally; in a real system the stock would reside in Redis or MySQL with a distributed lock to avoid overselling.

package main

import (
    "net/http"
    "strconv"
    "sync/atomic"
    "time"

    "github.com/gin-gonic/gin"
    "github.com/gofish2020/easyqueue"
)

// Create a queue with 1 partition, capacity 100, and 1 worker
var g_Queue = easyqueue.CreateEasyQueue(easyqueue.SetQueueParttion(1), easyqueue.SetQueueCapacity(100), easyqueue.SetWorkerNum(1))

var cacheNum atomic.Int64
var errCount atomic.Int64
var sucessCount atomic.Int64
var selloutCount atomic.Int64
var pingCount atomic.Int64

func main() {
    cacheNum.Store(20)
    r := gin.Default()

    r.GET("/cache", func(c *gin.Context) {
        c.JSON(http.StatusOK, gin.H{
            "商品剩余数量":      strconv.Itoa(int(cacheNum.Load())),
            "售罄->没买到的用户": strconv.Itoa(int(selloutCount.Load())),
            "队列满->丢弃的用户数": strconv.Itoa(int(errCount.Load())),
            "成功->抢购的用户数": strconv.Itoa(int(sucessCount.Load())),
            "抢购链接->总调用次数": strconv.Itoa(int(pingCount.Load())),
        })
    })

    r.GET("/add", func(c *gin.Context) {
        cacheNum.Add(20)
        c.JSON(http.StatusOK, gin.H{"message": "success"})
    })

    r.GET("/ping", func(c *gin.Context) {
        pingCount.Add(1)
        if cacheNum.Load() == 0 {
            selloutCount.Add(1)
            c.JSON(http.StatusOK, gin.H{"message": "商品售罄"})
            return
        }
        waitJob := g_Queue.Push(func() {
            if cacheNum.Load() == 0 {
                selloutCount.Add(1)
                c.JSON(http.StatusOK, gin.H{"message": "商品售罄"})
            } else {
                cacheNum.Add(-1)
                time.Sleep(500 * time.Millisecond)
                sucessCount.Add(1)
                c.JSON(http.StatusOK, gin.H{"message": "抢购成功"})
            }
        })
        <-waitJob.Done()
        if waitJob.Err() == easyqueue.ErrOverFlow {
            errCount.Add(1)
            c.JSON(http.StatusOK, gin.H{"message": "抢购人数过多，请重试..."})
        }
    })
    r.Run()
}

Benchmark with ab

The article suggests using the ApacheBench tool (ab) that ships with macOS. Example command: ab -c 100 -n 10000 http://127.0.0.1:8080/ping ‘-c 100’ sets 100 concurrent requests, ‘-n 10000’ sends 10,000 total requests. After the test, visit http://127.0.0.1:8080/cache to view statistics.

Code logic overview

The EasyQueue struct wraps a Queue and a worker group. Its only public method is Push, which creates a job and enqueues it.

type EasyQueue struct {
    config Config
    queue  Queue
    wg     *workerGroup
}

func (eq *EasyQueue) Push(fn func()) WaitJob {
    job := newJob(fn)
    eq.queue.Push(job)
    return job
}

func CreateEasyQueue(cfs ...configFunc) *EasyQueue {
    conf := Config{}
    for _, cf := range cfs { cf(&conf) }
    eq := EasyQueue{config: conf}
    eq.queue = createMultiJobQueue(conf.QueuePartition, conf.QueueCapacity)
    eq.wg = createWorkerMange(eq.queue, conf.WorkersNum)
    return &eq
}

createMultiJobQueue

This function initializes a slice of *jobQueue based on the configured partition count and capacity. The Push method distributes jobs across partitions using round‑robin modulo, ensuring load balancing.

type multiJobQueue struct {
    queues   []*jobQueue
    parition int
    pushIdx  *idGenerator
    popIdx   *idGenerator
}

func createMultiJobQueue(partition, queueCapacity int) *multiJobQueue {
    if partition < 1 { panic("partition must bigger than 0") }
    multi := &multiJobQueue{parition: partition, pushIdx: newIDGenerator(), popIdx: newIDGenerator()}
    for i := 0; i < partition; i++ {
        multi.queues = append(multi.queues, createJobQueue(queueCapacity))
    }
    return multi
}

func (mjq *multiJobQueue) Push(jb JobInterface) {
    mjq.queues[mjq.pushIdx.Next()%uint64(mjq.parition)].Push(jb)
}

createWorkerMange

Based on the configured worker number, this function creates that many workers, starts them, and stores them in a slice. The Adjust method can dynamically increase or decrease the worker count while holding a mutex to avoid race conditions.

type workerGroup struct {
    mu         sync.Mutex
    workers    []*worker
    workersNum int
    queue      Queue
}

func createWorkerMange(queue Queue, workerNum int) *workerGroup {
    mange := workerGroup{workersNum: workerNum, queue: queue}
    for i := 0; i < workerNum; i++ {
        worker := createWorker(queue)
        worker.Run()
        mange.workers = append(mange.workers, worker)
    }
    return &mange
}

func (wm *workerGroup) Adjust(workerNum int) {
    if workerNum == wm.workersNum { return }
    wm.mu.Lock(); defer wm.mu.Unlock()
    if workerNum > wm.workersNum {
        for i := wm.workersNum; i < workerNum; i++ {
            w := createWorker(wm.queue); w.Run(); wm.workers = append(wm.workers, w)
        }
    } else {
        for i := workerNum; i < wm.workersNum; i++ { wm.workers[i].Stop() }
        wm.workers = wm.workers[:wm.workersNum-workerNum]
    }
    wm.workersNum = workerNum
}

Consumption logic

Each worker repeatedly calls w.queue.PopTimeout(1 * time.Millisecond). If a job is returned, job.DoJob() executes it. The timeout prevents a worker from blocking indefinitely when its assigned queue is empty, allowing the scheduler to rotate to other queues and avoid deadlock.

type worker struct {
    queue  Queue
    closed atomic.Int32
}

func (w *worker) Run() {
    w.closed.Store(0)
    go func() {
        for {
            if w.closed.Load() > 0 { break }
            job := w.queue.PopTimeout(1 * time.Millisecond)
            if job != nil { job.DoJob() }
        }
    }()
}

func (w *worker) Stop() { w.closed.Store(1) }

Extended thinking

The presented design assumes that excess requests can be dropped. For systems where every request matters (e.g., posting on a forum with monetary rewards), the article suggests combining rate limiting with asynchronous processing to decouple heavy operations such as database writes or third‑party moderation, thereby improving user experience.