# @cloudamqp/amqp-client 中文文档

规范地址：https://npmcn.com/package/%40cloudamqp/amqp-client/

版本：4\.1\.2
npm 元数据更新时间：2026-10-19T07:35:14.644Z
中文译文更新时间：2026-11-10T02:33:28.660Z
软件包许可：Apache-2\.0

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40cloudamqp/amqp-client
原文对照：https://npmcn.com/package/%40cloudamqp/amqp-client/?lang=original
项目主页：https://github.com/cloudamqp/amqp-client.js#readme
源码仓库：https://github.com/cloudamqp/amqp-client.js

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：1.1 MB，167 个文件
- 模块格式：ESM + CJS
- TypeScript 类型：包内自带
- Node.js 要求：\>=18\.0\.0
- 运行依赖：0 个
- 首次发布：2021-02-16T21:55:01.825Z
- 最新版本发布：2026-10-19T07:35:14.265Z
- 近 30 天下载量：372202（2026-10-10 至 2026-11-08，npm 官方下载统计）
- 已发布版本数：41（近 12 个月 6 个）

## 中文简介

AMQP 0-9-1客户端,用于RabbitMQ,LavinMQ等AMQP经纪人,既用于浏览器(WebSocket)也用于节点(TCP Socket)

## 安装

```sh
npm install @cloudamqp/amqp-client
```

也可以使用：`pnpm add @cloudamqp/amqp-client`、`yarn add @cloudamqp/amqp-client`、`bun add @cloudamqp/amqp-client`

## README

# amqp - 客户端.js

AMQP 0-9-1 TypeScript客户端既用于Node.js也用于浏览器(使用WebSocket). 这个图书馆旨在取代所有其他的Node.js AMQP图书馆.

它与任何AMQP 0-9-1的经纪人:RabbitMQ,LavinMQ和Apache Qupid合作,无论是自办还是由CloudAMQP或Amazon MQ等提供商管理——他们都说出相同的协议,所以相同的代码会跑来反对他们所有人.

[API 文档](https://cloudamqp.github.io/amqp-client.js/)。 。 。 .

这个图书馆以承诺为基础,因此与Aync/awa合作得很好。 默认情况下,它的安全性,例如,在数据发送到电线上(从而尊重回放),或者如果频道已经启用了Publish Environments,在服务器承认消息被排入后,它就不会实现.

库的开发是为了方便已经熟悉AMQP的开发者在WebSocket上直接与AMQP服务器通信的浏览器应用程序.

## 支助

图书馆由下列机构开发和支持: [云梦QP](https://www.cloudamqp.com/)世界上最大的兔子MQ供应商.

## 安装

```shell
npm install @cloudamqp/amqp-client --save
```

开始节点 `--enable-source-maps` 当库从 TypeScript 中转出时,即可获取适当的堆栈跟踪.

## 示例用途

这个库提供了两个API:

-   **高级别** (单位:千美元)`AMQPSession`:自动重联,消费者恢复——使用 `queue()` 编号 : `exchange()` 用于重新连接安全操作的处理器
-   **低级** (单位:千美元)`AMQPClient` 编号 : `AMQPWebSocketClient`: 直接通道访问与 `queueDeclare`, (中文(简体) ), `basicPublish`, (中文(简体) ), `basicConsume`等 (中文(简体) ).

### 高级API(建议)

使用 `AMQPSession.connect(url, options)` 以获得一个自动重联和消费者恢复的会话。 从 URL 方案中选择传输(`amqp://` 编号 : `amqps://` - TCP; `ws://` 编号 : `wss://` – WebSocket: (英语)

```javascript
import { AMQPSession } from "@cloudamqp/amqp-client"

const session = await AMQPSession.connect("amqp://localhost")

// Declare a queue and publish a message (waits for broker confirmation)
const q = await session.queue("my-queue")
await q.publish("Hello World", { deliveryMode: 2 })

// Subscribe with a callback — consumer recovers automatically on reconnect.
// Messages are acked after the callback returns. If it throws, the message is
// nacked and requeued. Call msg.ack() / msg.nack() yourself to override.
const sub = await q.subscribe(async (msg) => {
  console.log(msg.bodyString())
})

// Or subscribe with an async iterator — messages are acked when the loop advances.
// Call msg.ack() / msg.nack() before the next iteration to override.
const iterSub = await q.subscribe()
for await (const msg of iterSub) {
  console.log(msg.bodyString())
}

// Exchanges work the same way
const x = await session.topicExchange("events")
await x.publish("user signed up", { routingKey: "events.user.created" })

// When done
await session.stop()
```

#### 自动编码/解码(代码)

选择到一个编码器注册,会话会根据 `content-type` 财务报告和审定财务报表 `content-encoding`。通过 `builtinParsers` (JSON,文本)和 `builtinCoders` (gzip, deflate),和可选设置的默认,以免每次发布时重复:

```javascript
import { AMQPSession, builtinParsers, builtinCoders } from "@cloudamqp/amqp-client"

const session = await AMQPSession.connect("amqp://localhost", {
  parsers: builtinParsers,
  coders: builtinCoders,
  defaultContentType: "application/json", // applied when publish() doesn't set one
  defaultContentEncoding: "gzip", // body is gzip-compressed on the wire
})

const q = await session.queue("jobs")

// Object is serialized to JSON and gzip-compressed automatically
await q.publish({ task: "resize", id: 42 })

// On consume, msg.body is the decoded value — no JSON.parse, no decompression
const sub = await q.subscribe(async (msg) => {
  console.log(msg.body.task) // "resize"
})
```

`msg.body` 实值 `Uint8Array` 当没有配置解码器和解码值时 `CodecMode` 一般在编译时推断 。 每份出版物通过取代 `contentType` 编号 : `contentEncoding` 在选项中。 通过与内置合并来注册自己的编码器 :

```javascript
const session = await AMQPSession.connect("amqp://localhost", {
  parsers: { ...builtinParsers, "text/csv": csvParser },
  coders: { ...builtinCoders, lz4: lz4Coder },
})
```

解码失败表面为平面 `Error` (无) `AMQPError`; 内 `subscribe()` 他们打听消息,尊重 `requeueOnNack`。 。 。 .

#### 重新连接选项

```javascript
const session = await AMQPSession.connect("amqp://localhost", {
  reconnectInterval: 1000, // initial delay before reconnecting (ms)
  maxReconnectInterval: 30000, // maximum delay between attempts (ms)
  backoffMultiplier: 2, // exponential backoff multiplier
  maxRetries: 0, // 0 = infinite retries
  onconnect: () => console.log("Connected"), // fires on initial connect + each reconnect
  ondisconnect: (err) => console.warn("Disconnected:", err?.message),
  onfailed: (err) => console.error("Gave up:", err?.message),
})
```

#### 消费者恢复

通过电子邮件创建的订阅 `queue.subscribe()` 重新连接后自动重建。 包含 `prefetch` 在订阅参数中设置每个连接的QoS:

```javascript
const q = await session.queue("my-queue", { durable: true })
const sub = await q.subscribe({ prefetch: 10 }, async (msg) => {
  // process msg — acked on return, nacked and requeued on throw
})

// sub.consumerTag and sub.channel reflect the current consumer (updated on reconnect)
// await sub.cancel()  // stops consuming and removes from auto-recovery
```

#### RPC (远程程序调用)

会议利用直接回复功能提供内部的RPC支持:

```javascript
import { AMQPSession } from "@cloudamqp/amqp-client"

const session = await AMQPSession.connect("amqp://localhost")

// Start an RPC server that listens on a queue
const server = await session.rpcServer("my_rpc_queue", async (msg) => {
  return `processed:${msg.bodyString()}`
})

// Simple RPC call — creates a temporary client per call
const reply = await session.rpcCall("my_rpc_queue", "hello", { timeout: 5000 })
console.log(reply.bodyToString()) // "processed:hello"

// For high-throughput scenarios, reuse a client to avoid per-call channel overhead
const rpc = await session.rpcClient()
const r1 = await rpc.call("my_rpc_queue", "a")
const r2 = await rpc.call("my_rpc_queue", "b")
await rpc.close()

await session.stop() // closes all RPC clients, servers, and consumers
```

RPC 服务器和可重复使用的客户端通过 `session.rpcClient()` 重新连接后自动恢复.

### 低级 API

为了全面控制渠道和资源,直接使用运输客户:

```javascript
import { AMQPClient } from "@cloudamqp/amqp-client"

const amqp = new AMQPClient("amqp://localhost")
const conn = await amqp.connect()
const ch = await conn.channel()

// Declare a queue
const q = await ch.queueDeclare("my-queue")

// Publish
await ch.basicPublish("", q.name, "Hello World", { deliveryMode: 2 })

// Consume with a callback
const consumer = await ch.basicConsume(q.name, { noAck: false }, async (msg) => {
  console.log(msg.bodyToString())
  await msg.ack()
  await consumer.cancel()
})
await consumer.wait()

// Or consume with an async iterator
const consumer = await ch.basicConsume(q.name, { noAck: false })
for await (const msg of consumer.messages) {
  console.log(msg.bodyToString())
  await msg.ack()
  break // breaking automatically cancels the consumer
}

await conn.close()
```

## 网络软件

这个库可以在浏览器中用于通过WebSockets访问一个AMQP服务器. 对于不支持WebSockets本土a的RabbitMQ等服务器而言 [WebSocket TCP 中继器](https://github.com/cloudamqp/websocket-tcp-relay/) 必须用作代理。 所有 CloudAMQP 服务器都配置了这个代理服务器 。 更多信息可以找到 [在博客文章中](https://www.cloudamqp.com/blog/cloudamqp-releases-amqp-websockets.html)。 。 。 .

对于网页浏览器 a [编译](https://www.typescriptlang.org/) 财务报告和审定财务报表 [卷起来](https://www.rollupjs.org/) 版本可用于 [https://github.com/cloudamqp/amqp-client.js/releases (中文(简体) )](https://github.com/cloudamqp/amqp-client.js/releases)。 。 。 .

`AMQPSession` 与 WebSocket URL 一起在框外工作 – 通过 a `ws://` 或者说 `wss://` 自动选择 URL 和运输 :

```javascript
import { AMQPSession } from "@cloudamqp/amqp-client"

const session = await AMQPSession.connect("wss://my.cloudamqp.com/ws/", {
  vhost: "my-vhost",
})
const q = await session.queue("my-queue")
const sub = await q.subscribe({ noAck: true }, (msg) => {
  console.log(msg.bodyString())
})
```

对于没有重联的下级控制,使用 `AMQPWebSocketClient` 直接 :

```html
<!DOCTYPE html>
<html>
  <head>
    <script type="module">
      import { AMQPWebSocketClient } from "./js/amqp-websocket-client.mjs"

      const textarea = document.getElementById("textarea")
      const input = document.getElementById("message")

      const tls = window.location.protocol === "https:"
      const url = `${tls ? "wss" : "ws"}://${window.location.host}`
      const amqp = new AMQPWebSocketClient(url, "/", "guest", "guest")

      async function start() {
        try {
          const conn = await amqp.connect()
          const ch = await conn.channel()
          attachPublish(ch)
          const q = await ch.queueDeclare("")
          await ch.queueBind(q.name, "amq.fanout", "")
          const consumer = await ch.basicConsume(q.name, { noAck: false }, (msg) => {
            console.log(msg)
            textarea.value += msg.bodyToString() + "\n"
            msg.ack()
          })
        } catch (err) {
          console.error("Error", err, "reconnecting in 1s")
          disablePublish()
          setTimeout(start, 1000)
        }
      }

      function attachPublish(ch) {
        document.forms[0].onsubmit = async (e) => {
          e.preventDefault()
          try {
            await ch.basicPublish("amq.fanout", "", input.value, {
              contentType: "text/plain",
            })
          } catch (err) {
            console.error("Error", err, "reconnecting in 1s")
            disablePublish()
            setTimeout(start, 1000)
          }
          input.value = ""
        }
      }

      function disablePublish() {
        document.forms[0].onsubmit = (e) => {
          alert("Disconnected, waiting to be reconnected")
        }
      }

      start()
    </script>
  </head>
  <body>
    <form>
      <textarea id="textarea" rows="10"></textarea>
      <br />
      <input id="message" />
      <button type="submit">Send</button>
    </form>
  </body>
</html>
```

## 业绩

带有 1字节正文的信件, 无属性 :

| 客户端 | 出版率 | 购买率 |
| --- | --- | --- |
| amqp - 客户端.js | 300 000毫秒/秒 | 512.000毫秒/秒 |
| amqplib 语录 | 172.000毫秒/秒 | 519.000 msgs/ss(每秒) |

带有一字节正文的信件,以及除标题外的所有属性:

| 客户端 | 出版率 | 购买率 |
| --- | --- | --- |
| amqp - 客户端.js | 144.000毫秒/秒 | 202,000毫秒/秒 |
| amqplib 语录 | 110 000毫升/秒 | 251,000毫秒/秒 |

带有一字节正文的信息,以及所有属性,包括标题:

| 客户端 | 出版率 | 购买率 |
| --- | --- | --- |
| amqp - 客户端.js | 70 000毫升/秒 | 89 000毫升/秒 |
| amqplib 语录 | 60 000毫秒/秒 | 99.000毫秒/秒 |

Amqp客户端消费速度较慢的原因是,为了维持websocket客户端的浏览器兼容性, [`DataView`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/DataView) 用于解析二进制协议而不是 [`Buffer`](https://nodejs.org/api/buffer.html)。 。 。 .

模块比较

| 客户端 | 运行时间依赖关系 | [代码行](https://github.com/AlDanial/cloc) |
| --- | --- | --- |
| amqp - 客户端.js | 0 个 | 1743 (英语) |
| amqplib 语录 | 页:1 | 6720 (w/o 依赖性) |

## 释放

这个项目使用自动发布脚本进行版本管理.

### 发布命令

Package.json包含数个npm脚本用于放出:

-   **`npm run release`** - 执行补丁版本的凸起(例如3.2.1-3.2.2)并创建发布
-   **`npm run release:minor`** - 执行一个小版本的凸起(例如3.2.1 → 3.3.0)并创建一个发布
-   **`npm run release:major`** - 执行一个大版本的凸起(例如,3.2.1-4.0.0)并创建发行版

### 释放期间会发生什么

1.  **测试**:所有测试都进行以确保一切通过(`preversion`(中文(简体) )
2.  **版本缓冲**: npm 自动更新该版本 `package.json`
3.  **文件更新**: 该版本更新于 `src/amqp-base-client.ts`,代码格式化,更改日志更新(`version`(中文(简体) )
4.  **粘贴**: 所有更改都是为承诺而设(`version`(中文(简体) )
5.  **Git 执行**: npm 自动将所有已分阶段的更改带入版本消息
6.  **Git 标签**: 创建附加注释的标签, 将全部更改日志内容作为标签信件( Name`postversion`(中文(简体) )
7.  **推开**: 承诺和标记都被推入远程存储器(`postversion`(中文(简体) )
8.  **CI 部署**: GitHub 动作工作流程自动将新版本发布到 npm

> **技术说明**:这个放出过程会利用 npm 的内置版本生命周期钩 (`preversion`, (中文(简体) ), `version`, (中文(简体) ), `postversion`) (中文(简体) ). 这个 `npm version` 命令在运行自定义后自动处理 git 承诺 `version` 剧本,这就是为什么我们 舞台改变与 `git add -A` 而不是手动承诺.

### 先决条件

在释放前:

1.  将您的更改添加到 `## [Unreleased]` 节 `CHANGELOG.md`
2.  所有测试均应通过(`npm test`(中文(简体) )
3.  工作目录应清洁(无未承诺的更改)

### 自动 npm 出版

当新标签被推出( 例如, ), `v3.3.0`), GitHub 动作工作流程(`.github/workflows/release.yml`自动 :

-   构建工程
-   将软件包发布到 npm, 并公开访问和来源
-   创建带有浏览器捆绑文物的 GitHub 发布

Git标签包含该版本的完整更改日志部分,包括版本头,所有更改,以及PR链接. 这使得人们很容易看到每一版直接从git标签中的变化.

## 常见问题

### @cloudamqp/amqp-client 是什么？

AMQP 0-9-1客户端,用于RabbitMQ,LavinMQ等AMQP经纪人,既用于浏览器\(WebSocket\)也用于节点\(TCP Socket\)

### 如何安装 @cloudamqp/amqp-client？

在项目目录运行 npm install @cloudamqp/amqp-client，或使用 pnpm add / yarn add / bun add。

### @cloudamqp/amqp-client 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @cloudamqp/amqp-client 支持 ESM 还是 CommonJS？

同时提供 ES Module 与 CommonJS 入口，import 和 require\(\) 都可以使用，打包工具会自动选择合适的版本。

### @cloudamqp/amqp-client 需要什么版本的 Node\.js？

package\.json 的 engines 字段要求 Node\.js \>=18\.0\.0。版本不符时 npm 默认只给出警告，开启 engine-strict 后会拒绝安装。

### @cloudamqp/amqp-client 可以商用吗？

@cloudamqp/amqp-client 使用 Apache-2\.0 许可。宽松许可：允许商业使用、修改和再分发，并附带明确的专利授权；分发时需保留许可与 NOTICE 文件，修改过的文件要注明改动。以上为概括，不构成法律意见，请以许可原文为准。

### @cloudamqp/amqp-client 的下载量有多少？

根据 npm 官方下载统计，@cloudamqp/amqp-client 在 2026/10/10 – 2026/11/08 这 30 天内被下载 372,202 次，日均约 12,407 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @cloudamqp/amqp-client 还在维护吗？

最新版本 4\.1\.2 发布于 2026/10/19（21 天前）；至今共发布 41 个版本，其中过去 12 个月发布了 6 个；首次发布于 2021/02/17。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。