go 46 lines · 7 steps

A graceful worker pool in Go

A worker pool that processes jobs from a channel and shuts every goroutine down cleanly with a broadcast signal.

Explained by highlit

Intermediate concurrency goroutines channels graceful-shutdown synchronization

1package worker
2 
3import (
4	"fmt"
5	"sync"
6	"time"
7)
8 
9type Pool struct {
10	done chan struct{}
11	once sync.Once
12	wg   sync.WaitGroup
13}
14 
15func NewPool() *Pool {
16	return &Pool{done: make(chan struct{})}
17}
18 
19func (p *Pool) Spawn(id int, jobs <-chan string) {
20	p.wg.Add(1)
21	go func() {
22		defer p.wg.Done()
23		for {
24			select {
25			case <-p.done:
26				fmt.Printf("worker %d: shutting down\n", id)
27				return
28			case job, ok := <-jobs:
29				if !ok {
30					return
31				}
32				select {
33				case <-p.done:
34					return
35				case <-time.After(50 * time.Millisecond):
36					fmt.Printf("worker %d: processed %s\n", id, job)
37				}
38			}
39		}
40	}()
41}
42 
43func (p *Pool) Shutdown() {
44	p.once.Do(func() { close(p.done) })
45	p.wg.Wait()
46}

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1Closing a channel broadcasts to every receiver at once, which makes it the idiomatic Go shutdown signal.
2A sync.Once guards close so Shutdown is safe to call repeatedly without panicking on a double close.
3Pairing a WaitGroup with the done channel lets shutdown both signal and block until all workers actually exit.

Related explainers

rust

use axum::{extract::{Path, State}, http::StatusCode, Json};
use dashmap::DashMap;
use serde::Serialize;
use std::sync::Arc;

Request coalescing in an Axum handler

caching concurrency request-coalescing

Advanced 8 steps

package handlers
 
import (
	"net/http"

Serving embedded static meta files in Gin

embedding static assets http caching

Intermediate 6 steps

package config
 
import (
	"fmt"

A thread-safe config singleton in Go

singleton concurrency environment-variables

Intermediate 7 steps

javascript

const express = require('express');
const app = express();
 
app.get('/health', (req, res) => res.json({ status: 'ok' }));

Graceful shutdown in an Express server

graceful-shutdown signal-handling connection-tracking

Advanced 9 steps

package scheduler
 
import (
	"container/heap"

A priority job queue with Go's container/heap

priority-queue heap interfaces

Intermediate 9 steps

func UploadChunk(c *gin.Context) {
	uploadID := c.Param("uploadID")
	if !validUploadID.MatchString(uploadID) {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid upload id"})