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

# PubSub

`PubSub` 是 Mastra 事件系统的抽象基类。它定义每个 pub/sub backend 实现的 contract，使 Mastra 的其他部分能够在不了解所用 transport 的情况下发布和订阅事件。

Mastra 在内部将 pub/sub 用于 Workflow 事件处理、streaming 和跨组件通信。大多数应用使用默认 [`EventEmitterPubSub`](https://mastra.zisheng.pro/reference/pubsub/event-emitter)，无需直接构造 `PubSub`。仅当需要自定义 transport 时实现此类。

有关内置实现，请参阅 [`EventEmitterPubSub`](https://mastra.zisheng.pro/reference/pubsub/event-emitter)、[`UnixSocketPubSub`](https://mastra.zisheng.pro/reference/pubsub/unix-socket-pubsub)、[`CachingPubSub`](https://mastra.zisheng.pro/reference/pubsub/caching-pubsub)、[`RedisStreamsPubSub`](https://mastra.zisheng.pro/reference/pubsub/redis-streams) 和 [`GoogleCloudPubSub`](https://mastra.zisheng.pro/reference/pubsub/google-cloud-pubsub)。

## 使用示例

扩展 `PubSub` 并实现四个抽象方法以添加自定义 backend。

```typescript
import { PubSub } from '@mastra/core/events'
import type { Event, EventCallback, SubscribeOptions } from '@mastra/core/events'

export class CustomPubSub extends PubSub {
  async publish(topic: string, event: Omit<Event, 'id' | 'createdAt'>): Promise<void> {
    // Deliver the event to subscribers of `topic`.
  }

  async subscribe(topic: string, cb: EventCallback, options?: SubscribeOptions): Promise<void> {
    // Register `cb` to receive events published to `topic`.
  }

  async unsubscribe(topic: string, cb: EventCallback): Promise<void> {
    // Remove a previously registered callback.
  }

  async flush(): Promise<void> {
    // Wait for any in-flight deliveries to settle.
  }
}
```

将实例传入 [Mastra](https://mastra.zisheng.pro/reference/core/mastra-class) 构造函数：

```typescript
import { Mastra } from '@mastra/core'
import { CustomPubSub } from './pubsub'

export const mastra = new Mastra({
  pubsub: new CustomPubSub(),
})
```

## 投递模式

`PubSub` 通过 `supportedModes` 属性声明所支持的投递模式。Mastra 读取它以决定是否运行拉取事件的长生命周期 Worker。

| 模式     | 说明                                                                            |
| ------ | ----------------------------------------------------------------------------- |
| `pull` | consumer 主动从 broker 读取，例如 Redis Streams `XREADGROUP`。Mastra 会运行编排 Worker 来读取。 |
| `push` | 事件在 consumer 未请求时到达，无论是在进程内还是通过 HTTP endpoint。无需读取循环。                         |

默认值是 `['pull']`，因此自定义实现除非选择加入 push 投递，否则会保留当前行为。

## 方法

### 核心方法

#### `publish(topic, event)`

向 topic 发布事件。`id` 和 `createdAt` 字段由实现赋值。

```typescript
await pubsub.publish('my-topic', {
  type: 'example',
  data: { value: 1 },
  runId: 'run-123',
})
```

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

注册 callback 以接收发布到 topic 的事件。设置 `options.group` 后，同一组的订阅者会竞争消息，每个事件仅投递给一个成员。没有 group 时，每个订阅者都会收到每个事件。

传入 `options.batch` 可选择批量投递。callback 签名不变：N 个事件的批次会按发布顺序作为 N 次连续 `cb(event, ack, nack)` 调用投递。仅当 backend 的 [`supportsNativeBatching`](#properties) 为 `true` 时支持批处理。其他 backend 会忽略该选项并逐个投递事件。

```typescript
await pubsub.subscribe('my-topic', (event, ack, nack) => {
  console.log(event)
})
```

#### `unsubscribe(topic, cb)`

从 topic 移除先前注册的 callback。

```typescript
await pubsub.unsubscribe('my-topic', callback)
```

#### `flush()`

等待所有进行中的投递完成。关闭前调用以避免丢失事件。

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

#### `clearTopic(topic)`

在不再向 topic 发布事件后，删除其所有保留状态（缓存历史、持久 stream 条目和 consumer group）。Mastra 的 run lifecycle 会在 run 达到终态时自动调用它，因此每个 run 的 topic 不会在保留消息的 transport 上累积。

默认实现是无操作：不为每个 topic 保留任何状态的 transport（如 `EventEmitterPubSub`）没有内容可清除。[`RedisStreamsPubSub`](https://mastra.zisheng.pro/reference/pubsub/redis-streams) 等持久化消息的 backend 会重写它。该 contract 是尽力而为的：实现记录失败而不抛出，因为调用者会在清理边界以 fire-and-forget 方式调用它。

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

### Replay 方法

这些方法支持在断开后恢复 stream。默认实现会回退到普通 `subscribe`，因此无历史支持的 backend 只提供实时行为。[`CachingPubSub`](https://mastra.zisheng.pro/reference/pubsub/caching-pubsub) 会重写它们以 replay 缓存事件。

#### `getHistory(topic, offset?)`

返回 topic 从 `offset` 开始的缓存事件。backend 没有历史记录时返回空数组。

```typescript
const events = await pubsub.getHistory('my-topic', 0)
```

返回：`Promise<Event[]>`

#### `subscribeWithReplay(topic, cb)`

replay 缓存事件，然后订阅实时事件。

```typescript
await pubsub.subscribeWithReplay('my-topic', event => {
  console.log(event)
})
```

#### `subscribeFromOffset(topic, offset, cb)`

从已知位置开始 replay 缓存事件，然后订阅实时事件。当 client 知道其最后位置时，这比完整 replay 更高效。

```typescript
await pubsub.subscribeFromOffset('my-topic', 42, event => {
  console.log(event)
})
```

## 属性

**supportedModes** (`ReadonlyArray<"pull" | "push">`): 实现支持的投递模式。默认值为 \["pull"]。

**supportsNativeBatching** (`boolean`): 实现是否在 subscribe() 中支持 options.batch。默认值为 false。在内部集成批处理的 backend 会重写此项并返回 true。

## 类型

### `Event`

**type** (`string`): 事件类型标识符。

**id** (`string`): 唯一事件 ID，由实现于发布时分配。

**data** (`any`): 事件 payload。

**runId** (`string`): 事件所属的 run。

**createdAt** (`Date`): 由实现于发布时分配的时间戳。

**index** (`number`): 用于从特定 offset 恢复的顺序位置。

**deliveryAttempt** (`number`): 事件已投递的次数。从 1 开始。当 backend 不跟踪重新投递时，默认为 1。

### `SubscribeOptions`

**group** (`string`): 设置后，具有相同 group 的订阅者会竞争消息，每个事件仅投递给一个成员。省略时，每个订阅者都会收到每个事件。

**batch** (`SubscribeBatchOptions`): 为此订阅选择批量投递。省略时，事件逐个投递。仅在 supportsNativeBatching 为 true 的 backend 上支持。

### `SubscribeBatchOptions`

每个订阅的批处理策略。callback 签名不会改变。N 个事件的批次会按发布顺序变为 N 次连续 callback 调用。

**maxSize** (`number`): 强制 flush 前保留的最大事件数。

**maxWaitMs** (`number`): 最早事件可在 buffer 中停留的最长毫秒数。buffer 从空变为非空时启动计时器。

**minIntervalMs** (`number`): 连续批次投递之间的最小毫秒数。即使 maxSize 或 maxWaitMs 触发，buffer 也会保留到距离上次投递达到此间隔。

**isImmediate** (`(event: Event) => boolean`): 当它对事件返回 true 时，buffer 会在发布时立即 flush，但仍受 minIntervalMs 约束。每事件逃生阀。

**coalesce** (`(events: Event[]) => Event[]`): 在投递前应用于排队批次以丢弃已被取代的事件。必须按引用身份返回输入的子集；返回新构造的 Event 对象违反 contract，并会丢弃整个批次。保留事件的顺序不变。

**maxBufferSize** (`number`): 在 overflow 处理开始前 buffer 可保留的最大事件数。标记为 immediate 的事件不会在 overflow 时丢弃。 (Default: `256`)

**overflow** (`"drop-oldest" | "drop-newest" | "coalesce-or-drop-oldest"`): buffer 超过 maxBufferSize 时的 overflow 策略。coalesce-or-drop-oldest 会先运行 coalesce，如仍超出预算则丢弃最早事件。 (Default: `coalesce-or-drop-oldest`)

### `EventCallback`

订阅者的 callback 签名：`(event: Event, ack?: () => Promise<void>, nack?: () => Promise<void>) => void`。

**event** (`Event`): 已投递的事件。

**ack** (`() => Promise<void>`): 确认处理成功。事件将从队列中移除。

**nack** (`() => Promise<void>`): 否定确认。事件将在延迟后重新加入队列以重新投递。