Automatic pipelining

Batch concurrent go-redis commands into pipelines automatically for high-throughput workloads.

Pipelining sends a batch of commands to the server in a single communication, which avoids the network and processing overhead of sending each command separately. Normally you build a pipeline by hand (see Pipelines and transactions), but this means you must know in advance which commands you want to batch.

Automatic pipelining removes that requirement. When many goroutines issue commands concurrently, go-redis coalesces them into deep pipelines for you, without any pipeline code in your application. This is useful in high-throughput or high-concurrency scenarios. At low concurrency, a plain client is simpler and just as fast, and a hand-written pipeline is generally faster than an auto-generated one.

Automatic pipelining requires github.com/redis/go-redis/v9 v9.22.0 or later.

Blocking and asynchronous pipelining

Automatic pipelining has two methods that share the same underlying engine:

• Blocking (AutoPipeline()) is a drop-in replacement for a normal client. Each command call blocks until it executes and returns its own value and error, exactly like a plain client, so existing code keeps working unchanged. Under concurrency, the engine batches commands from all goroutines into back-to-back pipelines behind the scenes. Per-goroutine ordering is preserved.
• Asynchronous (AsyncAutoPipeline()) offers the highest throughput. Command calls return immediately; reading a result with Val(), Result(), or Err() blocks until the batch executes. Submit a sequence of commands and then read the results afterwards to keep each pipeline as deep as possible.

Both methods are available on Client, ClusterClient, and Ring.

Blocking usage

Call AutoPipeline() to get an AutoPipeliner, then call command methods on it just as you would on a normal client. Each call blocks until it executes, but concurrent callers' commands are batched together automatically:

rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
defer rdb.Close()
ctx := context.Background()

// Blocking: a drop-in for a normal client, batched under the hood.
ap, err := rdb.AutoPipeline()
if err != nil { // only returned for invalid AutoPipelineOptions
    log.Fatal(err)
}
defer ap.Close()

var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
    wg.Add(1)
    go func(i int) {
        defer wg.Done()
        key := fmt.Sprintf("key:%d", i)
        if err := ap.Set(ctx, key, i, 0).Err(); err != nil { // blocks until executed
            log.Printf("set %s: %v", key, err)
        }
    }(i)
}
wg.Wait()

Asynchronous usage

For maximum throughput, use asynchronous execution. Command calls return immediately, so you can submit a sequence of commands and read their results afterwards:

ctx := context.Background()

ap, err := rdb.AsyncAutoPipeline() // ordered by default
if err != nil {
    log.Fatal(err)
}
defer ap.Close()

cmds := make([]*redis.StatusCmd, 0, 200)
for i := 0; i < 200; i++ {
    // Returns immediately without executing.
    cmds = append(cmds, ap.Set(ctx, fmt.Sprintf("key:%d", i), i, 0))
}
for _, cmd := range cmds {
    if err := cmd.Err(); err != nil { // blocks until the batch executes
        log.Printf("set: %v", err)
    }
}

Configuration

AutoPipeline() and AsyncAutoPipeline() take no arguments. They use the AutoPipelineOptions set on the client's options, if any, and otherwise use reasonable default values. To pass options for a single autopipeliner, use AutoPipelineWithOptions() or AsyncAutoPipelineWithOptions() instead:

// On the client, used by both methods.
rdb := redis.NewClient(&redis.Options{
    Addr:                "localhost:6379",
    AutoPipelineOptions: &redis.AutoPipelineOptions{MaxFlushDelay: 100 * time.Microsecond},
})

// Or for a single autopipeliner.
ap, err := rdb.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
    MaxConcurrentBatches: 80,
    Unordered:            true,
})

All four methods return (*AutoPipeliner, error). The error is non-nil only when the options are invalid (for example, setting MaxConcurrentBatches greater than one without also setting Unordered). Invalid options never cause a panic.

The configuration options are:

MaxBatchSize is the one default that differs between the two methods. If you set no options at all, AutoPipeline() uses a built-in preset that targets 300 commands instead of 200. As soon as you supply AutoPipelineOptions, either on the client or to AutoPipelineWithOptions(), that preset no longer applies, and a MaxBatchSize you leave unset means 200.

Connection and buffer tuning is not part of AutoPipelineOptions. Batches use the client's pipeline connections, which you size with the PipelineReadBufferSize, PipelineWriteBufferSize, and PipelinePoolSize fields of the client's options.

Each client holds at most two autopipeliners: one for the blocking method and one for the asynchronous method. Each of them is a singleton that the client creates on first use and then shares with every later caller.

Options therefore only take effect on the call that creates the singleton. If a blocking autopipeliner already exists, a later AutoPipelineWithOptions() call returns that same instance and ignores the options you passed, because AutoPipeline() and AutoPipelineWithOptions() share one singleton between them. Close() stops the singleton for every caller and the next call creates a fresh one, so closing is also how you apply different options. Closing the client is permanent: both methods then return ErrClosed.

Cluster usage

AutoPipeline() and AsyncAutoPipeline() also work on ClusterClient. Commands are routed to the correct shard by key, so the client installs slot-based shard routing to keep each shard's batch on a single master node (rather than splitting every batch across all nodes at flush time). This is why cluster clients default to several shards instead of one. A single batch may span many slots. Ordering is per key: same-key commands stay in order, while sub-pipelines on different nodes run concurrently.

Commands that must reach every node or shard, such as FLUSHALL, cannot be added to a pipeline, so the cluster client rejects them with an error rather than let them spoil a batch shared with other callers. Run them on the plain client instead.

Caveats and limitations

• A command's context is not honored once it is queued, because batches execute on the autopipeliner's own context. Use a plain client if you need per-command deadlines.
• Blocking commands such as BLPOP and WAIT are never batched and run directly on your context.
• The generic Do, DoRaw, and DoRawWriteTo methods run outside the pipeline, on a normal connection, because an arbitrary command name can carry connection state or block the connection. Prefer the typed methods (ap.Set(), ap.Get(), and so on), which are always batched.
• On a dropped connection, a batch is retried as a whole, up to the client's MaxRetries, so non-idempotent commands may execute twice. Set MaxRetries: -1, or use a plain client, for commands that must never be retransmitted.

More information

See the go-redis repository for the example/autopipeline usage tour and further API details.