Перейти к содержимому
actor-ts
Русский

Redis Streams

Это содержимое пока не доступно на вашем языке.

RedisStreamsActor integrates with Redis Streams (the XADD / XREADGROUP family). Right shape when Redis is already in your infrastructure and you want streaming without deploying Kafka or NATS.

import { ActorSystem, RedisStreamsActor, RedisStreamsOptions } from 'actor-ts';


const redisStreamsOptions = RedisStreamsOptions.create()
  .withUrl('redis://localhost:6379')
  .withStreams(['orders', 'payments'])
  // Consumer-group wiring + consumer-actor target live in
  // settings — there is no runtime `subscribe` surface.
  .withConsumerGroup({ group: 'order-processors', consumer: 'worker-1' })
  .withTarget(orderHandler);
const redis = system.spawn(
  () => new RedisStreamsActor(
    redisStreamsOptions,
  ),
  'redis',
);


// Publish (XADD under the hood):
redis.tell({
  kind: 'publish',
  publish: {
    stream: 'orders',
    fields: { sku: 'book-1', quantity: '2' },
  },
});

Settings

Section titled “Settings”
interface RedisStreamsOptionsType extends BrokerCommonOptionsType {
  url:           string;                          // 'redis://host:6379'
  streams?:      ReadonlyArray<string>;           // streams to consume
  consumerGroup?: {
    group:           string;
    consumer:        string;
    createIfMissing?: boolean;                    // default true
  };
  blockMs?:      number;                          // XREADGROUP block; default 5_000
  target?:       ActorRef<RedisStreamEntry>;      // consumer-actor for inbound entries
}

Consuming with a consumer group

Section titled “Consuming with a consumer group”

Consumption always goes through a consumer group — the actor uses XREADGROUP exclusively. The consumer starts only when consumerGroup, streams, and target are all set; there is no group-less XREAD broadcast path.

const redisStreamsOptions = RedisStreamsOptions.create()
  .withUrl(url)
  .withStreams(['orders'])
  .withConsumerGroup({ group: 'workers', consumer: 'worker-1' })
  .withTarget(handler);
new RedisStreamsActor(
  redisStreamsOptions,
);

Consumer-group mode:

  • Work-sharing — each message goes to one consumer in the group.
  • Ack semantics — consumers XACK after processing; un-acked messages are tracked in the pending entries list (PEL) and can be re-delivered.
  • Multiple workers — N actor instances in the same group share the workload.

This is the right mode for work queues — orders go to one worker, processed, acked, done.

Inbound messages

Section titled “Inbound messages”
class OrderProcessor extends Actor<RedisStreamEntry> {
  constructor(private readonly redis: ActorRef<RedisStreamsCommand>) { super(); }


  override async onReceive(entry: RedisStreamEntry): Promise<void> {
    const order = JSON.parse(entry.fields.body);
    await processOrder(order);
    this.redis.tell({ kind: 'acknowledgment', stream: entry.stream, id: entry.id });
  }
}

The entry:

type RedisStreamEntry = {
  stream:  string;
  id:      string;                          // e.g. "1684923847-0"
  fields:  Record<string, string>;          // your XADD payload
};

Stream IDs are <millis>-<seq> strings. Use the id as a dedup key for idempotent processing — even if Redis re-delivers, your handler can skip already-processed IDs.

Pending Entries List (PEL)

Section titled “Pending Entries List (PEL)”

Un-acked messages stay in the consumer group’s PEL — Redis tracks every delivered-but-not-acked entry per consumer. After a consumer crash, those entries can be manually re-claimed via XCLAIM against the Redis server (outside the actor surface). A future actor command would wrap that; for now, run XCLAIM from a separate ioredis client or use a sidecar process.

For at-least-once delivery in the common path: tell ack after processing, and a consumer that crashes mid-processing simply leaves the entry in the PEL — the next consumer that starts with the same group can pick up where the previous one stopped (via XPENDING + XCLAIM).

Peer dependency

Section titled “Peer dependency”
Terminal window
npm install ioredis
# or: bun add ioredis

ioredis is the underlying client. Versions 5+ are tested.

When to use Redis Streams

Section titled “When to use Redis Streams”

Three good fits:

  1. Existing Redis infrastructure — adding streaming without deploying a new broker.
  2. Sub-millisecond latency requirements — Redis is fast.
  3. Smaller workloads — millions per day, not billions. For huge scale, Kafka is purpose-built.

Not the right shape for:

  • Long-term retention — Redis Streams can persist but aren’t designed for years-long retention.
  • Massive parallelism — Redis is single-threaded; one Redis server bottlenecks.

Where to next

Section titled “Where to next”