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

goroutine

package module

Go to main page

Versions in this module

v1

v1.0.8

Jan 13, 2026

Changes in this version

+ var ErrBackendFailed = errors.New("backend operation failed")

+ var ErrChannelClosed = errors.New("channel is closed")

+ var ErrContextCanceled = errors.New("context canceled")

+ var ErrInvalidBackend = errors.New("invalid backend")

+ var ErrNotImplemented = errors.New("not implemented")

+ var ErrRetryExhausted = errors.New("retry attempts exhausted")

+ var ErrTimeout = errors.New("operation timed out")

+ func MapTo[T, U any](ss *SuperSlice[T], mapper func(index int, item T) U) []U

+ func Null()

+ func RecastToJSON(src, dst interface{})

+ func RecastToJSONBytes(src interface{}, opts *RecastOptions) ([]byte, error)

+ func RecastToJSONString(src interface{}, opts *RecastOptions) (string, error)

+ func RecastToJSONWithOptions(src, dst interface{}, opts *RecastOptions)

+ func RecastToMap(src interface{}, opts *RecastOptions) map[string]interface

+ func RegisterTaskHandler(name string, handler TaskHandler)

+ type BrokerJobData struct

+ Args map[string]interface{}

+ Class string

+ CreatedAt time.Time

+ CronExpr string

+ Delay int64

+ ETA *time.Time

+ Expires *time.Time

+ ExpiresAt *time.Time

+ HandlerName string

+ ID string

+ IsRecurring bool

+ Kwargs map[string]interface{}

+ Priority TaskPriority

+ Queue string

+ RetryPolicy *RetryPolicy

+ Task string

+ Timeout int64

+ type Cache struct

+ func NewCache[K comparable, V any]() *Cache[K, V]

+ func (c *Cache[K, V]) Cleanup()

+ func (c *Cache[K, V]) Clear()

+ func (c *Cache[K, V]) Delete(key K)

+ func (c *Cache[K, V]) Get(key K) (*CacheEntry[V], bool)

+ func (c *Cache[K, V]) Set(key K, value V, ttl time.Duration)

+ type CacheControl struct

+ MaxAge time.Duration

+ NoCache bool

+ StaleWhileRevalidate bool

+ func DefaultCacheControl() *CacheControl

+ type CacheEntry struct

+ CachedAt time.Time

+ ExpiresAt time.Time

+ Valid bool

+ Value T

+ func (ce *CacheEntry[T]) IsExpired() bool

+ func (ce *CacheEntry[T]) IsStale() bool

+ type CachedGroup struct

+ func NewCachedGroup() *CachedGroup

+ func (cg *CachedGroup) AssignWithCache(key string, result *any, fn func() any, control *CacheControl)

+ func (cg *CachedGroup) ClearCache()

+ func (cg *CachedGroup) GetCache() *Cache[string, any]

+ func (cg *CachedGroup) Resolve()

+ func (cg *CachedGroup) ResolveWithTimeout(timeout time.Duration) bool

+ type CircuitBreaker struct

+ func NewCircuitBreaker(threshold int) *CircuitBreaker

+ func (cb *CircuitBreaker) Allow() bool

+ func (cb *CircuitBreaker) RecordFailure()

+ func (cb *CircuitBreaker) RecordSuccess()

+ func (cb *CircuitBreaker) Reset()

+ func (cb *CircuitBreaker) State() CircuitState

+ type CircuitState int

+ const CircuitClosed

+ const CircuitHalfOpen

+ const CircuitOpen

+ type ConcurrencyConfig struct

+ CircuitBreakerThreshold int

+ EnableCircuitBreaker bool

+ MaxConcurrency int

+ RateLimit int

+ RetryAttempts int

+ RetryDelay time.Duration

+ Timeout time.Duration

+ func DefaultConcurrencyConfig() *ConcurrencyConfig

+ type ConcurrencyWrapper struct

+ func NewConcurrencyWrapper[T, R any](config *ConcurrencyConfig) *ConcurrencyWrapper[T, R]

+ func (cw *ConcurrencyWrapper[T, R]) Close()

+ func (cw *ConcurrencyWrapper[T, R]) Process(ctx context.Context, items []T, processor func(T) (R, error)) ([]R, error)

+ type CronSchedule struct

+ func NewCronSchedule(expr string) (*CronSchedule, error)

+ func (cs *CronSchedule) Next(from time.Time) time.Time

+ type DistributedBackend interface

+ Close func() error

+ Receive func(ctx context.Context, topic string) ([]byte, error)

+ Send func(ctx context.Context, topic string, message []byte) error

+ Subscribe func(ctx context.Context, topic string, handler func([]byte) error) error

+ Type func() string

+ type DistributedSafeChannel struct

+ func NewDistributedSafeChannel[T any](backend DistributedBackend, topic string, bufferSize int, ...) *DistributedSafeChannel[T]

+ func (dsc *DistributedSafeChannel[T]) BackendType() string

+ func (dsc *DistributedSafeChannel[T]) ClearCache()

+ func (dsc *DistributedSafeChannel[T]) Close() error

+ func (dsc *DistributedSafeChannel[T]) IsClosed() bool

+ func (dsc *DistributedSafeChannel[T]) Receive(ctx context.Context) (T, error)

+ func (dsc *DistributedSafeChannel[T]) ReceiveWithPreflight(ctx context.Context, key string) (T, error)

+ func (dsc *DistributedSafeChannel[T]) Send(ctx context.Context, value T) error

+ func (dsc *DistributedSafeChannel[T]) SetCacheControl(control *CacheControl)

+ func (dsc *DistributedSafeChannel[T]) Subscribe(ctx context.Context) error

+ type Environment string

+ const EnvDevelopment

+ const EnvProduction

+ const EnvStaging

+ type FanOut struct

+ func NewFanOut[T, R any](numWorkers int) *FanOut[T, R]

+ func (f *FanOut[T, R]) Process(ctx context.Context, items []T, processor func(T) R) []R

+ func (f *FanOut[T, R]) ProcessWithIndex(ctx context.Context, items []T, processor func(int, T) R) []R

+ type FeatureFlag struct

+ CreatedAt time.Time

+ Description string

+ Enabled bool

+ Environments map[Environment]bool

+ Name string

+ Rollout *RolloutConfig

+ UpdatedAt time.Time

+ func (f *FeatureFlag) IsEnabledForEnv(env Environment) bool

+ func (f *FeatureFlag) IsEnabledForUser(env Environment, userID string, userSegments []string) bool

+ type FeatureFlagSet struct

+ func NewFeatureFlagSet(config *FeatureFlagSetConfig) (*FeatureFlagSet, error)

+ func NewFeatureFlagSetSimple(redisAddr string, env Environment) (*FeatureFlagSet, error)

+ func (ffs *FeatureFlagSet) ClearCache()

+ func (ffs *FeatureFlagSet) Close() error

+ func (ffs *FeatureFlagSet) CreateFlag(ctx context.Context, name string, enabled bool, description string) error

+ func (ffs *FeatureFlagSet) DeleteFlag(ctx context.Context, flagName string) error

+ func (ffs *FeatureFlagSet) GetEnvironment() Environment

+ func (ffs *FeatureFlagSet) GetFlag(ctx context.Context, flagName string) (*FeatureFlag, error)

+ func (ffs *FeatureFlagSet) IsEnabled(ctx context.Context, flagName string) (bool, error)

+ func (ffs *FeatureFlagSet) IsEnabledForEnv(ctx context.Context, flagName string, env Environment) (bool, error)

+ func (ffs *FeatureFlagSet) IsEnabledForUser(ctx context.Context, flagName string, userID string, userSegments []string) (bool, error)

+ func (ffs *FeatureFlagSet) IsEnabledForUserInEnv(ctx context.Context, flagName string, env Environment, userID string, ...) (bool, error)

+ func (ffs *FeatureFlagSet) ListFlags(ctx context.Context) ([]*FeatureFlag, error)

+ func (ffs *FeatureFlagSet) SetAllAtOnceRollout(ctx context.Context, name string) error

+ func (ffs *FeatureFlagSet) SetCanaryRollout(ctx context.Context, name string, segments []string) error

+ func (ffs *FeatureFlagSet) SetEnvironment(env Environment)

+ func (ffs *FeatureFlagSet) SetFlag(ctx context.Context, flag *FeatureFlag) error

+ func (ffs *FeatureFlagSet) SetFlagForEnv(ctx context.Context, name string, env Environment, enabled bool) error

+ func (ffs *FeatureFlagSet) SetGradualRollout(ctx context.Context, name string, percentage int) error

+ func (ffs *FeatureFlagSet) SetRolloutPolicy(ctx context.Context, name string, rollout *RolloutConfig) error

+ func (ffs *FeatureFlagSet) SetTargetedRollout(ctx context.Context, name string, userIDs []string) error

+ func (ffs *FeatureFlagSet) UpdateFlag(ctx context.Context, name string, enabled bool) error

+ type FeatureFlagSetConfig struct

+ CacheTTL time.Duration

+ Environment Environment

+ KeyPrefix string

+ RedisAddr string

+ RedisDB int

+ RedisPassword string

+ func DefaultFeatureFlagSetConfig() *FeatureFlagSetConfig

+ type Func struct

+ FuncArgs []interface{}

+ FuncName interface{}

+ Name string

+ func NewFunc(f Func) *Func

+ type Generator struct

+ func NewGenerator[T any](ctx context.Context, bufferSize int, producer func(context.Context, chan<- T)) *Generator[T]

+ func (g *Generator[T]) Channel() <-chan T

+ func (g *Generator[T]) Close()

+ func (g *Generator[T]) Collect() []T

+ func (g *Generator[T]) CollectN(n int) []T

+ func (g *Generator[T]) Next() (T, bool)

+ type GoManager struct

+ func NewGoManager() *GoManager

+ func (GR *GoManager) AddCancelFunc(name string, cancelFunc context.CancelFunc)

+ func (GR *GoManager) Cancel(name string)

+ func (GR *GoManager) GO(name string, fn interface{}, argv ...interface{})

+ func (gr *GoManager) Poll(f ...Func)

+ type Group struct

+ func NewGroup() *Group

+ func (g *Group) Assign(result *any, fn func() any)

+ func (g *Group) Resolve()

+ func (g *Group) ResolveWithTimeout(timeout time.Duration) bool

+ type Iterator interface

+ Next func() bool

+ Value func() T

+ type KeyFunc func(params P) string

+ func FormatKeyFunc[P any](format string) KeyFunc[P]

+ func SimpleKeyFunc[P any](fn func(P) string) KeyFunc[P]

+ func StringKeyFunc() KeyFunc[string]

+ type ParametricFetcher struct

+ func NewParametricFetcher[P any, T any](fetchFunc func(ctx context.Context, params P) (T, error), keyFunc KeyFunc[P], ...) *ParametricFetcher[P, T]

+ func (pf *ParametricFetcher[P, T]) ClearCache()

+ func (pf *ParametricFetcher[P, T]) Fetch(ctx context.Context, params P) (T, error)

+ func (pf *ParametricFetcher[P, T]) FetchStaleWhileRevalidate(ctx context.Context, params P) (T, error)

+ func (pf *ParametricFetcher[P, T]) SetCacheControl(control *CacheControl)

+ type Pipeline struct

+ func NewPipeline[T any]() *Pipeline[T]

+ func (p *Pipeline[T]) AddStage(stage func(T) T) *Pipeline[T]

+ func (p *Pipeline[T]) Execute(item T) T

+ func (p *Pipeline[T]) ExecuteAsync(ctx context.Context, items []T) []T

+ type PreflightFetcher struct

+ func NewPreflightFetcher[T any](fetchFunc func(ctx context.Context, key string) (T, error), ...) *PreflightFetcher[T]

+ func (pf *PreflightFetcher[T]) ClearCache()

+ func (pf *PreflightFetcher[T]) Fetch(ctx context.Context, key string) (T, error)

+ func (pf *PreflightFetcher[T]) FetchStaleWhileRevalidate(ctx context.Context, key string) (T, error)

+ func (pf *PreflightFetcher[T]) SetCacheControl(control *CacheControl)

+ type RateLimiter struct

+ func NewRateLimiter(rate int) *RateLimiter

+ func (rl *RateLimiter) Stop()

+ func (rl *RateLimiter) TryAcquire() bool

+ func (rl *RateLimiter) Wait(ctx context.Context) error

+ type RecastOptions struct

+ OmitFields []string

+ RenameFields map[string]string

+ TransformFuncs map[string]TransformFunc

+ type RetryPolicy struct

+ BackoffFactor float64

+ MaxRetries int

+ MaxRetryDelay time.Duration

+ RetryDelay time.Duration

+ func DefaultRetryPolicy() *RetryPolicy

+ type RolloutConfig struct

+ CanarySegments []string

+ Percentage int

+ Policy RolloutPolicy

+ TargetUserIDs []string

+ type RolloutPolicy string

+ const RolloutAllAtOnce

+ const RolloutCanary

+ const RolloutGradual

+ const RolloutTargeted

+ type SafeChannel struct

+ func NewSafeChannel[T any](bufferSize int, defaultTimeout time.Duration) *SafeChannel[T]

+ func (sc *SafeChannel[T]) Cap() int

+ func (sc *SafeChannel[T]) Close()

+ func (sc *SafeChannel[T]) IsClosed() bool

+ func (sc *SafeChannel[T]) Len() int

+ func (sc *SafeChannel[T]) Receive(ctx context.Context) (T, error)

+ func (sc *SafeChannel[T]) Send(ctx context.Context, value T) error

+ func (sc *SafeChannel[T]) TryReceive() (T, error)

+ func (sc *SafeChannel[T]) TrySend(value T) error

+ type Semaphore struct

+ func NewSemaphore(permits int) *Semaphore

+ func (s *Semaphore) Acquire(ctx context.Context) error

+ func (s *Semaphore) AcquireAll(ctx context.Context) error

+ func (s *Semaphore) Release()

+ func (s *Semaphore) ReleaseAll()

+ func (s *Semaphore) TryAcquire() bool

+ type Stream struct

+ func FromSlice[T any](slice []T) *Stream[T]

+ func Map[T, U any](s *Stream[T], mapper func(T) U) *Stream[U]

+ func (s *Stream[T]) Collect() []T

+ func (s *Stream[T]) Filter(predicate func(T) bool) *Stream[T]

+ func (s *Stream[T]) ForEach(action func(T))

+ func (s *Stream[T]) Reduce(initial T, reducer func(T, T) T) T

+ type SuperSlice struct

+ func NewSuperSliceWithConfig[T any](data []T, config *SuperSliceConfig) *SuperSlice[T]

+ func NewSuperSlice[T any](data []T) *SuperSlice[T]

+ func (ss *SuperSlice[T]) FilterSlice(predicate func(index int, item T) bool) []T

+ func (ss *SuperSlice[T]) ForEach(callback func(index int, item T))

+ func (ss *SuperSlice[T]) GetData() []T

+ func (ss *SuperSlice[T]) Len() int

+ func (ss *SuperSlice[T]) Process(callback func(index int, item T) T) []T

+ func (ss *SuperSlice[T]) ProcessWithError(callback func(index int, item T) (T, error)) ([]T, error)

+ func (ss *SuperSlice[T]) WithConfig(config *SuperSliceConfig) *SuperSlice[T]

+ func (ss *SuperSlice[T]) WithInPlace() *SuperSlice[T]

+ func (ss *SuperSlice[T]) WithIterable() *SuperSlice[T]

+ func (ss *SuperSlice[T]) WithThreshold(threshold int) *SuperSlice[T]

+ func (ss *SuperSlice[T]) WithWorkers(numWorkers int) *SuperSlice[T]

+ type SuperSliceConfig struct

+ InPlace bool

+ NumWorkers int

+ Threshold int

+ UseIterable bool

+ func DefaultSuperSliceConfig() *SuperSliceConfig

+ type SwissMap struct

+ func NewSwissMapWithShards[K comparable, V any](shardCount uint32) *SwissMap[K, V]

+ func NewSwissMap[K comparable, V any]() *SwissMap[K, V]

+ func (sm *SwissMap[K, V]) Clear()

+ func (sm *SwissMap[K, V]) Delete(key K)

+ func (sm *SwissMap[K, V]) Get(key K) (V, bool)

+ func (sm *SwissMap[K, V]) GetOrCompute(key K, compute func() V) V

+ func (sm *SwissMap[K, V]) GetOrSet(key K, value V) (V, bool)

+ func (sm *SwissMap[K, V]) Has(key K) bool

+ func (sm *SwissMap[K, V]) Keys() []K

+ func (sm *SwissMap[K, V]) Len() int

+ func (sm *SwissMap[K, V]) Range(f func(key K, value V) bool)

+ func (sm *SwissMap[K, V]) Set(key K, value V)

+ func (sm *SwissMap[K, V]) ToMap() map[K]V

+ func (sm *SwissMap[K, V]) Values() []V

+ type Task struct

+ type TaskFunc func(ctx context.Context) error

+ type TaskHandler func(ctx context.Context, args map[string]interface{}) error

+ func GetTaskHandler(name string) (TaskHandler, error)

+ type TaskPriority int

+ const PriorityCritical

+ const PriorityHigh

+ const PriorityLow

+ const PriorityNormal

+ type TaskResult struct

+ Attempts int

+ EndTime time.Time

+ Error error

+ Result interface{}

+ StartTime time.Time

+ Status TaskStatus

+ TaskID string

+ type TaskStatus string

+ const StatusExpired

+ const StatusFailed

+ const StatusPending

+ const StatusRetrying

+ const StatusRunning

+ const StatusSuccess

+ type TransformFunc func(interface{}) interface

+ type WorkerPool struct

+ func NewWorkerPool(numWorkers int) *WorkerPool

+ func NewWorkerPoolWithConfig(config *WorkerPoolConfig) *WorkerPool

+ func (wp *WorkerPool) AckTask(taskID string)

+ func (wp *WorkerPool) CancelCron(taskID string)

+ func (wp *WorkerPool) ClearResults()

+ func (wp *WorkerPool) GetDeadLetterTasks() []*WorkerTask

+ func (wp *WorkerPool) GetQueueLength(queueName string) int

+ func (wp *WorkerPool) GetResult(taskID string) (*TaskResult, bool)

+ func (wp *WorkerPool) IsRunning() bool

+ func (wp *WorkerPool) RejectTask(taskID string, requeue bool) error

+ func (wp *WorkerPool) RequeueDeadLetter(taskID string) error

+ func (wp *WorkerPool) Start()

+ func (wp *WorkerPool) Stop()

+ func (wp *WorkerPool) Submit(task *WorkerTask) error

+ func (wp *WorkerPool) SubmitCron(task *WorkerTask, cronExpr string) error

+ func (wp *WorkerPool) SubmitDelayed(task *WorkerTask, delay time.Duration) error

+ func (wp *WorkerPool) SubmitToQueue(queueName string, task *WorkerTask) error

+ func (wp *WorkerPool) WorkerCount() int

+ type WorkerPoolConfig struct

+ EnableQueues bool

+ EnableResults bool

+ MaxDeadLetter int

+ NumWorkers int

+ OnTaskComplete func(task *WorkerTask, result *TaskResult)

+ OnTaskFailed func(task *WorkerTask, err error)

+ OnTaskStart func(task *WorkerTask)

+ func DefaultWorkerPoolConfig() *WorkerPoolConfig

+ type WorkerTask struct

+ Args map[string]interface{}

+ CronExpr string

+ Delay time.Duration

+ ExpiresAt *time.Time

+ Func TaskFunc

+ HandlerName string

+ ID string

+ IsRecurring bool

+ Priority TaskPriority

+ Queue string

+ RetryPolicy *RetryPolicy

+ Timeout time.Duration

+ func DecodeFromBroker(data []byte) (*WorkerTask, error)

+ func (wt *WorkerTask) EncodeToBroker() ([]byte, error)

+ func (wt *WorkerTask) GetAttempts() int

+ func (wt *WorkerTask) GetStatus() TaskStatus

+ func (wt *WorkerTask) MarshalJSON() ([]byte, error)

v1.0.7

Nov 20, 2025

v1.0.6

Nov 20, 2025

v1.0.5

Nov 20, 2025

v1.0.2

Oct 9, 2024

v1.0.1

Jul 29, 2024

v1.0.0

Jun 24, 2024

v0

v0.0.2

Jun 19, 2024

v0.0.1

Jun 19, 2024