> Discover all available pages from the documentation index: https://mastra.zisheng.pro/zh-TW/llms.txt

# RedisStreamsPubSub

`RedisStreamsPubSub` 是以 [Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/) 為基礎的 [`PubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/base) 實作。它可在多個 process 與主機間傳遞 event，並支援持久化、consumer group 與失敗時重新傳遞。它也實作 [`LeaseProvider`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/lease-provider)，因此 signal layer 可跨 instance 為每項資源選出單一擁有者，讓 signal 能在分散式與 serverless 部署中協調 run。

適用於多個服務共用 event stream 的分散式部署。單一 process 傳遞請使用 [`EventEmitterPubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/event-emitter)。Google Cloud 請使用 [`GoogleCloudPubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/google-cloud-pubsub)。

每個 topic 對應至 Redis stream key。具有 group 的 subscription 會使用 Redis consumer group，因此成員以 round-robin 方式分擔工作。沒有 group 的 subscription 會建立私有 consumer group，因此每個 subscriber 都會收到每個 event。

`RedisStreamsPubSub` 是 pull transport：consumer 使用 `XREADGROUP` 讀取 event，因此 Mastra 會代為執行 orchestration worker。

## 安裝

**npm**:

```bash
npm install @mastra/redis-streams
```

**pnpm**:

```bash
pnpm add @mastra/redis-streams
```

**Yarn**:

```bash
yarn add @mastra/redis-streams
```

**Bun**:

```bash
bun add @mastra/redis-streams
```

## 使用範例

提供 Redis 連線 URL。

```typescript
import { Mastra } from '@mastra/core'
import { RedisStreamsPubSub } from '@mastra/redis-streams'

export const mastra = new Mastra({
  pubsub: new RedisStreamsPubSub({
    url: 'redis://localhost:6379',
  }),
})
```

## Constructor 參數

**url** (`string`): Redis 連線 URL。未提供時改用 redisOptions.url。 (Default: `redis://localhost:6379`)

**keyPrefix** (`string`): stream key 的前綴。每個 topic 對應至 \<keyPrefix>:\<topic>。 (Default: `mastra:topic`)

**blockMs** (`number`): 每次讀取等待新 event 時，最多阻塞的毫秒數。 (Default: `1000`)

**redisOptions** (`RedisClientOptions`): 為進階設定傳給底層 redis client 的選項。

**maxStreamLength** (`number`): 每個 stream 保留的大約最大 entry 數。設定為 0 可停用裁剪。 (Default: `10000`)

**streamIdleTtlMs** (`number`): 閒置到期時間（毫秒）：每次寫入（publish、nack retry、重新建立 group）都會重新整理的滑動 TTL。每次寫入都會重設，因此持續寫入的 stream 不會在處理期間到期；完整閒置至指定期間的 stream 會由 Redis 自動刪除。請注意，只有寫入會重新整理 TTL，緩慢清空 backlog 的 consumer 不會，因此此值應遠高於即時 topic 的最長預期寫入間隔。這是後備機制，而不是主要清理方式；clearTopic 會處理一般生命週期結束時的刪除，此設定只限制從未呼叫 clearTopic 的 stream（例如當機的 run）所占用的記憶體。必須是非負整數。預設為 0（停用）。 (Default: `0`)

**reclaimIntervalMs** (`number`): subscription 重新取得先前 consumer 已讀取但未確認的 event 之時間間隔（毫秒）。設定為 0 可停用。 (Default: `30000`)

**reclaimIdleMs** (`number`): 待處理 event 符合重新取得資格前的最短閒置時間（毫秒）。此值應遠高於一般處理時間，以避免重複傳遞。 (Default: `60000`)

**maxDeliveryAttempts** (`number`): event 透過 nack 重新傳遞後，遭到捨棄前的最大次數。傳入 Infinity 可停用上限。 (Default: `5`)

**logger** (`{ debug?: Function; warn?: Function }`): 用於診斷的選填 logger。省略時，被抑制的錯誤不會顯示。

## 屬性

**supportedModes** (`ReadonlyArray<"pull" | "push">`): 傳回 \["pull"]。

## 方法

`RedisStreamsPubSub` 實作 [`PubSub`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/base) contract。以下方法具有此實作專屬的行為。

### `subscribe(topic, cb, options?)`

訂閱 topic。設定 `options.group` 後，group 成員會透過 Redis consumer group 共用 event。未設定 group 時，subscriber 會透過私有 consumer group 收到每個 event。

```typescript
await pubsub.subscribe('workflow.events', (event, ack, nack) => {
  console.log(event)
})
```

### `flush()`

等待進行中的 publish 完成。

```typescript
await pubsub.flush()
```

### `clearTopic(topic)`

刪除 topic 的 stream 及其上的每個 consumer group，釋放已完成 topic 原本會持續占用的記憶體。Mastra 的 run 生命週期（durable Agent 與 event 型 Workflow engine）會在 run 進入終止 state 時自動呼叫。只有確認不會再有人讀取 topic 時，才自行呼叫。此方法採盡力而為，絕不會擲回例外；失敗會以 warn 層級記錄。刪除 stream 時仍連接的 subscriber 會自行恢復，但會錯過已刪除的 entry。

自動清理需要 `@mastra/core` 與 `@mastra/redis-streams` 版本都支援 `clearTopic`：runtime 會透過其 caching layer 路由呼叫，因此請同時升級兩個 package，才能在 run 結束時刪除 stream。

```typescript
await pubsub.clearTopic('workflow.events.run-123')
```

### `close()`

關閉 Redis 連線並停止所有 subscription。請在正常關閉期間呼叫此方法。

```typescript
await pubsub.close()
```

## 重新傳遞與重新取得

subscriber 呼叫 `nack` 時，event 會以遞增的 `deliveryAttempt` 重新發布，原始 event 則會獲得確認。event 達到 `maxDeliveryAttempts` 後會遭到捨棄，不再重新傳遞。另外，每個 subscription 會定期重新取得同一 group 中先前 consumer 已讀取但未確認的 event，此行為由 `reclaimIntervalMs` 與 `reclaimIdleMs` 控制。

## 分散式租約

`RedisStreamsPubSub` 會在相同 Redis 連線之上實作 [`LeaseProvider`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/lease-provider) contract。[signal runtime](https://mastra.zisheng.pro/zh-TW/docs/long-running-agents/signals) 會使用它選出單一擁有者（通常以 thread key 為單位），讓多個 instance 中只會有一個 process 喚醒並執行 Agent，其他 process 則將後續工作路由給持有者。這讓 signal 可在 serverless 與多 instance 部署中運作；如果沒有共用租約，每個 instance 都會啟動自己的競爭 run。

租約 key 與 topic 使用相同的 `keyPrefix` namespace，格式為 `<keyPrefix>:lease:<key>`。所有操作皆為原子操作：`acquireLease` 使用 `SET NX PX`，並以冪等方式重新整理自己的 TTL；`releaseLease`、`renewLease` 與 `transferLease` 則使用 Lua script，在變更前檢查擁有權，因此絕不會覆蓋其他 owner 的並行續約。

你不需直接呼叫這些方法。將 `RedisStreamsPubSub` 設定為 `pubsub` 後端，runtime 就能偵測並使用此能力。完整方法 contract 請參閱 [`LeaseProvider`](https://mastra.zisheng.pro/zh-TW/reference/pubsub/lease-provider)。