Help us shape the future of Go! Take the 2026 Go Developer Survey

Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

flow

package module

Go to main page

Versions in this module

v0

v0.5.0

Sep 22, 2026

Changes in this version

+ type Counter interface

+ Add func(n int64)

+ type Histogram interface

+ Record func(d time.Duration)

+ type Meter interface

+ Counter func(name string) Counter

+ Histogram func(name string) Histogram

type Option

+ func WithMeter(m Meter) Option

v0.4.0

Sep 16, 2026

Changes in this version

type Report

+ PerWorker []WorkerStat

+ ProcessLatency StageLatency

+ type StageLatency struct

+ Max time.Duration

+ P50 time.Duration

+ P95 time.Duration

+ P99 time.Duration

+ type WorkerStat struct

+ Blocked time.Duration

+ Busy time.Duration

+ Idle time.Duration

+ Processed int64

v0.3.2

Sep 16, 2026

v0.3.1

Sep 14, 2026

v0.3.0

Sep 10, 2026

Changes in this version

type Option

+ func Observe(dst *Report) Option

+ type Report struct

+ Advice string

+ Bottleneck Stage

+ Consumed int64

+ ConsumerBusy time.Duration

+ Emitted int64

+ Processed int64

+ Produced int64

+ ProducerBlocked time.Duration

+ Wall time.Duration

+ WorkerBlocked time.Duration

+ WorkerBusy time.Duration

+ WorkerIdle time.Duration

+ Workers int

+ func (r *Report) String() string

+ func (r *Report) Util() float64

+ type Stage int

+ const StageConsumer

+ const StageProcessor

+ const StageProducer

+ func (s Stage) String() string

v0.2.0

Sep 10, 2026

Changes in this version

+ func RunPartitioned[I, O any, K comparable](ctx context.Context, p Producer[I], proc Processor[I, O], c Consumer[O], ...) error

type Option

+ func Prefetch(n int) Option

v0.1.0

Sep 6, 2026

Changes in this version

+ func Run[I, O any](ctx context.Context, p Producer[I], proc Processor[I, O], c Consumer[O], ...) error

+ type Consumer interface

+ Consume func(ctx context.Context, v T) error

+ func Each[T any](fn func(T) error) Consumer[T]

+ func Into[T any](dst *[]T) Consumer[T]

+ type ConsumerFunc func(ctx context.Context, v T) error

+ func (f ConsumerFunc[T]) Consume(ctx context.Context, v T) error

+ type Option func(*config)

+ func Ordered() Option

+ func Workers(n int) Option

+ type Processor interface

+ Process func(ctx context.Context, in I, emit func(O) error) error

+ func Filter[T any](pred func(T) bool) Processor[T, T]

+ func FlatMap[I, O any](fn func(I) []O) Processor[I, O]

+ func Map[I, O any](fn func(I) O) Processor[I, O]

+ func Then[A, B, C any](p1 Processor[A, B], p2 Processor[B, C]) Processor[A, C]

+ func TryFilter[T any](pred func(T) (bool, error)) Processor[T, T]

+ func TryFlatMap[I, O any](fn func(I) ([]O, error)) Processor[I, O]

+ func TryMap[I, O any](fn func(I) (O, error)) Processor[I, O]

+ type ProcessorFunc func(ctx context.Context, in I, emit func(O) error) error

+ func (f ProcessorFunc[I, O]) Process(ctx context.Context, in I, emit func(O) error) error

+ type Producer interface

+ Produce func(ctx context.Context, emit func(T) error) error

+ func FromSeq[T any](seq iter.Seq[T]) Producer[T]

+ func Slice[T any](xs ...T) Producer[T]

+ type ProducerFunc func(ctx context.Context, emit func(T) error) error

+ func (f ProducerFunc[T]) Produce(ctx context.Context, emit func(T) error) error