PubSub

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

Mastra 在内部将 pub/sub 用于 Workflow 事件处理、streaming 和跨组件通信。大多数应用使用默认 EventEmitterPubSub，无需直接构造 PubSub。仅当需要自定义 transport 时实现此类。

有关内置实现，请参阅 EventEmitterPubSub、UnixSocketPubSub、CachingPubSub、RedisStreamsPubSub 和 GoogleCloudPubSub。

使用示例

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

src/mastra/pubsub.ts

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 构造函数：

src/mastra/index.ts

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 字段由实现赋值。

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 为 true 时支持批处理。其他 backend 会忽略该选项并逐个投递事件。

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

unsubscribe(topic, cb)

从 topic 移除先前注册的 callback。

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

flush()

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

await pubsub.flush()

clearTopic(topic)

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

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

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

Replay 方法

这些方法支持在断开后恢复 stream。默认实现会回退到普通 subscribe，因此无历史支持的 backend 只提供实时行为。CachingPubSub 会重写它们以 replay 缓存事件。

getHistory(topic, offset?)

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

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

返回：Promise<Event[]>

subscribeWithReplay(topic, cb)

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

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

subscribeFromOffset(topic, offset, cb)

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

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

= 256

在 overflow 处理开始前 buffer 可保留的最大事件数。标记为 immediate 的事件不会在 overflow 时丢弃。

overflow?:

"drop-oldest" | "drop-newest" | "coalesce-or-drop-oldest"

= coalesce-or-drop-oldest

buffer 超过 maxBufferSize 时的 overflow 策略。coalesce-or-drop-oldest 会先运行 coalesce，如仍超出预算则丢弃最早事件。

EventCallback

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

event:

Event

已投递的事件。

ack?:

() => Promise<void>

确认处理成功。事件将从队列中移除。

nack?:

() => Promise<void>

否定确认。事件将在延迟后重新加入队列以重新投递。