PubSub

Mastra 使用发布/订阅（pub/sub）系统作为其内部事件总线。组件将事件发布到主题，其他组件订阅这些主题以作出反应。后端决定了事件是在单个进程内、同一主机的进程之间，还是在不同实例之间传递。

在 Mastra 实例上设置一次后端，即可在整个系统中使用。默认是无需任何设置的进程内后端。

Mastra 如何使用 PubSub

多个内置系统通过同一 pub/sub 总线发布和订阅事件：

• 工作流执行：调度器发布 workflow.start 事件，长期运行的 worker 消费它以运行工作流。执行期间，步骤和生命周期事件通过 pub/sub 流动。
• 定时工作流：调度器通过发布到工作流主题来分发到期运行。参见定时工作流。
• 后台任务：任务管理器将工作分发到 worker 组，并向订阅者扇出任务生命周期更新，后台任务流因此保持实时。参见后台任务流式传输。
• Agent 信号：向活动的 agent 运行发送信号会在 thread 主题上发布事件，因此另一个进程中执行的运行也能收到该信号。
• 可恢复流：流块按运行发布，因此重新连接的客户端可以重放错过的内容。

由于这些系统运行在同一条总线上，你选择的后端会同时应用于所有这些系统。

传递模式

后端以两种模式之一传递事件，由 PubSub 契约定义：

• 拉取（Pull）：消费者自行从后端读取，Mastra 通过长期运行的 worker 循环执行此操作。诸如 RedisStreamsPubSub 的分布式后端使用此模式。
• 推送（Push）：事件无需消费者请求即可到达，可在进程内或通过 HTTP 传递。默认的 EventEmitterPubSub 以这种方式在进程内传递。

订阅者还可以通过消费者组分发工作。同一组的成员会拆分事件，使每个事件只被处理一次。没有组的订阅者会接收每个事件，从而将流扇出到所有无组订阅者。

默认后端

当你未设置 pubsub 选项时，Mastra 使用 EventEmitterPubSub。它使用 Node.js EventEmitter 在进程内传递事件，因此无需任何外部服务即可工作。

src/mastra/index.ts

import { Mastra } from '@mastra/core'

// No pubsub option: Mastra uses EventEmitterPubSub
export const mastra = new Mastra({})

由于它在进程内运行，事件不会被持久化，也不会到达其他进程。默认后端适用于单个实例，这覆盖了大多数应用。

选择后端

在 Mastra 实例上设置 pubsub 选项以选择后端。每个后端都实现相同的 PubSub 契约，因此应用的其他部分无需更改。

决定性的问题是订阅者在哪里运行。

后端	范围	模式	包
EventEmitterPubSub	单进程	拉取和推送	@mastra/core
UnixSocketPubSub	多进程，单主机	推送	@mastra/core
RedisStreamsPubSub	分布式，多主机	拉取	@mastra/redis-streams
GoogleCloudPubSub	分布式，多主机	拉取	@mastra/google-cloud-pubsub

同一主机上的多个进程

当同一台机器上的多个进程需要共享一条流时，请使用 UnixSocketPubSub。它通过 Unix 域套接字传递事件，并选举一个进程作为 broker。如果 broker 退出，其余进程会选举新的 broker。

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { UnixSocketPubSub } from '@mastra/core/events'

export const mastra = new Mastra({
  pubsub: new UnixSocketPubSub('/tmp/mastra/events.sock'),
})

分布式部署

当你运行多个实例或主机，且每个实例都需要接收相同事件时，请使用分布式后端。只要一个实例处理的请求必须到达另一个实例上运行的工作，这一点就很重要。

例如，向 agent 运行发送信号需要信号事件跨越进程边界到达拥有该运行的实例。使用进程内默认后端时，该实例永远收不到事件。

当存储适配器不支持运行隔离（run fencing）时，持久化 agent 的崩溃恢复也依赖后端。此时恢复使用后端的 lease 来判断另一个实例是否仍在驱动某个运行。RedisStreamsPubSub 提供跨主机的 lease，GoogleCloudPubSub 则不提供。

下面两个后端都能跨进程和主机传递事件，并持久化事件以便重新投递。

以下示例使用 Redis Streams：

src/mastra/index.ts

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

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

以下示例使用 Google Cloud Pub/Sub：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { GoogleCloudPubSub } from '@mastra/google-cloud-pubsub'

export const mastra = new Mastra({
  pubsub: new GoogleCloudPubSub({
    projectId: 'my-project',
  }),
})

可恢复流

可恢复流让客户端重新连接并重放错过的事件，这需要后端保留最近历史。诸如 RedisStreamsPubSub 的分布式后端会持久化事件，因此本身支持重放。

进程内传递不保留历史。要在 EventEmitterPubSub 之上添加重放能力，请将其包装在 CachingPubSub 中，后者按主题记录已发布事件，以便迟到或重新连接的订阅者可以在继续接收实时事件之前赶上进度。

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { CachingPubSub, EventEmitterPubSub } from '@mastra/core/events'
import { InMemoryServerCache } from '@mastra/core/cache'

const cache = new InMemoryServerCache()

export const mastra = new Mastra({
  pubsub: new CachingPubSub(new EventEmitterPubSub(), cache),
})

完整的传递契约以及每个后端的配置选项，请访问 PubSub 参考。

相关内容

• PubSub 参考
• Mastra 类
• Workers：使用 PubSub 在专用进程中运行工作流编排和后台任务
• 后台任务流式传输
• 定时工作流