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 文档。 。 。 .

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

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

支助

图书馆由下列机构开发和支持: 云梦QP世界上最大的兔子MQ供应商.

安装

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: (英语)

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),和可选设置的默认,以免每次发布时重复:

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 在选项中。 通过与内置合并来注册自己的编码器 :

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

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

重新连接选项

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:

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支持:

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

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

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 中继器 必须用作代理。 所有 CloudAMQP 服务器都配置了这个代理服务器 。 更多信息可以找到 在博客文章中。 。 。 .

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

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

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 直接 :

<!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客户端消费速度较慢的原因是,为了维持websocket客户端的浏览器兼容性, DataView 用于解析二进制协议而不是 Buffer。 。 。 .

模块比较

释放

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

发布命令

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 标签: 创建附加注释的标签, 将全部更改日志内容作为标签信件( Namepostversion(中文(简体) )
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标签中的变化.