A dispatcher is the final stage of a concurrent request pipeline. It receives requests from an upstream producer or throttler, launches a fixed number of worker goroutines, and emits each completed Result on an output channel. Callers can begin receiving results immediately rather than waiting for every request to finish. section 6.6.2 section 6.9.1
A practical dispatcher has four responsibilities:
- Start workers according to a concurrency setting. Each worker ranges over the input channel and handles requests as they arrive. section 6.9.1
- Invoke the configured send function, then forward each Result to the output channel. section 6.9.1
- Track all workers with a WaitGroup so completion can be detected safely. section 6.9.1
- Close the output channel only after every worker exits, signaling that no further results can arrive. section 6.9 section 6.9.1
Dispatching concurrent HTTP requests
Create an output channel, add the configured worker count to a WaitGroup, and start that many goroutines. Each goroutine reads requests until the input channel closes, sends each request, and writes its Result. A separate monitor waits for the group and then closes the output channel. Returning that channel immediately lets an iterator yield results as workers finish, while preserving a synchronous-looking SendN API. section 6.9.1 section 6.9.2
Place the dispatcher after optional throttling: the producer creates requests, the throttler limits their passage when enabled, and the dispatcher performs concurrent sends. This arrangement can measure server behavior under concurrent load, but it needs cancellation support when consumers stop early; otherwise workers may continue in the background. section 6.9 section 6.9.2
Listing 6.16 Implementing the dispatcher (hit/pipe.go)
package hit
import (
. . .
"sync"
)
. . .
func dispatch(
in <-chan *http.Request, #1
concurrency int, #2
send SendFunc, #3
) <-chan Result { #4
out := make(chan Result)
var wg sync.WaitGroup #5
wg.Add(concurrency) #6
for range concurrency { #7
go func() {
defer wg.Done() #8
for req := range in { #9
out <- send(req) #10
}
}()
}
go func() {
wg.Wait() #11
close(out) #11
}()
return out #12
}
The implementation shows the core structure: workers range over input, publish results, and a WaitGroup-backed monitor closes output after all workers finish. section 6.9.1
