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

# GoogleCloudPubSub

`GoogleCloudPubSub` 是由 [Google Cloud Pub/Sub](https://cloud.google.com/pubsub/docs) 支持的 [`PubSub`](https://mastra.zisheng.pro/reference/pubsub/base) 实现。它使用 Google Cloud topic 和 subscription 跨进程及主机投递事件，并提供有序投递和消息确认。

将它用于 Google Cloud 上的分布式部署。对于单进程投递，请使用 [`EventEmitterPubSub`](https://mastra.zisheng.pro/reference/pubsub/event-emitter)。对于 Redis，请使用 [`RedisStreamsPubSub`](https://mastra.zisheng.pro/reference/pubsub/redis-streams)。

每个 topic 对应一个 Google Cloud topic。带 group 的订阅共享一个 subscription，因此成员会竞争事件。没有 group 的订阅会创建每实例 subscription，因此每个实例都会收到每个事件。

## 安装

**npm**:

```bash
npm install @mastra/google-cloud-pubsub
```

**pnpm**:

```bash
pnpm add @mastra/google-cloud-pubsub
```

**Yarn**:

```bash
yarn add @mastra/google-cloud-pubsub
```

**Bun**:

```bash
bun add @mastra/google-cloud-pubsub
```

## 使用示例

传入 Google Cloud client 配置，例如项目 ID。

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

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

## 构造函数参数

**config** (`ClientConfig`): Google Cloud Pub/Sub client 的配置，包括凭据和项目 ID。有关所有字段，请参阅 @google-cloud/pubsub client 文档。

## 方法

`GoogleCloudPubSub` 实现 [`PubSub`](https://mastra.zisheng.pro/reference/pubsub/base) contract。以下方法是此实现特有的。

### `init(topicName, group?)`

如果 topic 和 subscription 尚不存在，则创建它们，并返回 subscription。`subscribe` 会在内部调用此方法，因此很少需要直接调用。

```typescript
await pubsub.init('workflow.events')
```

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

订阅一个 topic。使用 `options.group` 时，组成员共享 subscription 并竞争事件。没有 group 时，该实例会通过自己的 subscription 收到每个事件。

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

### `flush()`

等待待处理确认完成。

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

### `destroy(topicName)`

移除一个 topic 名称对应的 subscription 和 topic。使用此方法清理 Google Cloud 资源。

```typescript
await pubsub.destroy('workflow.events')
```

## 确认

每个已投递事件均包含 `ack` 和 `nack` 函数。成功处理后调用 `ack`，以从 subscription 移除事件。当两者都未调用时，Google Cloud 会在其确认截止时间过期后重新投递该事件。