Go: Go Select and Concurrency Patterns
selectis the Swiss Army knife of concurrent programming in Go—it listens to multiple channels simultaneously and processes whichever one becomes ready first. It acts as a "smart switch" connecting goroutines.
If we think of a channel as a phone line connecting goroutines, then select is the switchboard operator—it listens to all calls at once and answers whichever one rings. In this lesson, you'll master all the core patterns of select.
1. You will learn
- Syntax and Behavior of
selectMultiplexing - Random selection (multiple cases ready at the same time)
time.AfterTimeout Controldefault: non-blocking operationfor-selectحلقة exit (done channel)- Distribute tasks using the fan-out mode
- Aggregation results in fan-in mode
- Build a data processing pipeline using the pipeline model
2. The True Story of a Quantitative Trading Engineer
(1) Pain point: Polling three data sources causes the CPU to run at full capacity
Bob is a واجهة خلفية engineer on the quantitative trading team. He needs to monitor three stock data sources simultaneously:
"Our strategy requires retrieving real-time quotes from three exchanges simultaneously: the NYSE, NASDAQ, and LSE. The previous Java implementation used a single خيط to poll three WebSockets, with a 10-millisecond busy wait each time, consuming 30% of the CPU—my boss said I was using more power than the trading system itself."
He opened the current code:
GO
// Bad code: busy polling
func pollDataSources() {
for {
// Poll every 10ms, wasting CPU
if data1 := pollNYSE(); data1 != nil {
process(data1)
}
if data2 := pollNASDAQ(); data2 != nil {
process(data2)
}
if data3 := pollLSE(); data3 != nil {
process(data3)
}
time.Sleep(10 * time.Millisecond)
}
}
Three issues: (1) A 10-millisecond polling interval wastes CPU resources; (2) Data arrival and processing are not synchronized; (3) Polling delays of 0–10 milliseconds are unpredictable.
(2) Go solution: use select to listen to three channels
GO
// market_data.go
package main
import (
"fmt"
"math/rand"
"time"
)
type Quote struct {
Source سلسلة
Symbol سلسلة
Price float64
}
func simulateExchange(name سلسلة, out chan<- Quote) {
symbols := []سلسلة{"AAPL", "GOOGL", "MSFT", "AMZN"}
for {
time.Sleep(time.Duration(50+rand.Intn(200)) * time.Millisecond)
quote := Quote{
Source: name,
Symbol: symbols[rand.Intn(len(symbols))],
Price: 100 + rand.Float64()*200,
}
out <- quote
}
}
func main() {
nyse := make(chan Quote)
nasdaq := make(chan Quote)
lse := make(chan Quote)
go simulateExchange("NYSE", nyse)
go simulateExchange("NASDAQ", nasdaq)
go simulateExchange("LSE", lse)
// select listens to all three channels simultaneously
timeout := time.After(2 * time.Second)
for {
select {
case q := <-nyse:
fmt.Printf("[NYSE] %s: $%.2f\n", q.Symbol, q.Price)
case q := <-nasdaq:
fmt.Printf("[NASDAQ] %s: $%.2f\n", q.Symbol, q.Price)
case q := <-lse:
fmt.Printf("[LSE] %s: $%.2f\n", q.Symbol, q.Price)
case <-timeout:
fmt.Println("Demo ended")
return
}
}
}
Output:
TEXT
📖 Display only
[NASDAQ] AMZN: $198.32
[NYSE] AAPL: $150.45
[LSE] MSFT: $287.10
...
(3) Performance: SELECT Listen vs. Polling
| Method | CPU Usage | Response Latency | Code Complexity |
|---|---|---|---|
| Polling 10 ms | 30% | 0–10 ms | Low |
| Polling 100 ms | 3% | 0–100 ms | Low |
| select Listen | 0% | 0 ms (real-time) | Moderate |
💡 Tip:
select + channel is حدث-driven—the goroutine sleeps (without consuming CPU) when there is no data, and is woken up by the runtime when data arrives. This is what makes Go's concurrency model so efficient.
3. SELECT Basics
(1) select syntax
GO
select {
case v := <-ch1:
// ch1 is ready
case v := <-ch2:
// ch2 is ready
case ch3 <- value:
// Can send to ch3 (ch3 has space or has a receiver)
default:
// All channels are not ready
}
▶ Example: select (random selection)
GO
package main
import (
"fmt"
)
func main() {
ch1 := make(chan سلسلة, 1)
ch2 := make(chan سلسلة, 1)
ch1 <- "from ch1"
ch2 <- "from ch2"
// Both channels are ready; select picks one at random
for i := 0; i < 2; i++ {
select {
case msg := <-ch1:
fmt.Println(msg)
case msg := <-ch2:
fmt.Println(msg)
}
}
}
Output (may vary each time):
TEXT
📖 Display only
from ch1
from ch2
4. Timeout Control
(1) Timeout Mode
GO
package main
import (
"fmt"
"time"
)
func slowOperation() string {
time.Sleep(2 * time.Second)
return "result"
}
func main() {
ch := make(chan string)
go func() {
ch <- slowOperation()
}()
select {
case result := <-ch:
fmt.Println("Success:", result)
case <-time.After(1 * time.Second):
fmt.Println("Timeout! Operation exceeded 1 second")
}
}
Output:
TEXT
📖 Display only
Timeout! Operation exceeded 1 second
▶ Example: Tiered Timeouts (Tiered Waiting)
GO
📖 Display only
package main
import (
"fmt"
"time"
)
func fetchFromCache() سلسلة {
time.Sleep(50 * time.Millisecond)
return "ذاكرة مخبأة-data"
}
func fetchFromDB() سلسلة {
time.Sleep(200 * time.Millisecond)
return "db-data"
}
func fetchFromAPI() سلسلة {
time.Sleep(500 * time.Millisecond)
return "api-data"
}
func main() {
ذاكرة مخبأة := make(chan سلسلة)
db := make(chan سلسلة)
api := make(chan سلسلة)
go func() { ذاكرة مخبأة <- fetchFromCache() }()
go func() { db <- fetchFromDB() }()
go func() { api <- fetchFromAPI() }()
select {
case r := <-ذاكرة مخبأة:
fmt.Println("Cache hit:", r)
case <-time.After(100 * time.Millisecond):
select {
case r := <-db:
fmt.Println("DB returned:", r)
case <-time.After(300 * time.Millisecond):
select {
case r := <-api:
fmt.Println("API returned:", r)
case <-time.After(600 * time.Millisecond):
fmt.Println("All data sources timed out!")
}
}
}
}
💡 Tip:
time.After(d) returns <-chan time.Time and sends the current time after d seconds. Each time select calls time.After, it creates a new timer—if you call it frequently in a loop, be sure to use time.NewTimer and stop it to avoid resource leaks.
5. default: non-blocking operation
(1) Non-blocking send/receive
GO
package main
import (
"fmt"
)
func main() {
ch := make(chan int, 1)
// Non-blocking receive
select {
case v := <-ch:
fmt.Println("Received:", v)
default:
fmt.Println("No data (non-blocking)")
}
// Non-blocking send
ch <- 1
select {
case ch <- 2:
fmt.Println("Send successful")
default:
fmt.Println("Buffer full (non-blocking)")
}
}
Output:
TEXT
📖 Display only
No data (non-blocking)
Buffer full (non-blocking)
▶ Example: Non-blocking channel + polling circuit breaker
GO
package main
import (
"fmt"
"time"
)
func main() {
ch := make(chan int, 3)
go func() {
for i := 1; i <= 10; i++ {
select {
case ch <- i:
// Send successful
default:
fmt.Printf("Buffer full, dropped %d\n", i)
}
time.Sleep(10 * time.Millisecond)
}
close(ch)
}()
// Slow consumer
for v := range ch {
fmt.Printf("Processing: %d\n", v)
time.Sleep(50 * time.Millisecond)
}
}
Output:
TEXT
📖 Display only
Processing: 1
Processing: 2
Processing: 3
Buffer full, dropped 4
Buffer full, dropped 5
Processing: 6
...
🔥 Common Mistake:
default causes the select to return immediately—the default is executed if none of the channels are ready. This is ideal for non-blocking channel operations, but be careful: using default in a for loop can cause a busy loop.
6. for-select Loops and the done Channel
(1) "done channel" exit mode
GO
package main
import (
"fmt"
"time"
)
func worker(done <-chan struct{}) {
for {
select {
case <-done:
fmt.Println("worker exiting")
return
default:
fmt.Println("worker working...")
time.Sleep(200 * time.Millisecond)
}
}
}
func main() {
done := make(chan struct{})
go worker(done)
time.Sleep(1 * time.Second)
close(done)
time.Sleep(100 * time.Millisecond)
fmt.Println("main exiting")
}
▶ Example: Three Ways to Exit a for-select Loop
GO
package main
import (
"fmt"
"time"
)
func main() {
ch := make(chan int)
done := make(chan struct{})
// Producer
go func() {
for i := 1; i <= 5; i++ {
ch <- i
time.Sleep(100 * time.Millisecond)
}
close(ch)
}()
// Consumer (for-select loop)
go func() {
for {
select {
case v, ok := <-ch:
if !ok {
fmt.Println("Style 1: channel closed exit")
close(done)
return
}
fmt.Printf("Processing: %d\n", v)
case <-time.After(1 * time.Second):
fmt.Println("Style 2: timeout exit")
close(done)
return
}
}
}()
<-done
fmt.Println("main exiting")
}
(3) Comparison of for-select Exit Modes
| Mode | Trigger Conditions | Advantages | Disadvantages |
|---|---|---|---|
| close(ch) | Sender closes the channel | Natural termination | Can only be used by the receiver |
| done channel | close(done) signal |
Flexible; can be triggered externally | Requires an additional channel |
| timeout | time.After timeout |
prevents deadlock | hard timeouts aren't flexible enough |
| context | ctx.Done() |
Can send a cancellation signal | Lesson 17: In-Depth |
7. Fan-out / Fan-in Mode
(1) Fan-out: Task distribution
GO
package main
import (
"fmt"
"sync"
)
func fanOut(jobs <-chan int, workers int) []<-chan int {
channels := make([]<-chan int, workers)
for w := 0; w < workers; w++ {
ch := make(chan int, 10)
channels[w] = ch
go func(id int, out chan<- int) {
defer close(out)
for job := range jobs {
result := job * job
fmt.Printf("Worker %d: %d^2 = %d\n", id, job, result)
out <- result
}
}(w, ch)
}
return channels
}
▶ Example: fan-in: result aggregation
GO
package main
import (
"fmt"
"sync"
)
func fanIn(channels ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
for _, ch := range channels {
wg.Add(1)
go func(c <-chan int) {
defer wg.Done()
for v := range c {
out <- v
}
}(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
jobs := make(chan int, 10)
for i := 1; i <= 6; i++ {
jobs <- i
}
close(jobs)
// fan-out: distribute to 3 workers
workers := fanOut(jobs, 3)
// fan-in: aggregate all worker results
results := fanIn(workers...)
// Collect results
sum := 0
count := 0
for r := range results {
sum += r
count++
}
fmt.Printf("Total %d results, sum = %d\n", count, sum)
}
Output:
TEXT
📖 Display only
Worker 2: 1^2 = 1
Worker 0: 3^2 = 9
Worker 1: 2^2 = 4
Worker 1: 5^2 = 25
Worker 0: 4^2 = 16
Worker 2: 6^2 = 36
Total 6 results, sum = 91
(3) fan-out vs fan-in
| Mode | Direction | Purpose |
|---|---|---|
| fan-out | 1 channel → N channels | task distribution, parallel computation |
| fan-in | N channels → 1 channel | result aggregation, log collection |
flowchart LR
subgraph FanOut [Fan-Out]
J[Jobs Channel] --> W1[Worker 1]
J --> W2[Worker 2]
J --> W3[Worker 3]
end
subgraph FanIn [Fan-In]
W1 --> R[Results Channel]
W2 --> R
W3 --> R
end
style J fill:#e1f5fe
style R fill:#fff3e0
8. Complete Example: Aggregating Stock Quotes from Three Data Sources
GO
// market_aggregator.go
package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
// ---------- Data Types ----------
type Trade struct {
Source string
Symbol string
Price float64
Volume int
Time time.Time
}
type AggregatedQuote struct {
Symbol string
AvgPrice float64
TotalVolume int
Sources int
High float64
Low float64
Time time.Time
}
// ---------- Simulated Exchange Data Source ----------
var symbols = []string{"AAPL", "GOOGL", "MSFT", "AMZN", "TSLA", "META"}
func simulateExchange(name string, out chan<- Trade, done <-chan struct{}) {
for {
select {
case <-done:
return
default:
time.Sleep(time.Duration(100+rand.Intn(300)) * time.Millisecond)
trade := Trade{
Source: name,
Symbol: symbols[rand.Intn(len(symbols))],
Price: 100 + rand.Float64()*200,
Volume: rand.Intn(1000) + 100,
Time: time.Now(),
}
select {
case out <- trade:
case <-done:
return
}
}
}
}
// ---------- Aggregator ----------
type Aggregator struct {
trades chan Trade
done chan struct{}
wg sync.WaitGroup
}
func NewAggregator() *Aggregator {
return &Aggregator{
trades: make(chan Trade, 100),
done: make(chan struct{}),
}
}
func (a *Aggregator) AddSource(name string) {
a.wg.Add(1)
go func() {
defer a.wg.Done()
simulateExchange(name, a.trades, a.done)
}()
fmt.Printf("Added data source: %s\n", name)
}
func (a *Aggregator) Start(interval time.Duration, callback func(map[string]*AggregatedQuote)) {
// Aggregation window
window := make(map[string][]Trade)
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case trade := <-a.trades:
window[trade.Symbol] = append(window[trade.Symbol], trade)
case <-ticker.C:
// Time window reached; compute aggregated quotes
quotes := make(map[string]*AggregatedQuote)
for symbol, trades := range window {
if len(trades) == 0 {
continue
}
var sumPrice float64
var totalVol int
high := trades[0].Price
low := trades[0].Price
for _, t := range trades {
sumPrice += t.Price * float64(t.Volume)
totalVol += t.Volume
if t.Price > high {
high = t.Price
}
if t.Price < low {
low = t.Price
}
}
sources := make(map[string]bool)
for _, t := range trades {
sources[t.Source] = true
}
quotes[symbol] = &AggregatedQuote{
Symbol: symbol,
AvgPrice: sumPrice / float64(totalVol),
TotalVolume: totalVol,
Sources: len(sources),
High: high,
Low: low,
Time: time.Now(),
}
}
callback(quotes)
window = make(map[string][]Trade) // Reset window
case <-a.done:
return
}
}
}
func (a *Aggregator) Stop() {
close(a.done)
a.wg.Wait()
}
// ---------- Main Function ----------
func main() {
aggregator := NewAggregator()
// Add 3 data sources
aggregator.AddSource("NYSE")
aggregator.AddSource("NASDAQ")
aggregator.AddSource("LSE")
fmt.Println("\nStarting aggregation (output every 2 seconds)...\n")
// Start aggregation with 2-second time window
done := make(chan struct{})
go func() {
aggregator.Start(2*time.Second, func(quotes map[string]*AggregatedQuote) {
fmt.Printf("=== Aggregation Report %s ===\n", time.Now().Format("15:04:05"))
for _, q := range quotes {
fmt.Printf("%-6s | $%.2f (avg) | vol=%d | src=%d | H=%.2f L=%.2f\n",
q.Symbol, q.AvgPrice, q.TotalVolume, q.Sources, q.High, q.Low)
}
fmt.Println()
})
close(done)
}()
// Run for 10 seconds then stop
time.Sleep(10 * time.Second)
aggregator.Stop()
<-done
fmt.Println("Quote aggregator stopped")
}
Expected Output:
TEXT
📖 Display only
Added data source: NYSE
Added data source: NASDAQ
Added data source: LSE
Starting aggregation (output every 2 seconds)...
=== Aggregation Report 10:00:02 ===
AAPL | $152.34 (avg) | vol=2341 | src=3 | H=165.20 L=142.10
GOOGL | $178.90 (avg) | vol=1567 | src=2 | H=185.00 L=172.30
MSFT | $295.40 (avg) | vol=890 | src=3 | H=301.20 L=288.50
=== Aggregation Report 10:00:04 ===
...
Quote aggregator stopped
🔥 Common Mistake: In the
select block, both case <-done: and case trade := <-a.trades: must be subscribed to. If you only subscribe to the trades channel, the Stop signal will not be delivered. Every for-select loop must include an exit condition.
❓ FAQ
Q What should I do if multiple channels are ready at the same time?
A Go will pseudo-randomly select one case to execute. This is part of the language specification—to prevent developers from relying on channel priorities. If you need priorities, use two nested
select statements or a default clause combined with manual checks.Q How does
time.After handle timeouts?A
time.After(d) returns a channel that receives a value after d units of time. In a select statement, use case <- time.After(d): as the timeout branch. Note: Each select call to time.After creates a new timer; in a loop, it is recommended to reuse it with time.NewTimer.Q What does
default do in a select statement?A
default makes the select statement non-blocking—it executes the default block immediately when none of the channels are ready. Suitable for: attempting to send/receive without blocking, calculating failure rates, and circuit breaking/degradation. Note: Using default in a for loop creates a busy loop; you typically need to add time.Sleep or limit the frequency.Q How do I exit a for-select loop using a done channel?
A Create
done := make(chan struct{}), listen for case <-done: in the select block, and call close(done) when you want to exit. All goroutines listening to done will receive the exit signal simultaneously. close is preferable to sending a value—it can be received multiple times.Q What are fan-in and fan-out?
A Fan-out = 1 input channel distributed to N worker goroutines (parallel processing); fan-in = N channels aggregated into 1 output channel (merging results). The combination of the two is a classic pattern in Go parallel computing.
Q Can
select be used for sending?A Yes.
case ch <- value: is also a valid select case. It becomes ready when the channel has space (buffered) or a receiver (unbuffered). This feature is used for rate-limiting and semaphore modes.Q What happens with an empty
select {}?A
select {} will block indefinitely—because there are no cases and no default. This can serve as a signal to "wait indefinitely." However, in production code, you should ensure there is an exit path; otherwise, goroutines will leak.Q Can a
break statement in a for-select loop exit the loop?A No. A
break statement within a select block only exits the select block; it does not exit the for loop. To exit a for-select loop, use a label followed by break label, return, or the done channel.📖 Summary
- select listens to multiple channels simultaneously and processes the one that becomes ready first
- When multiple cases are ready at the same time, select one at random
time.Afteris used for timeout control, andtime.NewTimeris reused in loopsdefaultmakes the select operation non-blocking, which is suitable for circuit-breaking or retry operationsdone channelis the standard way to exit a for-select loop- "Fan-out" distributes tasks, "fan-in" aggregates results, and the two combine to form a parallel pipeline
- Every for-select must include an exit condition
📝 Exercises
-
Basic Problem (Difficulty ⭐): Write a program that starts three goroutines, each of which sends an integer to its own channel. Use
selectto receive and print them simultaneously. Add a timeout branch to exit after 3 seconds. -
Advanced Problem (Difficulty ⭐⭐): Implement a timeout cache:
func FetchWithCache(key string, cache map[string]string) string. First, check the cache channel (10 ms timeout); if there is no hit, query the "simulated database" (500 ms) and populate the cache with the result. You must useselectcombined withtime.Afterto implement tiered timeouts. -
Challenge (Difficulty: ⭐⭐⭐): Implement a log aggregation system: Three log sources (goroutines) each generate logs of different levels (INFO/WARN/ERROR), which are aggregated into a single channel using a fan-in operation, and then processed by level using
select—ERROR logs trigger an immediate alert (print), WARN logs are buffered (flushed in batches of 5), and INFO logs are written in batches (flushed in batches of 10). Requirements: graceful exit via timeout flush anddonechannel.
