README
¶
go-batch
Batch processing and chunking utilities for Go. Generic, concurrent, zero dependencies
Installation
go get github.com/philiprehberger/go-batch
Usage
Chunking
import "github.com/philiprehberger/go-batch"
records := []Record{...} // 1000 records
for _, chunk := range batch.Chunk(records, 100) {
db.BulkInsert(chunk) // insert 100 at a time
}
Concurrent Processing
import "github.com/philiprehberger/go-batch"
err := batch.Process(ctx, userIDs, 50, func(ctx context.Context, ids []int) error {
return sendNotifications(ctx, ids)
}, batch.WithWorkers(4))
Error Handling
import "github.com/philiprehberger/go-batch"
errs := batch.ProcessWithErrors(records, 100, 4, func(batch []Record) error {
return db.BulkInsert(batch)
})
for _, err := range errs {
log.Printf("batch failed: %v", err)
}
ChunkBy
import "github.com/philiprehberger/go-batch"
type Order struct {
Status string
ID int
}
orders := []Order{{Status: "pending", ID: 1}, {Status: "shipped", ID: 2}, {Status: "pending", ID: 3}}
grouped := batch.ChunkBy(orders, func(o Order) string { return o.Status })
// grouped["pending"] = [{pending 1} {pending 3}]
// grouped["shipped"] = [{shipped 2}]
Auto-flushing Accumulator
import "github.com/philiprehberger/go-batch"
acc := batch.NewAccumulator(func(events []Event) {
publishEvents(events)
}, batch.FlushSize[Event](100), batch.FlushInterval[Event](5*time.Second))
for event := range incomingEvents {
acc.Add(event)
}
acc.Stop()
Stats
acc := batch.NewAccumulator(func(events []Event) {
publishEvents(events)
}, batch.FlushSize[Event](100))
acc.Add(event1)
acc.Add(event2)
stats := acc.Stats()
fmt.Printf("flushes: %d, total: %d, pending: %d\n", stats.FlushCount, stats.TotalItems, stats.Pending)
Peek
acc := batch.NewAccumulator[Event](func(events []Event) {
publishEvents(events)
}, batch.FlushSize[Event](100))
acc.Add(event1)
acc.Add(event2)
buffered := acc.Peek() // returns copy of buffered items without flushing
fmt.Println(len(buffered)) // 2
API
| Function | Description |
|---|---|
Chunk[T](items, size) |
Split slice into batches |
ChunkBy[T, K](items, key) |
Group items into map by key function |
Process[T](ctx, items, size, fn, opts...) |
Concurrent batch processing |
ProcessWithErrors[T](items, size, workers, fn) |
Concurrent processing, returns error slice |
WithWorkers(n) |
Set number of concurrent workers |
NewAccumulator[T](fn, opts...) |
Auto-flushing item accumulator |
FlushSize[T](n) |
Flush when N items accumulated |
FlushInterval[T](d) |
Flush on time interval |
OnFlush[T](fn) |
Callback invoked after each flush |
Accumulator.Stats() |
Returns flush count, total items, pending |
Accumulator.Peek() |
Returns copy of buffered items without flushing |
Development
go test ./...
go vet ./...
License
MIT
Documentation
¶
Overview ¶
Package batch provides batch processing and chunking utilities for Go.
It includes generic functions for splitting slices into chunks, processing batches concurrently with bounded parallelism, and an auto-flushing accumulator with size and time-based triggers.
Index ¶
- func Chunk[T any](items []T, size int) [][]T
- func ChunkBy[T any, K comparable](items []T, key func(T) K) map[K][]T
- func Process[T any](ctx context.Context, items []T, batchSize int, ...) error
- func ProcessWithErrors[T any](items []T, size int, workers int, fn func([]T) error) []error
- type Accumulator
- type AccumulatorOption
- type AccumulatorStats
- type ProcessOption
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Chunk ¶
Chunk splits items into sub-slices of the given size. The last chunk may contain fewer than size elements. Chunk panics if size is less than or equal to zero.
func ChunkBy ¶ added in v0.2.0
func ChunkBy[T any, K comparable](items []T, key func(T) K) map[K][]T
ChunkBy groups items by the value returned from the key function. Items sharing the same key are collected into the same slice. The iteration order of the returned map is not guaranteed.
func Process ¶
func Process[T any](ctx context.Context, items []T, batchSize int, fn func(ctx context.Context, batch []T) error, opts ...ProcessOption) error
Process splits items into batches of batchSize and calls fn for each batch. Batches are processed concurrently up to the number of workers configured via WithWorkers (default 1). If any invocation of fn returns an error, all errors are collected and returned as a single combined error. Processing respects context cancellation — remaining batches are skipped when ctx is cancelled.
func ProcessWithErrors ¶ added in v0.2.0
ProcessWithErrors splits items into batches of batchSize and calls fn for each batch using the given number of concurrent workers. Unlike Process, it does not use a context and returns all non-nil errors as a slice rather than combining them.
Types ¶
type Accumulator ¶
type Accumulator[T any] struct { // contains filtered or unexported fields }
Accumulator collects items and flushes them in batches. Flushing can be triggered manually, by reaching a size threshold, or on a time interval. All methods are safe for concurrent use.
func NewAccumulator ¶
func NewAccumulator[T any](fn func(items []T), opts ...AccumulatorOption[T]) *Accumulator[T]
NewAccumulator creates a new Accumulator that calls fn with the buffered items on each flush. Configure automatic flushing with FlushSize and FlushInterval options.
func (*Accumulator[T]) Add ¶
func (a *Accumulator[T]) Add(item T)
Add appends an item to the accumulator. If a FlushSize has been configured and the number of buffered items reaches that threshold, a flush is triggered automatically.
func (*Accumulator[T]) Flush ¶
func (a *Accumulator[T]) Flush()
Flush sends all buffered items to the flush callback and resets the buffer. It is safe to call concurrently.
func (*Accumulator[T]) Peek ¶ added in v0.2.0
func (a *Accumulator[T]) Peek() []T
Peek returns a copy of the currently buffered items without flushing. It is safe to call concurrently.
func (*Accumulator[T]) Stats ¶ added in v0.2.0
func (a *Accumulator[T]) Stats() AccumulatorStats
Stats returns statistics about the accumulator's lifetime. It is safe to call concurrently.
func (*Accumulator[T]) Stop ¶
func (a *Accumulator[T]) Stop()
Stop stops the interval timer (if running) and flushes any remaining buffered items. After Stop returns the accumulator should not be used.
type AccumulatorOption ¶
type AccumulatorOption[T any] func(*accumulatorConfig)
AccumulatorOption configures the behaviour of an Accumulator.
func FlushInterval ¶
func FlushInterval[T any](d time.Duration) AccumulatorOption[T]
FlushInterval configures the Accumulator to automatically flush on a recurring time interval. A background goroutine is started and runs until Stop is called.
func FlushSize ¶
func FlushSize[T any](n int) AccumulatorOption[T]
FlushSize configures the Accumulator to automatically flush when n items have been accumulated.
func OnFlush ¶ added in v0.2.0
func OnFlush[T any](fn func([]T)) AccumulatorOption[T]
OnFlush registers a callback that is invoked after each flush with the flushed items. The callback runs after the main flush function.
type AccumulatorStats ¶ added in v0.2.0
type AccumulatorStats struct {
// FlushCount is the total number of flushes performed.
FlushCount int64
// TotalItems is the total number of items flushed.
TotalItems int64
// Pending is the number of items currently buffered.
Pending int
}
AccumulatorStats contains statistics about an Accumulator's lifetime.
type ProcessOption ¶
type ProcessOption func(*processConfig)
ProcessOption configures the behaviour of Process.
func WithWorkers ¶
func WithWorkers(n int) ProcessOption
WithWorkers sets the number of concurrent workers used by Process. The default is 1 (sequential processing). If n is less than 1 it is treated as 1.
Source Files