DroidSolutions amqp- ts (AMQP TypeScript) (中文(简体) )

这是叉子 秒数。 。 。 .

目录

• 概览
• 使用量
• 从 3.x 移动

概览

Amqp-ts是一个用于节点的库,用于简化与用Typescript写成的AMQP消息总线的通信. 已在RabbitMQ上测试过. 它使用 amqplib 语录 库 迈克尔·布里奇恩(squaremo)。 。 。 .

与原始图书馆的差异

变动 原声调 软件包 :

• 使用最新数据 amqplib 语录 (很不幸,原始软件包使用了只支持已过时的节点JS版本的四年版本)
• 已过时的NodeJS版本已放弃支持,4版本需要Node.js 22.13或更新,并作为ES模块发布.
• dropp winston 以依赖性形式,相反,可以向返回任何日志的连接构造器提供工厂.
• 贬值的方法 Exchange.publish, (中文(简体) ), Exchange.startConsumer, (中文(简体) ), Queue.publish 财务报告和审定财务报表 Queue.startConsumer 被从这个图书馆中移除,因为他们的替代者喜欢 Exchange.send 财务报告和审定财务报表 Exchange.activateConsumer 已经存在.
• 用于 https://github.com/abreits/amqp-ts/pull/46 (英语) 财务报告和审定财务报表 https://github.com/abreits/amqp-ts/issues/20 互联网档案馆的存檔,存档日期2013-12-02 已执行
• 添加基本衡量标准

特征定义

• [高层无意见文库],不必担心频道等.
• '懒'初始化,自动解决 AMQP 依赖性
• 自动重接,当与 AMQP 服务器的连接失败时,整个连接和配置会自动重建(使用 amqplib 2 的连接恢复)
• 以打字符写出 .

目前情况

图书馆被视为已准备好.

它取决于以下 npm 库 :

• amqplib 语录

懒惰初始化

不需要嵌入功能,只需创建连接,声明您的交换,队列和 绑定并发送和接收信件。 图书馆负责任何直接的依赖.

如果您定义了交换和队列,并将队列绑定在交换中,并且想要在向交换发送消息之前确保队列与交换连接,您可以调用 connection.completeConfiguration() 并按承诺行事 这也是你开始在地形学上申请的方法 启动时等待地形。 。 。 .

使用量

安装软件包( 它需要节点. js) >=22.13.0 并带来自己的打字,不 @types/amqplib 所需:

npm install @droidsolutions-oss/amqp-ts

该包是一个ES模块. 使用以下所有示例 import 高级 await,因此将它们作为ES模块运行(.mjs 文档,或 "type": "module" 在你身边 package.json) (中文(简体) ). 从软件包根导出所有东西; 见 从 3.x 移动 如果你使用普通JS的话.

有多种方式可以声明您的交换,队列和绑定. 每一个都有一个 initialized 当初始化完成后,它就会解决。 你可以自己等,也可以打电话 completeConfiguration 关于连接,如解释 懒惰初始化。 。 。 .

Async 键盘示例

import { Connection, Message } from "@droidsolutions-oss/amqp-ts";

const connection = new Connection("amqp://localhost");

// publish
const exchange = await connection.declareExchangeAsync("ExchangeName");

const msg = new Message("Test");
exchange.send(msg);

// subscribe/consume
const queue = await connection.declareQueueAsync("QueueName");
await queue.bind(exchange);
await queue.activateConsumer((message: Message) => {
  console.log("Message received: " + message.getContent());
});

懒惰打字示例

import { Connection, Message } from "@droidsolutions-oss/amqp-ts";

const connection = new Connection("amqp://localhost");
const exchange = connection.declareExchange("ExchangeName");
const queue = connection.declareQueue("QueueName");
void queue.bind(exchange);
void queue.activateConsumer((message: Message) => {
  console.log("Message received: " + message.getContent());
});

// it is possible that the following message is not received because
// it can be sent before the queue, binding or consumer exist
const msg = new Message("Test");
exchange.send(msg);

// This waits until all exchanges/queues/bindings are initialized
await connection.completeConfiguration();
// the following message will be received because
// everything you defined earlier for this connection now exists
const msg2 = new Message("Test2");
exchange.send(msg2);

更多实例可见于 教程目录。 。 。 .

信件

一个消息包含内容(您想要发送的数据)和一些由AMQP定义的可选属性. 这种元数据为消费者和提供者提供了一些方法,让他们知道如何处理信息。 例如,当您提供一个对象作为内容时,它将被转换为 JSON 字符串和 contentType 属性设定为 application/json消费者知道应该从Json那里分析信息。 关于属性的详细指南,可以参考: 兔子MQ文档。 。 。 .

此库设置的属性为:

• 内容 Type - 当您给信件构建器除缓冲和字符串以外的任何东西时
• 内容编码 - 当您给信件构建器除缓冲和字符串以外的任何东西时( 目前总是) utf-8(中文(简体) )
• 时间戳 - 如果您没有设置, 它将设置为 UTC 中的当前时间戳
• 相关性Id - 如果被收到的信息所指定,将设定为自动回复(见下文)

日志

最后的论点 Connection 构造器需要一个伐木厂的功能. 如果您指定了, 函数应当返回一个匹配 SimpleLogger 接口。 这种类型在很大程度上受到 皮诺但是,你可以为任何喜欢的日志写一个包装. 它应支持 %s, (中文(简体) ), %d 财务报告和审定财务报表 %o 字符串插值,因为用于内部日志消息.

如果您没有指定伐木厂, 则不会释放日志 。 这包括您自己的事件听众所抛出的错误,这些错误会在错误级别被捕获并记录下来(参见 活动)和必须删除的信息(见 发送信件) (中文(简体) ). 所以在生产过程中通过一个伐木厂,否则这些问题是看不见的.

皮诺语的用法示例:

const logger = pino({
  name: "amqp-ts-using-app",
  level: "info",
  formatters: {
    level: (label, _number) => {
      return { level: label };
    },
  },
  redact: [],
  serializers: { err: pino.stdSerializers.err },
});

loggerFactory = (context, meta) => {
  if (!meta) {
    meta = {};
  }
  const className = typeof context === "string" ? context : context.name;
  meta.context = className;

  return logger.child(context, meta);
};

const connection = new Connection("amqp://localhost", {}, {}, loggerFactory);

上面的例子将导致这样的日志消息:

{
  "level": "info",
  "time": 1586871242321,
  "pid": 80174,
  "hostname": "some-machine",
  "name": "amqp-ts-using-app",
  "module": "amqp-ts",
  "context": "Connection",
  "msg": "Connection established."
}

{
  "level": "error",
  "time": 1586871243860,
  "pid": 80174,
  "hostname": "some-machine",
  "name": "amqp-ts-using-app",
  "exchange": "TestExchange_33",
  "context": "Exchange",
  "err": {
    "type": "Error",
    "message": "Operation failed: ExchangeDeclare; 404 (NOT-FOUND) with message \"NOT_FOUND - no exchange 'TestExchange_33' in vhost '/'\"",
    "stack": "Error: Operation failed: ExchangeDeclare; 404 (NOT-FOUND) with message \"NOT_FOUND - no exchange 'TestExchange_33' in vhost '/'\"\n    at reply ...",
    "code": 404,
    "classId": 40,
    "methodId": 10
  },
  "msg": "Failed to create exchange 'TestExchange_33'."
}

注意: 要在日志中出现错误, 您必须像给定的一样为 Pino 使用序列器 pino.stdSerializers.err 用于以上.

这个方法的优点是,你会看到准确的是谁抛出错误以及附加上下文,比如无法宣布交换. 也可以用另一个记录器 winston数字 :

const logger = new winston.Logger({
  transports: [
    new winston.transports.Console({
      level: process.env.AMQPTS_LOGLEVEL || "error",
      format: winston.format.combine(
        winston.format.splat(), // needed for string interpolation
        // ... any additional formats
      ),
    }),
  ],
});

const loggerFactory = (context, meta) => {
  return {
    info(metaOrMsg: object | string, msg?: string, ...args: any[]): void {
      if (typeof metaOrMsg === "string") {
        logger.info(metaOrMsg, ...args);
      } else {
        logger.info(msg, ...args);
        logger.info("additional meta %o", metaOrMsg);
      }
    },
    /// implement other methods like this ...
  };
};

连接状态

了解连接状况: connection.isConnected。如果连接存在且不正确,则返回真实.

connection.getConnection 返回当前 amqplib 语录 ChannelModel,或 undefined 当连接关闭时( 在第一次连接之前, 在重新连接期间和之后) close()) (中文(简体) ). 每次重新连接时都会更换,所以不要提及它。 通讯应该通过这个库的交换和排队,这些库在重联中生存.

connection.initialized 是一个承诺,一旦建立联系和建立已宣布的地貌,就能够解决。 虽然连接在下降,但被下一个成功重新连接后解决的待定承诺所取代。 它拒绝当 close() 或恢复之前放弃。 也一样 initialized 承诺交换、排队、捆绑和排队消费者(activateConsumer他们不信 close() 当恢复停止,当诵读 initialized 一实体在 close() 返回一个被拒绝的承诺.

connection.initialized 只反映连接。 单独失败的交换、队列、绑定或消费者(例如 noCreate 队列不存在,或经纪人拒绝的消费者)拒绝自己的承诺,并通过报告进行 error_connection但它没有保留 connection.initialized 从解析,所以一个被破坏的实体不能把整个连接拆掉.

启动时等待地形

只开始你宣布的一切工作,等待 connection.completeConfiguration()。它等待每个宣布的交换,队列和绑定被初始化,每个激活的消费者被注册,当其中之一失败时,在第一次连接时或之后,它拒绝. 它拒绝只要失败实体不再被宣布(一个被拒绝的消费者:直到下一次重新连接注册),所以再次询问不会隐藏失败.

import { Connection, Message } from "@droidsolutions-oss/amqp-ts";

const connection = new Connection("amqp://localhost");
const queue = connection.declareQueue("orders", { durable: true });
void queue.bind(connection.declareExchange("events", "topic"), "order.*");
void queue.activateConsumer((message: Message) => console.log("order", message.getContent()));

try {
  await connection.completeConfiguration();
} catch (err) {
  console.error("the topology could not be set up", err);
  await connection.close();
  process.exit(1);
}

发送信件

exchange.send(message, routingKey) 财务报告和审定财务报表 queue.send(message) 返回时 void。在连接或交换/排队准备就绪之前发送的信息,一旦得到缓冲和发送。 连接或信道丢失或关闭时发送的信息也是如此,例如amqplib在协议出错后关闭连接:一旦交换/排队再次初始化,就会发送. 无法发送的信息, 因为恢复放弃, close() 被调用,或交换/排队被关闭或被删除,被记录在警告级别并被丢弃。 (在3.x中,发送到一个封闭的实体扔出一个 TypeError. ) (中文(简体) ). 如果你需要知道 留言留下,等待 connection.completeConfiguration() 或 initialized 在发送前答应目标.

活动

这个 Connection 是一个 EventEmitter它揭示了这些事件:

connection.on("open_connection", () => console.log("connected"));
connection.on("lost_connection", () => console.log("connection lost, reconnecting"));
connection.on("re_established_connection", () => console.log("connected again"));
connection.on("error_connection", (err) => console.error("connection problem", err));

注释:

• error_connection 这不是一个“连接是死的”信号。 当只有一个实体失败时,它也会起火,尽管连接本身是好的:例如: noCreate 不存在的交换或队列,其声明与现有队列相冲突,约束经纪人拒绝,经纪人拒绝的消费者,或者再次注册后再次取消。 其中包括第一次申报,还有无人等待的实体申报,以及回收期间重复的申报。 恢复后放弃(无限期) maxRetries它随着放弃错误而起火,不再发生再连接.
• 要知道地形学是否起作用, 等待 completeConfiguration() 而不是听这个事件.
• 您的听众所抛出的例外被捕获并记录在错误级别而不是宣传,失败者之后的听众仍然运行. 默认( 空) 记录器 错误是隐形的, 所以通过 a 伐木厂。 。 。 .

自动重接

当库检测到与AMQP服务器的连接已丢失时,它会自动重接并重建整个地貌(交换,队列,绑定和消费者). 这使用了 amqplib 2. 的连接恢复. 第三个论点 Connection 构造器取 RecoveryOptions数字 :

默认它会永远重复。 当经纪人在启动时无法联系到时快速失败的示例, 但一旦连接就继续重试 :

import { Connection, type RecoveryOptions } from "@droidsolutions-oss/amqp-ts";

const recovery: RecoveryOptions = { initialMaxRetries: 3, initialDelay: 500, maxDelay: 10000 };
const connection = new Connection("amqp://localhost", {}, recovery);
connection.on("error_connection", (err) => {
  // fired at startup after 3 failed attempts, or when a single exchange, queue, binding or consumer fails
  console.error("connection problem", err);
});

await connection.initialized; // rejects if the broker is still unreachable after 3 retries

RecoveryOptions 是 amqplib 类型没有 setup 财务报告和审定财务报表 waitForConnect,它使用自己.

失去连接时会发生什么

• 打开 lost_connection 连接和所有交换、队列、绑定和消费者成为“非初始化”及其 initialized 承诺被尚未兑现的承诺取而代之。 您同时发送的信件是缓冲的 .
• 每次重新连接后, 地形再次按依赖顺序( 交换、 队列、 绑定、 消费者) 宣布 re_established_connection (或 open_connection 第一种联系)将释放出来,有待兑现的承诺将得到解决。 B. 对另一交易所具有约束力 Connection 是例外:它等待该连接初始化了交换,因此它可能在以后创建.
• 如果恢复放弃(无限) maxRetries, (中文(简体) ), error_connection 随放弃错误而起火,每一次待决 initialized 拒绝, 缓冲消息会以警告日志被删除 .
• 没有再连接:当经纪人关闭了单个交换或队列的频道(例如双相板上出错406,或发布给被删除的交换时出错404)时,只有该实体获得一个新的频道,包括其装订和消费者. 被经纪人取消的消费者(例如因其队列被删除)再次被注册. 如果经纪人在第二秒内再次取消,它会一直被取消,直到下一次重新连接 error_connection 已释放.
• 经纪人拒绝的消费者(例如403,因为另一个连接在队列上持有独家消费者)没有在同一连接上重新测试. 在下次重新连接时, 或者在您打电话时, 它再次注册 stopConsumer() 接下来 activateConsumer() 在那之前,他的承诺是: completeConfiguration() 拒绝.

自动 JSON 内容

如果您的消息内容不是字符串,也不是缓冲,则自动转换为JSON字符串. 这个 contentType 设置为 application/json 编辑 contentEncoding 设置为 utf-8。反之亦然,如果你打电话 getContent() 关于消息和 contentType 设置为 application/json 字符串将被解析并返回解析结果。 如果你想避开 any 类型 getJsonContent<T>() 换句话说 如果 contentType 未设定为 application/json。 。 。 .

自动回复和 RPC

如果您的信息消费者返回一个值和 replyTo 属性设置在信件属性中,值发送到给定的队列 replyTo 字段。 如果请求中设定了响应消息,则设定了相关ID. 如果您想要设置其它属性, 您可以在调用时指定它们 activateConsumer 在选项中。 如果您不返回任何值( a. k. a 返回) undefined没有自动发送答复。 那你得自己去办.

您使用此库执行请求/ 答复的方式( RPC) : 请求者发送信件 replyTo (和a) correlationId )并消耗其回复队列,消费者只需返回结果.

实例:

供应商/供应商方面:

import { Connection, Message } from "@droidsolutions-oss/amqp-ts";

const connection = new Connection("amqp://localhost");
const exchange = await connection.declareExchangeAsync("ExchangeName");
const queue = await connection.declareQueueAsync("rpc_reply");
await queue.activateConsumer((message: Message) => {
  console.log("RPC reply Message received: " + message.properties.correlationId); // should log the ID you set in the message below

  const content = message.getJsonContent<{ my: string }>();
  console.log("Result is ", content.my); // "Result is result"
});

// sending the message
const msg = new Message(
  { some: "value" },
  {
    correlationId: "jC7oTFEeWIfmn_c8Z5Aa",
    replyTo: "rpc_reply",
    appId: "app:my-app component:rpc-client",
  },
);
exchange.send(msg, "rpc_queue");

回复队列的消费者是正常的消费者,在重新连接后再次注册. 下来的时候得到的回复 留在队列里.

消费者/消费者方面:

import { Connection, Message } from "@droidsolutions-oss/amqp-ts";

const connection = new Connection("amqp://localhost");
const exchange = await connection.declareExchangeAsync("ExchangeName");
const queue = await connection.declareQueueAsync("rpc_queue");
await queue.bind(exchange);
await queue.activateConsumer((message: Message) => {
  console.log(`Message from ${message.properties.appId} received: ${message.properties.correlationId}`); // "Message from app:my-app component:rpc-client received: jC7oTFEeWIfmn_c8Z5Aa

  const content = message.getJsonContent<{ some: string }>();
  console.log("Content is ", content.some); // "Content is value"
  // process message content

  return { my: "result" };
});

计量

连接实例提供了一个返回连接基本度量的方法. 对于衡量标准,有一个 TypeScript 接口 ConnectionMetrics 包含可能的连接参数.

要接收他们,你可以打电话 getMetrics() 在连接实例.

从 3.x 移动

4.0版本是主要发行版本. 本节列出您需要改变的一切, 以及您可能注意到的行为变化 .

中断更改

• 仅限无害环境管理. 该包是一个ES模块. 在无害环境管理编码中,与以往一样从包根导入(import { Connection } from "@droidsolutions-oss/amqp-ts") (中文(简体) ). 在普通JS代码中,仍然可以加载 require() 在节点上 >=22.13.0 (这可以 require() 没有旗帜的ES模块 :

  const { Connection, Message } = require("@droidsolutions-oss/amqp-ts");
  

  在类型脚本中设置 "module": "node20" 或者说 "nodenext" (配对) moduleResolution所以,这样 require 对一个ES模块的理解. module: "commonjs" 将不会解决新的一揽子方案.

• 节点.js >=22.13.0。 。 。 . 旧版本不支持( engines 字段是这样写的).

• 无入深入. 只有包根导出 . import { Message } from "@droidsolutions-oss/amqp-ts/lib/Message" (和所有其他) lib/... 路径)失败 ERR_PACKAGE_PATH_NOT_EXPORTED。所有公共类别和类型均从根导出: Connection, (中文(简体) ), Exchange, (中文(简体) ), Queue, (中文(简体) ), Binding, (中文(简体) ), Message, (中文(简体) ), ApplicationName 类型 AmqpProperties, (中文(简体) ), ConnectionMetrics, (中文(简体) ), RecoveryOptions, (中文(简体) ), Topology, (中文(简体) ), LoggerFactory, (中文(简体) ), SimpleLogger, (中文(简体) ), ExchangeType, (中文(简体) ), ExchangeDeclarationOptions, (中文(简体) ), QueueDeclarationOptions, (中文(简体) ), ExchangeInitializeResult, (中文(简体) ), QueueInitializeResult, (中文(简体) ), DeleteResult, (中文(简体) ), StartConsumerResult。 。 。 .

• RecoveryOptions 替换 ReconnectStrategy。 。 。 . 第三个 Connection 参数现在是 amqplib 的 RecoveryOptions中。固定间隔被用快感的指数回放所取代.

  // 3.x
  new Connection(url, {}, { retries: 0, interval: 1500 }); // retry forever, every 1.5 s
  new Connection(url, {}, { retries: 5, interval: 1500 }); // give up after 5 attempts
  
  // 4.x
  new Connection(url); // the default retries forever (maxRetries: Infinity) with backoff
  new Connection(url, {}, { maxRetries: 5 }); // give up after 5 attempts
  new Connection(url, {}, { initialDelay: 1500, maxDelay: 1500, factor: 1, jitter: 0 }); // fixed delay, if you really want it
  

  { retries: 0 } 变成默认者, { retries: n } 变成 { maxRetries: n },则 interval 变成 initialDelay编号 :maxDelay编号 :factor编号 :jitter新建为 initialMaxRetries,这使得启动迅速失败,而后恢复会永远地重复. 这个 ReconnectStrategy 类型已消失.

• heartbeat=0 禁用心跳 . amqplib 从连接 URL 的查询读取心跳(amqp://localhost?heartbeat=0),而不是套接字选项。 amqplib 2 处理 heartbeat=0 作为"没有心跳"而不是"使用服务器的值". 如果您的 URL 包含 heartbeat=0 要获取服务器的值,请删除参数.

• Exchange.rpc 财务报告和审定财务报表 Queue.rpc 被删除. rpc 仅以答复方式解决了第一次通话,并在重新通话后失去了答复的消费者。 发送信件 replyTo 财务报告和审定财务报表 correlationId 并输入回复队列; 消费者的返回值自动返回。 见 自动回复和 RPC数字 :

  // 3.x
  const result = await queue.rpc(21);
  console.log(result.getContent());
  
  // 4.x
  const replies = await connection.declareQueueAsync("my_replies", { exclusive: true });
  await replies.activateConsumer((reply) => console.log(reply.getContent()), { noAck: true });
  queue.send(new Message(21, { replyTo: "my_replies", correlationId: "some-unique-id" }));
  

  对请求的答复匹配 correlationId 如果你有一个以上的电话在飞行.

• getConnection 财务报告和审定财务报表 connection。 。 。 . connection.getConnection 现在返回当前 amqplib ChannelModel (或 undefined 而不连接 amqplib Connection,与公众 connection 字段不见了。 使用 getConnection 检查 undefined。 。 。 .

• tryToConnect 删除. 连接是自动建立和重新建立的,没有什么可以调用. 使用 事件 财务报告和审定财务报表 connection.initialized 以反应其状态.

• 打字声. @types/amqplib 不再需要(amqplib ships its type),从您的依赖中移除它。 强调成员_channel, (中文(简体) ), _exchanges, (中文(简体) ), _queues,.)是内部的,不再出现在打字中.

行为变化

• error_connection 范围更广. 除了"恢复放弃"(无限) maxRetries,随着放弃错误),当无法宣布单一交换,排队或绑定,或者消费者无法注册时,无论是在第一次申报时还是在回收期间再次申报时,它也会起火. 连接本身是罚款的,所以不要将事件视为"连接死了".
• 门启动 completeConfiguration()。 。 。 . connection.initialized 财务报告和审定财务报表 open_connection 只反映连接:一个自行失败的实体不会拒绝或拖延它们. connection.completeConfiguration() 拒绝时, 已声明的交换、 队列或绑定失败 或已激活的消费者 也被拒绝 。 见 启动时等待地形。 。 。 .
• deleteConfiguration() 解决 undefined。 。 。 . 它删除了捆绑,再排队,再进行交换,相接相接,并用第一次失败来拒绝. 在3.x中,它一并运行了一切,并以各种结果解决(Promise.all) (中文(简体) ).
• unbind 一个未知的 约束性决定. queue.unbind() 财务报告和审定财务报表 exchange.unbind() 在没有这种约束力时(例如两次无约束力时)解决。 在3.x 他们扔了一个 TypeError。删除仍在创建的约束力,将拒绝其待决 initialized。 。 。 .
• 待决 initialized 承诺拒绝. 每一次待决 initialized (连接、交换、排队、绑定和消费者) close() 而当康复放弃。 读取 initialized 一实体在 close() 返回一个被拒绝的承诺。 添加错误处理, 您正在等待它们。 Name.
• 信件无法发送 . send() 仍然返回 void无法发送的信息(恢复放弃, close() 被调用,交换或队列被关闭或被删除)在警告级别被记录并被丢弃. 在3.x中,发送到一个封闭的实体扔出一个 TypeError。 。 。 .
• 监听员例外记录. 您的事件听众所抛出的例外被捕获并记录在错误级别, 而不是宣传 。 默认的(空)记录器是看不见的,所以通过一个 loggerFactory。 。 。 .
• 捆绑等待他们的源交换. 一旦其源交换通过此库初始化, 就会创建约束, 如果无法声明则失败 。 在 3.x 中,一旦队列准备就绪,当交易在经纪人身上偶然存在时,就立即创建了队列绑定. 在同一连接上声明源交换( :connection.declareExchange(...))在约束之前.
• activateConsumer 如果队列声明失败, 请拒绝 . 返回的承诺与声明错误相悖 .
• 重接时间. 默认为由 jitter (100 ms 初始延迟, 30 s 最大, 系数 2, jitter 0.2) 的指数回放, 而不是固定的 1. 5 s 间隔 .
• 中介-侧道和消费者故障被修复. 仅为该交换或队列重新创建了经纪人关闭的频道(例如双板上),经纪人取消的消费者再次注册,参见 失去连接时会发生什么。 。 。 .