英语 · 法语

rabbitmq @ddd-ts-kit /

RabbitMQ的集成事件提供者：an EventBus 发布的内核的 发行者确认，因此发件箱中继可以安全地标记已发布的记录，并从 每个消费者和事件类型一个具有手动确认的持久队列，数量有限 延迟重试和死队列。的策略表的“队列代理”行 @ddd-ts-kit/events，实现，然后通过 eventBusContract 的 @ddd-ts-kit/testing.

npm i @ddd-ts-kit/rabbitmq @ddd-ts-kit/events amqplib

使用amqplib 2.2对RabbitMQ 4.3进行了测试。所有AMQP 0-9-1经纪人都有出版商确认 应该与…合作 queueType: 'classic';只运行了RabbitMQ。

作文根

处理程序永远不会看到这个包。他们发布到发件箱；中继发布到RabbitMQ； 每个消费上下文都以自己的名称通过总线侦听。这辆公共汽车是给 应用程序作为内核的两个端口： IntegrationEventPublisher 保留发件箱，并且 EventSubscribers 为每个上下文提供其自身消费者名称的订阅者。

init --broker rabbitmq 在项目的平台中编写此布线：由 platform/events/create-event-provider.ts， platform/composition/platform.module.ts 重新绑定 EventSubscribers 在上面，还有 main.ts 追踪到发件箱中继。用手， 同样的内容是：

// platform/messaging.module.ts
import { defineContextModule } from '@ddd-ts-kit/di';
import { EventSubscribers, IntegrationEventPublisher, type Logger, type Metrics } from '@ddd-ts-kit/kernel';
import { OutboxPublisher, OutboxStore } from '@ddd-ts-kit/events';
import { RabbitMqEventBus } from '@ddd-ts-kit/rabbitmq';

declare const config: { readonly amqpUrl: string };
declare const logger: Logger;
declare const metrics: Metrics;

export const bus = new RabbitMqEventBus({
  connection: config.amqpUrl, // or amqplib options, or a factory opening a new connection
  exchange: 'integration-events', // the default
  retry: { maxAttempts: 5, delayMs: 5_000 }, // the defaults
  prefetch: 10, // the default
  onFailure: (failure) => logger.warn('event bus failure', { ...failure }), // codes only, never an error message
  // The label `outcome` of the kernel is success or failure: the four outcomes of a delivery take another name.
  onDelivery: ({ queue, outcome }) => metrics.counter('event_deliveries_total').increment({ queue, status: outcome }),
});

export const messagingModule = defineContextModule({
  name: 'messaging',
  register(builder) {
    // Publishing side: handlers write to the outbox, in the transaction of the command.
    builder.rebind(IntegrationEventPublisher).toClass(OutboxPublisher, [OutboxStore]);
    // Consuming side: one subscriber per consuming context, named after it.
    builder.rebind(EventSubscribers).toValue({ forContext: (name) => bus.forConsumer(name) });
  },
});

import { Command, CommandBus, systemActor, type Actor, type IntegrationEvent } from '@ddd-ts-kit/kernel';
import { IntegrationEventHandler } from '@ddd-ts-kit/kernel/declarations';
import { defineIntegrationEvent, type PayloadOf } from '@ddd-ts-kit/events';
import { zodSchema } from '@ddd-ts-kit/validation/zod';
import { z } from 'zod';

// contexts/ordering/published-language: the contract billing listens to.
export const orderPlaced = defineIntegrationEvent({
  type: 'ordering.order.placed',
  version: 1,
  payload: zodSchema(z.object({ orderId: z.string() })),
});

// contexts/billing/application/commands/issue-invoice/issue-invoice.command.ts
export class IssueInvoice extends Command {
  constructor(
    readonly actor: Actor,
    readonly orderId: string,
  ) {
    super();
  }
}

// contexts/billing/application/integration/handlers/invoice-on-order-placed.integration-handler.ts
@IntegrationEventHandler(orderPlaced, { consumer: 'billing.invoice-on-order-placed' })
export class InvoiceOnOrderPlacedIntegrationHandler {
  constructor(private readonly commands: CommandBus) {}

  // Nobody is calling: the handler acts as a system actor named after its consumer.
  readonly actor = systemActor('billing.invoice-on-order-placed');

  async handle(event: IntegrationEvent<PayloadOf<typeof orderPlaced>>): Promise<void> {
    await this.commands.execute(new IssueInvoice(this.actor, event.payload.orderId));
  }
}

// main.ts
import { contextModule, kernelModule, type ContextModule, type DeclaredSource } from '@ddd-ts-kit/di';
import { createApplication } from '@ddd-ts-kit/di/tsyringe';
import { Clock } from '@ddd-ts-kit/kernel';
import { OutboxRelay, OutboxStore } from '@ddd-ts-kit/events';
import type { RabbitMqEventBus } from '@ddd-ts-kit/rabbitmq';

// In a project: import { bus, messagingModule } from '@/platform/messaging.module';
declare const bus: RabbitMqEventBus;
declare const messagingModule: ContextModule;
declare const persistenceModule: ContextModule;
declare const orderingDeclarations: DeclaredSource;
declare const billingDeclarations: DeclaredSource;

const app = createApplication({
  modules: [
    kernelModule(),
    persistenceModule,
    messagingModule,
    contextModule('ordering', orderingDeclarations),
    contextModule('billing', billingDeclarations),
  ],
});

// The relay target is the bus. A record is marked published when the broker confirmed it.
const relay = new OutboxRelay({ store: app.resolve(OutboxStore), publisher: bus, clock: app.resolve(Clock) });

await app.start(); // subscribes the handler of billing on bus.forConsumer('billing')
await bus.assertTopology(); // fail at startup, not at the first message
relay.start({ intervalMs: 500 });

process.on('SIGTERM', async () => {
  await relay.stop(); // the tick in progress finishes
  await app.stop(); // the handlers are unsubscribed
  await bus.close(); // consumers stop, handlers in progress finish and acknowledge
});

• persistenceModule 结合 OutboxStore，并重新结合 UnitOfWork 和 ProcessedEventStore 到数据库：请参阅持久性适配器的README。声明的集成事件 处理程序在其自身上下文的订阅者(forContext 接收名称 的模块)，以及since UnitOfWork 和 ProcessedEventStore 是束缚的，它的幂等 标记和它的工作在一个工作单元中提交：一次重新交付是重复数据删除，一次失败 没有留下任何东西。
• 如果没有发件箱，总线就是内核模块的事件总线，处理程序发布 直接给经纪人： kernelModule({ eventBus: bus, eventSubscribers: () => ({ forContext: (name) => bus.forConsumer(name) }) }).
• 如果没有容器，同样的对象是手工连接的： bus.forConsumer('billing') 是一个 IntegrationEventSubscriber给予，给予 subscribeAll 和 consume 的 @ddd-ts-kit/events. 一辆没有 consumer 选项没有名称给它的队列，所以 bus.subscribe(...) 抛出 ConsumerNameRequiredError：改为命名消费上下文。

import { consume, defineIntegrationEvent, subscribeAll } from '@ddd-ts-kit/events';
import type { ProcessedEventStore, UnitOfWork } from '@ddd-ts-kit/kernel';
import { RabbitMqEventBus } from '@ddd-ts-kit/rabbitmq';
import { zodSchema } from '@ddd-ts-kit/validation/zod';
import { z } from 'zod';

declare const processed: ProcessedEventStore; // the store of the persistence adapter
declare const unitOfWork: UnitOfWork; // the transaction of the database
declare function issueInvoice(orderId: string): Promise<void>;

const orderPlaced = defineIntegrationEvent({
  type: 'ordering.order.placed',
  version: 1,
  payload: zodSchema(z.object({ orderId: z.string() })),
});

const bus = new RabbitMqEventBus({ connection: 'amqp://kit:[email protected]:5672' });

// Instead of: bus.subscribe(orderPlaced.type, handler), which throws ConsumerNameRequiredError here.
// Write: the subscriber of one consuming context, whose name its queues carry.
export const billing = subscribeAll(bus.forConsumer('billing'), [
  consume(orderPlaced, { consumer: 'billing.invoice-on-order-placed', processed, unitOfWork }, async (event) => {
    await issueInvoice(event.payload.orderId);
  }),
]);

await bus.assertTopology(); // declares billing.ordering.order.placed and its .retry and .dead, then consumes

• publish(events) 仅在代理确认时解析 每一个 消息的批处理和 否则拒绝(PUBLISH_NOT_CONFIRMED， BROKER_UNAVAILABLE， UNROUTABLE_EVENT， EVENT_BUS_CLOSED）. 以后不会缓冲任何东西：发件箱就是缓冲区。
• 订阅时必须提供消费者姓名： bus.forConsumer('billing') 给出了一个 IntegrationEventSubscriber 对于共享连接（模块化单体）上的一个上下文 有几个)，和 consumer 选项名称 bus.subscribe 为某项服务工作 这是一个单一的上下文)。
• 跟踪是从外部在两侧进行的，与任何提供程序一样： tracedPublisher(bus) 作为中继目标，在消费端对内核模块进行检测 （kernelModule({ instrumentation: traceApplication() }))，它给出每个声明的处理程序 一个跨度 process <consumer> 每一个尝试。 tracedSubscriber(bus.forConsumer(name)) 是为 没有的作文 kernelModule． 公共汽车载有 traceparent 和 correlationId 在信封和邮件头里都没动过。
• 这种相关性是双向的。通过发布 scopedPublisher(publisher, scope) 的 内核：声明的处理程序(或 consume(definition, { ..., scope }, handler) 用手) 在执行上下文中运行，其 correlationId 是其中一个发布的请求， 谁的 causationId 是事件，失败后的重新交付在同一事件中运行。一个 操作员在不打开正文的情况下读取消息，在 correlation_id 财产 和 x-correlation-id 头。
• 提取到其自己的服务的上下文保留其消费者名称，因此保留其队列：什么都没有 代理上的更改。

拓扑结构

bus.describeTopology() 返回交换器、队列和绑定作为数据，以供查看或 生成的文档； bus.assertTopology() 幂等地声明它们。公共汽车还声明 他们在每一个（重新）连接。为 forConsumer('billing') 订阅 ordering.order.placed：

重试和死信通过代理的默认交换(路由密钥= 队列名)，因此只有一个交换器需要管理，除了上面的交换器之外没有绑定。法定人数 重试队列也需要 x-dead-letter-strategy: at-least-once 和 x-overflow: reject-publish， 以便过期的重试不会在返回工作队列的途中丢失。

消息是JSON格式的信封(id， type， version， occurredAt， payload， traceparent， correlationId， tenantId)，持久的，带有路由密钥和 type 属性设置为事件类型， messageId 到事件id， correlationId 到相关性 Id和标头 traceparent， x-correlation-id， x-tenant-id 和 x-event-version． 任何 AMQP客户端可以使用它。

一种类型的两个消费者各有一个队列，并且都接收每个事件。两个例子 一个消费者共享它的队列：每个事件进入其中一个。

重试和死信行为

1. 消息在每个订阅到其队列的处理程序被解析后才被确认。一个声明 @IntegrationEventHandler (或 consume 的 @ddd-ts-kit/events 手工)提交它的工作 它的幂等性标记在一个工作单元中：然后确认提交， 然后再发送一次重复数据删除。
2. 当处理程序抛出时，总线将消息的确认副本发送到重试队列 有了标题 x-attempt 增加和 x-error-code 设置，然后确认 原创。副本等待 retry.delayMs 然后返回到工作队列。一个失败的 消息永远不会在现场排队：没有热循环。
3. 在 retry.maxAttempts-交付失败的副本将转到死队列，并带有 x-dead-reason: max-attempts.
4. 消息体不是信封的消息(无法解析，没有。 id， type， version 或者 occurredAt，或者用 tenantId 那 isTenantId （内核的拒绝）是毒药：它 直接去排队等候 x-dead-reason: poison 和 x-error-code: POISON_MESSAGE，没有处理程序看到它。
5. 只有代码传播： x-error-code 是 code 错误或其类名的。的消息 永远不会记录、报告或附加错误，因为它可能引用有效负载数据。

送货至少一次，而且是无序的。复制-然后确认的顺序意味着崩溃 Between传递消息两次，而不是零次。

在失去连接时，总线以指数回退重新连接(reconnect， 200毫秒 加倍到30秒)，再次声明其拓扑并恢复其消费者。出版物中 飞行拒绝，也是如此 publish 在两次尝试之间，因此发件箱中继保持 记录并重试它们。正在处理的消息由代理重新传递。

凭证。 的URL或选项 connection 由打开连接的函数持有，不包含任何字段。一个 RabbitMqEventBus 这是记录，传播或 连载显示 RabbitMqEventBus [connection concealed] 没有别的了：它的域是不可数的， 和 toJSON 和 util.inspect 回答那个名字。它报告的失败是代码。什么 客户端库本身显示，当它自己的对象之一被检查时，不在 这个包：不记录那些。

处理器被告知的内容。 在事件旁边，每个处理程序接收传递 （EventDelivery 的内核)： { attempt, maxAttempts, system: 'rabbitmq', destination, consumer }， 哪里 destination 工作队列是和吗 consumer 消费者名称(billing）. 应用程序的处理程序忽略它； @ddd-ts-kit/tracing 读它 每次尝试创建一个CONSUMER span，并标记重新交付和之后的尝试 事件被掩埋。公共汽车自己命名 messagingSystem: 'rabbitmq', traceparent 旅行在 并将信封作为真实的报头，完好无损地跨一次重试。

操作

看点：

有了错误跟踪器， DeadLetterReports 核的旋转数 onFailure 的报告 被放弃的事情，而且仅仅是那些： onFailure: (failure) => reports.onFailure(failure), reports.observing(bus.forConsumer(name)) 作为每一个上下文的订阅者，使一个报告 携带处理程序抛出的内容。 init 当项目有一个时连接它。

重播死消息，一旦原因被修复：

import type { RabbitMqEventBus } from '@ddd-ts-kit/rabbitmq';

declare const bus: RabbitMqEventBus;

// Up to 100 messages of billing.ordering.order.placed.dead, back to the work queue with x-attempt 1.
const replayed = await bus.replayDead({ consumer: 'billing', type: 'ordering.order.placed' }, 100);
console.log(`${String(replayed)} message(s) replayed`);

它将多达这么多的消息从死队列移动回工作队列 尝试预算，在确认前各有一份确认副本。管理UI或 一把铲子也可以移动它们；重置或删除 x-attempt 头，或者消息 在第一次失败时再次死亡。看看 x-dead-reason 和 x-error-code 第一:重放 一条有毒的信息只会再次杀死它。

要永久停止一个消费者，请删除它的三个队列：取消订阅使它们绑定，并且 工作队列不断填满。

限制

• 没有订购。 重试和竞争实例重新排序。在有效负载中携带一个版本。
• 一个延迟。 重试等待时间是重试队列的TTL：每次尝试都是一样的，不是 指数倒扣。
• 队列参数在创建时是固定的。 改变 retry.delayMs 或者 queueType 对于一个 现有队列被代理拒绝(报告为 AMQP_406 而公共汽车一直试图 重新连接)：删除队列，或在另一个消费者名称下迁移。
• 重新交付将再次运行队列的每个处理程序，那些成功的包括，当 一个使用者的多个处理程序订阅同一类型。每个都必须是幂等的。
• 批处理不是原子的。 当 publish 不合格品，一些事件的批次可能已经 被经纪人拿走了。中继逐个发布并重试，消费者重复数据删除。
• onUnroutable: 'fail' 拒绝没有队列绑定的事件；在发件箱中继后面 这样的记录被重试，然后死亡。只有当消费者失踪是一件意外事件时才使用它。
• publishTimeoutMs （默认为30000）：broker尚未确认的发布 拒绝与 PUBLISH_NOT_CONFIRMED，就像在NATS和Kafka上一样，所以中继会重新尝试它，而不是 等待一个停滞的broker永远不会发送的确认。
• 在处理程序中间崩溃的消费者 不返回它的消息 x-attempt 被 递增。仲裁队列将其与它们的传递限制绑定在一起，在此之后，消息将被删除 被代理死信到死队列；经典队列不绑定它。这条路径是 代理的和不是由测试套件执行的。
• 经典的队列 对过期的死信最多重试一次：broker在此崩溃 瞬间可以失去一个。仲裁队列（默认）使用“至少一次”策略。
• 连接工厂必须在每次调用时打开一个新连接;公共汽车关门了 打开了。不能给出与其他代码共享的连接。
• 批量写入时无需等待套接字耗尽：内存随着socket的增长而增长 批处理。继电器逐个发布记录，从而避免了这种情况。
• 队列名称如下 <consumer>.<type> 加上一个后缀，在255字节的AMQP中。

其他经纪人

@ddd-ts-kit/kafka， @ddd-ts-kit/nats 和 @ddd-ts-kit/redis （Redis Streams）实现相同的两个端口和 传递相同的合同套件。 文档/ other-brokers.md 说什么 provider欠，将这个概念映射到Kafka和JetStream，就像那些包做的那样， 并概述了未建立的SQS/SNS。

测试

每个套件都在单元运行中运行，没有服务器：总线连接到其中的AMQP代理 内存(test/support/amqp/)，它实现了amqplib的接口并保持了总线的内容 依赖于主题交换和默认的死信参数， x-message-ttl出版商 确认并返回，预取，重新发送关闭通道未确认的内容，406和 404拒绝)。与 DDD_TS_KIT_TEST_AMQP_URL，同样的套间会与雷亚尔进行第二次比对 Broker，它使内存中的对象保持诚实；

docker run -d --rm --name ddd-ts-kit-test-rabbitmq -e RABBITMQ_DEFAULT_USER=kit -e RABBITMQ_DEFAULT_PASS=kit -p 127.0.0.1:5672:5672 rabbitmq:4-management-alpine
DDD_TS_KIT_TEST_AMQP_URL=amqp://kit:[email protected]:5672 npx vitest run packages/rabbitmq
docker rm -f ddd-ts-kit-test-rabbitmq

CI的集成工作是在每个到达这个包的拉取请求上针对一个真正的代理运行它们， RabbitMQ 4.3.6版本 pipeline/github/scripts/start-rabbitmq.sh 开始，它的形象被文摘钉住了。

他们在仲裁和经典队列上运行合约套件，然后是它的行为 提供者，每个行为一个套件(publishing， consuming， correlation， retries， dead-letters， topology， connection-loss 通过网络测试切割，TCP中继在 在一个真正的经纪人面前 closing） 以及从发件箱到幂等消费者的整个链条。每个测试都在一个交换器和它的队列上工作 拥有并删除它们。