goroutine-patterns.md
Go Goroutine Patterns
A practical tour of the concurrency patterns you actually reach for in Go, each with a runnable example and a short note on when it fits.
1. WaitGroup: fan-out, wait for all
Launch N independent goroutines and block until every one has finished. Add before the goroutine starts, defer wg.Done() inside it, wg.Wait() in the caller.
func main() {
var wg sync.WaitGroup
for i := 1; i <= 5; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
fmt.Printf("worker %d done\n", id)
}(i)
}
wg.Wait()
fmt.Println("all workers finished")
}Use when you need "do these things in parallel, continue when all are done" and don't need results streamed back. No channel bookkeeping, no leaks. It's the shape of almost every "load the dashboard" handler: the page wants a user profile, a list of recent orders, and a feature-flag lookup, all from different services, so you fire off three goroutines writing into their own pre-allocated slots and wait before rendering. Response time becomes the slowest call instead of the sum of all three, which on a page with six widgets is the difference between 900ms and 200ms.
2. Worker pool: bounded parallelism
A fixed number of goroutines consume from a jobs channel and push to a results channel. This caps concurrency so you don't spawn 10,000 goroutines against a database.
func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for j := range jobs { // exits when jobs is closed
results <- j * j
}
}
func main() {
jobs := make(chan int, 100)
results := make(chan int, 100)
var wg sync.WaitGroup
for w := 1; w <= 3; w++ { // only 3 concurrent workers
wg.Add(1)
go worker(w, jobs, results, &wg)
}
for j := 1; j <= 9; j++ {
jobs <- j
}
close(jobs) // signals workers to exit their range loop
wg.Wait()
close(results)
for r := range results {
fmt.Println(r)
}
}Use when work items are numerous but the resource behind them (CPU, DB, API rate limit) is not. The producer closes jobs; a separate waiter closes results. Image and video services live on this: a single upload might need six thumbnail sizes, an EXIF strip, and a perceptual hash, and a goroutine per job will allocate more decode buffers than the box has RAM before it finishes the queue. A pool sized to runtime.NumCPU() keeps memory flat and throughput predictable. Anything talking to Postgres has the same constraint, except there the ceiling is max_connections rather than memory.
3. Generator / pipeline stage
A function that owns a goroutine, returns a channel, and closes it when done. Stages chain together, each one transforming the previous stage's output.
func gen(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out) // the sender always closes
for _, n := range nums {
out <- n
}
}()
return out
}
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
out <- n * n
}
}()
return out
}
func main() {
for v := range square(gen(1, 2, 3, 4)) {
fmt.Println(v) // 1 4 9 16
}
}Use when data flows through successive transformations. Key rule: the goroutine that sends on a channel is the one that closes it. Log and event processing is built exactly this way: read lines from a file or a Kafka partition, parse them into structs, drop the noise, enrich with GeoIP, batch for the sink. Each stage is a small testable function that takes a channel and returns a channel, and because every stage runs in its own goroutine, a slow enrichment step never blocks the read. Tools in the Logstash and Vector family are conceptually this, and Go's streaming JSON decoders drop straight into the front of one.
4. Fan-in: merge many channels into one
Multiple producers, one consumer. A WaitGroup tracks the copiers so the merged channel is closed exactly once, after every input drains.
func merge(chans ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
for _, c := range chans {
wg.Add(1)
go func(c <-chan int) {
defer wg.Done()
for v := range c {
out <- v
}
}(c)
}
go func() {
wg.Wait()
close(out)
}()
return out
}Use when you fanned work out to several stages and want a single stream back. Output order is non-deterministic by design, which is usually fine: a Prometheus-style scraper hits fifty targets in parallel, each on its own goroutine and channel, and the merge collapses every sample into one stream for the writer to batch. Nobody cares which target answered first, only that the merged channel closes once when they're all finished. Sharded search works the same way: query every shard concurrently, merge, then rank whatever came back.
5. Context cancellation
context.Context is the standard way to tell a goroutine to stop. Every blocking operation should also select on ctx.Done().
func work(ctx context.Context, out chan<- int) {
for i := 0; ; i++ {
select {
case <-ctx.Done():
fmt.Println("stopping:", ctx.Err())
return
case out <- i:
}
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
out := make(chan int)
go work(ctx, out)
for v := range out {
if v > 3 {
break
}
fmt.Println(v)
}
<-ctx.Done()
}Use when a goroutine's lifetime is tied to a request, a timeout, or a parent operation. Pass ctx as the first parameter; never store it in a struct. This is the backbone of Go's own net/http: every incoming request carries a context that gets cancelled when the client disconnects, and any database driver, HTTP client, or gRPC call that accepts it will abort mid-flight. That's why a user hitting Escape on a slow search page can actually free up the query behind it instead of leaving it running for another thirty seconds. Kubernetes controllers and the AWS SDK thread context through for the same reason.
6. Done channel (cancellation without context)
The pre-context idiom, still useful for internal plumbing. Closing a channel broadcasts to every receiver at once.
func producer(done <-chan struct{}) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for i := 0; ; i++ {
select {
case <-done:
return
case out <- i:
}
}
}()
return out
}
func main() {
done := make(chan struct{})
stream := producer(done)
for v := range stream {
if v == 5 {
close(done) // broadcast: every receiver unblocks
break
}
}
}Use when you want a lightweight stop signal inside a package and don't need deadlines or values carried along. It fits long-lived background workers: a cache warmer, a metrics flusher, a file watcher, all started at boot and all needing to exit cleanly on SIGTERM. Close one stopCh in the shutdown handler and every one of them unblocks at the same instant. Kubernetes' client-go informers still use this signature, passing a stopCh down through the whole watch machinery, because the pattern predates context being everywhere.
7. Select with timeout
select picks whichever case is ready first; time.After gives you a deadline on any single operation.
func main() {
c := make(chan string)
go func() {
time.Sleep(2 * time.Second)
c <- "result"
}()
select {
case res := <-c:
fmt.Println(res)
case <-time.After(1 * time.Second):
fmt.Println("timed out")
}
}Use when an operation must not block forever, which in practice means any call to a third-party service you don't control. Payment processors, geocoding APIs and partner webhooks all have that one bad day where p99 goes to forty seconds, and without a deadline your goroutines pile up until the process falls over. It's also the core of health checks and circuit breakers: give a dependency one second to prove it's alive, and serve degraded if it can't. Note the goroutine above leaks on timeout because nobody receives; use a buffered channel (make(chan string, 1)) so the send always succeeds.
8. Semaphore: limit concurrency with a buffered channel
A buffered channel used as a counting semaphore. Cheaper than a full worker pool when the work is already enumerated.
func main() {
sem := make(chan struct{}, 3) // at most 3 in flight
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
sem <- struct{}{} // acquire
defer func() { <-sem }() // release
fmt.Println("processing", i)
time.Sleep(100 * time.Millisecond)
}(i)
}
wg.Wait()
}Use when you have a slice of tasks and just want to throttle how many run at once. Crawlers and bulk API clients are the obvious case: ten thousand URLs to fetch against an endpoint that allows ten concurrent connections before it starts handing back 429s, and a ten-slot channel is the entire rate limiter. The same trick caps concurrent uploads in an S3 migration script, or limits how many git clone subprocesses a CI runner spawns before it saturates the network interface.
9. errgroup: fan-out with error propagation
golang.org/x/sync/errgroup is a WaitGroup that also collects the first error and cancels the shared context.
func main() {
g, ctx := errgroup.WithContext(context.Background())
urls := []string{"https://a.example", "https://b.example"}
results := make([]string, len(urls))
for i, url := range urls {
i, url := i, url
g.Go(func() error {
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return err
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err // cancels ctx for the siblings
}
defer resp.Body.Close()
results[i] = resp.Status
return nil
})
}
if err := g.Wait(); err != nil {
fmt.Println("failed:", err)
}
}Use when parallel tasks can fail and one failure should abort the rest. This is the default choice for concurrent I/O in modern Go. Picture a checkout page calling inventory, pricing and fraud detection at the same time: if fraud returns an error, continuing to wait on pricing is pure waste, and the cancelled context stops the siblings so you return in milliseconds rather than at the timeout. It's equally the standard way to run parallel migrations, build test fixtures, or issue a batch of gRPC calls where partial success isn't a meaningful result.
10. Mutex: protecting shared state
Channels pass ownership; mutexes protect state that must stay in place. Use whichever matches the problem, not whichever is more idiomatic-sounding.
type Counter struct {
mu sync.Mutex
n map[string]int
}
func (c *Counter) Inc(key string) {
c.mu.Lock()
defer c.mu.Unlock()
c.n[key]++
}
func (c *Counter) Get(key string) int {
c.mu.Lock()
defer c.mu.Unlock()
return c.n[key]
}Use when several goroutines read and write the same struct or map. For read-heavy data use sync.RWMutex; for a single value use sync/atomic. In-memory caches, connection pools and rate-limiter state all end up here, because they're shaped like a map that every request handler touches at once and moving ownership between goroutines would be absurd. The RWMutex variant is what sits behind most config hot-reload implementations: thousands of readers a second against the current config, one writer when the file changes on disk.
11. sync.Once: exactly-once initialization
Safe lazy init, even under concurrent access.
var (
once sync.Once
instance *Config
)
func Get() *Config {
once.Do(func() {
instance = loadConfig()
})
return instance
}Use when an expensive singleton must be built on first use. Do blocks other callers until the first invocation returns, so the second goroutine through the door gets a fully constructed value rather than a half-initialized one. Database handles, gRPC client connections, compiled regexes and parsed templates all belong here: you don't want to open a connection pool at package init, because tests and CLI subcommands that never touch the database shouldn't pay for it. It's also the correct answer to the double-checked locking people try to hand-roll first.
12. Result struct over a channel
Channels can't carry a second error return, so bundle value and error together.
type Result struct {
Value string
Err error
}
func fetch(ctx context.Context, urls []string) <-chan Result {
out := make(chan Result)
var wg sync.WaitGroup
for _, u := range urls {
wg.Add(1)
go func(u string) {
defer wg.Done()
body, err := get(ctx, u)
select {
case out <- Result{Value: body, Err: err}:
case <-ctx.Done():
}
}(u)
}
go func() { wg.Wait(); close(out) }()
return out
}Use when you want results as they arrive rather than all-or-nothing. The select on the send prevents a goroutine leak if the consumer walks away early. This is what fan-out to flaky dependencies looks like when partial results still have value: a price comparison page queries eight airline APIs, two time out, six answer, and you render the six instead of failing the page. Streaming as results land also lets you paint the UI progressively rather than waiting on the slowest source. Search aggregators, multi-region status dashboards, and "check this domain against N blocklists" tools all converge on this shape.
Rules of thumb
- The sender closes the channel, never the receiver, and only once.
- Never start a goroutine without knowing how it stops. Every launch needs a
ctx, a done channel, or a closed input channel. - A
nilchannel blocks forever. Handy insideselectto disable a case dynamically. - Unbuffered = synchronization; buffered = decoupling. Pick deliberately.
- Run
go test -race. The race detector catches what review misses. - Don't over-parallelize. Goroutines are cheap, but the resources behind them are not.
点赞
收藏
点踩
分享