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

# RedisStreamsPubSub

`RedisStreamsPubSub` 是由 [Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/) 支援的 [`PubSub`](https://mastra.zisheng.pro/zh-HK/reference/pubsub/base) 實作。它可跨程序及主機傳送事件，並支援持久化、使用者群組及失敗後重新傳送。它亦實作 [`LeaseProvider`](https://mastra.zisheng.pro/zh-HK/reference/pubsub/lease-provider)，因此 signals 層可以跨實例為每項資源選出一名擁有者，讓 signals 能夠在分散式及 serverless 部署中協調執行。

適用於多個服務共用事件串流的分散式部署。如要在單一程序內傳送，請使用 [`EventEmitterPubSub`](https://mastra.zisheng.pro/zh-HK/reference/pubsub/event-emitter)。如使用 Google Cloud，請使用 [`GoogleCloudPubSub`](https://mastra.zisheng.pro/zh-HK/reference/pubsub/google-cloud-pubsub)。

每個主題都會對應至 Redis 串流 key。設有群組的訂閱會使用 Redis 使用者群組，讓成員以 round-robin 方式分擔工作。沒有群組的訂閱則會建立私人使用者群組，讓每名訂閱者都收到每個事件。

`RedisStreamsPubSub` 是拉取式傳輸：使用者透過 `XREADGROUP` 讀取事件，因此 Mastra 會代為執行協調 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',
  }),
})
```

## 建構函式參數

**url** (`string`): Redis 連線 URL。如未設定，會使用 redisOptions.url。 (Default: `redis://localhost:6379`)

**keyPrefix** (`string`): 串流 key 的前綴。每個主題都會對應至 \<keyPrefix>:\<topic>。 (Default: `mastra:topic`)

**blockMs** (`number`): 每次讀取在等待新事件時阻塞的時間（毫秒）。 (Default: `1000`)

**redisOptions** (`RedisClientOptions`): 傳入底層 redis client、用於進階設定的選項。

**maxStreamLength** (`number`): 每個串流大約保留的項目數目上限。設為 0 可停用修剪。 (Default: `10000`)

**streamIdleTtlMs** (`number`): 閒置到期時間（毫秒）：每次寫入（發佈、nack 重試、重新建立群組）時都會重新整理的滑動 TTL。每次寫入都會重設 TTL，因此持續有內容寫入的串流不會在處理期間到期；串流閒置整段指定時間後，Redis 會自動將其刪除。請注意，只有寫入操作會重新整理 TTL，緩慢清理 backlog 的使用者並不會重新整理；因此，此值應遠高於即時主題兩次寫入之間預期的最長間隔。這是最後保障，而非主要清理機制；clearTopic 會處理生命週期正常結束時的刪除，而此設定只會限制從未呼叫 clearTopic 的串流（例如執行崩潰）的記憶體用量。必須是非負整數。預設為 0（停用）。 (Default: `0`)

**reclaimIntervalMs** (`number`): 訂閱收回先前使用者已讀取但從未確認之事件的頻率（毫秒）。設為 0 可停用。 (Default: `30000`)

**reclaimIdleMs** (`number`): 待處理事件符合收回資格前的最短閒置時間（毫秒）。此值應遠高於一般處理時間，以免重複傳送。 (Default: `60000`)

**maxDeliveryAttempts** (`number`): 事件透過 nack 重新傳送並最終被捨棄前的最大次數。傳入 Infinity 可停用此上限。 (Default: `5`)

**logger** (`{ debug?: Function; warn?: Function }`): 用於診斷的可選 logger。如省略此項，被抑制的錯誤不會產生任何輸出。

## 屬性

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

## 方法

`RedisStreamsPubSub` 會實作 [`PubSub`](https://mastra.zisheng.pro/zh-HK/reference/pubsub/base) 協定。以下方法具有此實作特有的行為。

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

訂閱主題。設有 `options.group` 時，群組成員會透過 Redis 使用者群組共用事件。沒有群組時，訂閱者會透過私人使用者群組收到每個事件。

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

### `flush()`

等待進行中的發佈完成。

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

### `clearTopic(topic)`

刪除主題的串流及其中所有使用者群組，釋放已完成主題原本會佔用的記憶體。當一次執行到達終止狀態時，Mastra 的執行生命週期（durable Agent 及事件驅動的 Workflow 引擎）會自動呼叫此方法。只有在不會再有任何內容讀取該主題後，才應自行呼叫此方法。此方法會盡力完成操作，絕不會拋出錯誤。失敗會記錄在 warn level。刪除串流時仍保持連接的訂閱者會自行恢復，但會錯過已刪除的項目。

自動清理需要 `@mastra/core` 及 `@mastra/redis-streams` 的版本都支援 `clearTopic`：runtime 會透過快取層轉送呼叫，因此請同時升級兩個依賴套件，以便在執行結束時刪除串流。

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

### `close()`

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

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

## 重新傳送及收回

訂閱者呼叫 `nack` 時，事件會以遞增的 `deliveryAttempt` 重新發佈，而原始事件會被確認。事件到達 `maxDeliveryAttempts` 後，會被捨棄而不再傳送。此外，每個訂閱都會定期收回群組中先前使用者已讀取但從未確認的事件；此行為由 `reclaimIntervalMs` 及 `reclaimIdleMs` 控制。

## 分散式租約

`RedisStreamsPubSub` 會在同一 Redis 連線上實作 [`LeaseProvider`](https://mastra.zisheng.pro/zh-HK/reference/pubsub/lease-provider) 協定。[signals runtime](https://mastra.zisheng.pro/zh-HK/docs/long-running-agents/signals) 會使用它選出單一擁有者（通常按 thread key 區分），確保跨實例只有一個程序會喚醒及執行 Agent，其他程序則將後續工作轉送給租約持有人。這正是 signals 能夠在 serverless 及多實例部署中運作的原因；如沒有共用租約，每個實例都會開始各自互相競逐的執行。

租約 key 會使用與主題相同的 `keyPrefix` namespace，格式為 `<keyPrefix>:lease:<key>`。所有操作都是原子操作：`acquireLease` 使用 `SET NX PX`，並以 idempotent 方式重新整理自身的 TTL；`releaseLease`、`renewLease` 及 `transferLease` 則使用 Lua scripts，在修改前檢查擁有權，因此不會覆蓋另一名擁有者同時進行的續期操作。

你毋須直接呼叫這些方法。將 `RedisStreamsPubSub` 設定為 `pubsub` 後端，runtime 便會自動偵測並使用此功能。如要了解完整的方法協定，請參閱 [`LeaseProvider`](https://mastra.zisheng.pro/zh-HK/reference/pubsub/lease-provider)。