@bgaldino/nestjs-rabbitmq (英语)
为RabbitMQ提供评分的NestJS模块,内置重试策略,死信队列,法定人数队列支持.
目录
安装
pnpm add @bgaldino/nestjs-rabbitmq
yarn add @bgaldino/nestjs-rabbitmq
npm add @bgaldino/nestjs-rabbitmq
所需资源
@nestjs/common财务报告和审定财务报表@nestjs/core第9版或以上- RabbitMQ 3.10+(每条消息的q法定人数队列 TTL 支持)
不需要额外的插件 。 重试机制使用本地信件 TTL 和死信交换,而不是延迟信件插件.
开始
RabbitMQModule.forRoot() 标记为 @Global,所以导入一次是
足够注入 RabbitMQService 你申请的任何地方。 它处理
连接和出版.
为方便消费者,进口 RabbitMQModule.withConsumers() 并行
forRoot()。没有它,只打开出版商的连接,没有消息
被消耗.
旋转
对于配置为静态的简单设置 :
import { RabbitMQModule } from '@bgaldino/nestjs-rabbitmq';
@Module({
imports: [
RabbitMQModule.forRoot({
connectionString: 'amqp://user:password@localhost:5672/vhost',
assertExchanges: [
{ name: 'orders', type: 'topic' },
{ name: 'notifications', type: 'fanout' },
],
}),
RabbitMQModule.withConsumers(),
],
})
export class AppModule {}
用于RootAsync
当需要同步注入依赖或解析配置时:
import { RabbitMQModule, RabbitMQOptionsFactory, ModuleOptions } from '@bgaldino/nestjs-rabbitmq';
@Injectable()
class RabbitConfig implements RabbitMQOptionsFactory {
constructor(private readonly configService: ConfigService) {}
createRabbitOptions(): ModuleOptions {
return {
connectionString: this.configService.get('RABBIT_URL'),
assertExchanges: [
{ name: 'orders', type: 'topic' },
],
};
}
}
@Module({
imports: [
RabbitMQModule.forRootAsync({
useClass: RabbitConfig,
imports: [ConfigModule],
}),
],
})
export class AppModule {}
您还可以使用 useFactory 直接 :
RabbitMQModule.forRootAsync({
useFactory: (configService: ConfigService) => ({
connectionString: configService.get('RABBIT_URL'),
assertExchanges: [],
}),
inject: [ConfigService],
imports: [ConfigModule],
})
消费者
注册消费者有两种方式:装饰和配置. 都一样 同步使用,运行时产生相同的结果。 图书馆 验证两个来源之间不存在重复的队列名称.
所有队列创建为 法定人数队列
默认。 消费者不创建交易所,他们只约束交易所
已经存在(通过 assertExchanges) (中文(简体) ).
消费者只有在 RabbitMQModule.withConsumers() 输入.
没有它,就没有消费者的联系,也没有信息被消耗.
扶持消费者
导入 withConsumers() 并行 forRoot() 能够发现消费者
并启动。 无弧形会发现所有 @RabbitConsumer 装饰
整个应用和进程的方法 consumerChannels 从
连接配置 :
@Module({
imports: [
RabbitMQModule.forRoot({ ... }),
RabbitMQModule.withConsumers(),
],
})
export class AppModule {}
选择性消费者激活
当应用程序有多个部署时(例如 API 服务器和
背景工作者),您可以直接通过操作员课程 withConsumers()
控制在每次部署中哪些消费者被激活:
// API deployment — publish only, no consumers
@Module({
imports: [
RabbitMQModule.forRoot({ ... }),
],
})
export class ApiAppModule {}
// Worker deployment — only order and payment consumers
@Module({
imports: [
RabbitMQModule.forRoot({ ... }),
RabbitMQModule.withConsumers([OrderHandler, PaymentHandler]),
],
})
export class WorkerAppModule {}
// Report deployment — only report consumers
@Module({
imports: [
RabbitMQModule.forRoot({ ... }),
RabbitMQModule.withConsumers([ReportHandler]),
],
})
export class ReportAppModule {}
当操作员的班级通过时, 只能 @RabbitConsumer 关于这些方法
课程被激活。 配置消费者(consumerChannels总是
处理完毕.
记住 @RabbitConsumer 特定类别上的方法
启动在一起。 如果一个阶层的消费者 意味着不同的部署,
将它分成不同的类别——每个部署问题一个:
// Each class serves a single deployment
class OrderCreateHandler {
@RabbitConsumer({ queue: 'orders.create', ... })
async handle() { ... }
}
class OrderReportHandler {
@RabbitConsumer({ queue: 'orders.report', ... })
async handle() { ... }
}
基于装饰品的消费者
装饰任何方法 @RabbitConsumer() 和图书馆将
应用时自动发现并接通.
这是建议大多数案件采用的办法,因为它保留了
消费者逻辑定位与处理器.
import { Injectable } from '@nestjs/common';
import { RabbitConsumer, ConsumerOptions, MessageParams } from '@bgaldino/nestjs-rabbitmq';
@Injectable()
export class OrderService {
@RabbitConsumer({
queue: 'order.created',
exchangeName: 'orders',
routingKey: 'order.created',
prefetch: 5,
})
async handleOrderCreated(content: OrderPayload, params?: MessageParams) {
// handle message
}
}
供应商必须在雀巢JS模块中注册. 图书馆全部扫描 装饰方法的提供者和控制器.
基于配置的消费者
如果您喜欢集中配置, 或者需要有条件的注册
基于环境的消费者可以使用 consumerChannels 阵列.
这个 defineRabbitConsumer 帮助提供类型安全 :
import { defineRabbitConsumer } from '@bgaldino/nestjs-rabbitmq';
RabbitMQModule.forRoot({
connectionString: 'amqp://localhost',
assertExchanges: [{ name: 'orders', type: 'topic' }],
consumerChannels: [
defineRabbitConsumer({
queue: 'order.created',
exchangeName: 'orders',
routingKey: 'order.created',
prefetch: 5,
handler: {
provider: OrderService,
methodName: 'handleOrderCreated',
},
}),
],
})
库自动解决提供者实例 。 不需要手动
绑定 this 或将服务输入您的配置.
混合使用
这两种方法可以共存。 一个常见的模式是使用装饰器进行静态 消费者和有条件者的配置:
// Static consumer via decorator
@RabbitConsumer({ queue: 'audit.log', exchangeName: 'events', routingKey: '#' })
async auditLog(content: any) { ... }
// Conditional consumer via config
consumerChannels: isProduction ? [
defineRabbitConsumer({ queue: 'debug.trace', ... })
] : [],
处理器签名
每个消费者处理者无论注册与否都遵循同样的签名 方法 :
async handler(content: T, params?: MessageParams): Promise<void>;
这个 MessageParams 对象是可选的,并包含:
type MessageParams = {
message: ConsumeMessage;
channel: ConfirmChannel;
queue: string;
originalRoutingKey?: string;
correlationId?: string | null; // AMQP correlationId (or x-correlation-id header)
retryCount?: number; // retry attempts already performed for this message
};
这个 content 参数是解析后的消息正文。 自动库
试图将消息解析为JSON. 如果解析失败, 原始字符串是
而不是通过.
您可以执行 ConsumerHandler<T> 要强烈键入您的界面
消费者:
import { ConsumerHandler, MessageParams } from '@bgaldino/nestjs-rabbitmq';
@Injectable()
export class OrderService implements ConsumerHandler<OrderPayload> {
async messageHandler(content: OrderPayload, params?: MessageParams): Promise<void> {
console.log(content.orderId);
}
}
出版商
发布消息
弹出 RabbitMQService 打电话 publish()数字 :
import { Injectable } from '@nestjs/common';
import { RabbitMQService } from '@bgaldino/nestjs-rabbitmq';
@Injectable()
export class OrderService {
constructor(private readonly rabbit: RabbitMQService) {}
async createOrder(order: Order) {
const published = await this.rabbit.publish('orders', 'order.created', order);
if (!published) {
// handle publish failure
}
}
}
这个 publish() 方法的使用 出版者确认
在解决问题之前,向经纪人传递了信息
答应。 回返 true 关于成功, false 失败.
类型出版
您可以在编译时通过通用类型强制执行消息形状 :
await this.rabbit.publish<OrderPayload>('orders', 'order.created', {
orderId: '123',
amount: 99.90,
});
您也可以通过自定义发布选项作为第四个参数,例如 其他标题或属性.
多种子连接
如果您的应用程序需要从多个 RabbitMQ 中消费或发布
vhosts (或完全不同的经纪人), 您可以使用命名的连接 .
每一个连接都是自成一体的单元 connectionString, (中文(简体) ),
assertExchanges,以及 consumerChannels。 。 。 .
已命名的连接
将平面连接字段替换为 a connections 阵列。 每个条目
一定有一个独特的 name数字 :
import { RabbitMQModule, ConnectionConfig } from '@bgaldino/nestjs-rabbitmq';
RabbitMQModule.forRoot({
connections: [
{
name: 'default',
connectionString: 'amqp://localhost/main',
assertExchanges: [{ name: 'orders', type: 'topic' }],
},
{
name: 'shared-bus',
connectionString: 'amqp://localhost/shared',
assertExchanges: [{ name: 'events', type: 'topic' }],
},
],
})
平地简指( 二connectionString (在基层一级)
单连接设置. 在内部它变成了一个连接 命名
"default"您无法设置两者 connectionString 财务报告和审定财务报表 connections 地点
同一时间.
以消费者的联系为目标
添加数据 connection 字段改为 @RabbitConsumer() 或基于配置的
消费者可以指定它应该附加的连接。 如果省略,则,
它默认为 "default"数字 :
@RabbitConsumer({
queue: 'events.audit',
exchangeName: 'events',
routingKey: '#',
connection: 'shared-bus',
})
async auditEvents(content: any) { ... }
基于配置的消费者在连接中声明 consumerChannels 为:
自动扩展到此连接 :
connections: [
{
name: 'default',
connectionString: 'amqp://localhost/main',
assertExchanges: [{ name: 'orders', type: 'topic' }],
consumerChannels: [
defineRabbitConsumer({
queue: 'order.created',
exchangeName: 'orders',
routingKey: 'order.created',
handler: { provider: OrderService, methodName: 'handleOrderCreated' },
}),
],
},
],
发布到特定连接
通过 connection 选项到 publish()数字 :
await this.rabbit.publish('events', 'audit.created', payload, {
connection: 'shared-bus',
});
省略时,信件会发布到 "default" 连接.
重试策略
每个消费者都可以界定 retryStrategy 处理瞬间故障。 何时
处理器丢出出错, 信件会发布到重试队列
(单位:千美元){queue}.retry以TTL为主. TTL 过期后, 信件会被路由
返回原来的队列,以进行另一次尝试.
@RabbitConsumer({
queue: 'order.process',
exchangeName: 'orders',
routingKey: 'order.process',
retryStrategy: {
enabled: true,
maxAttempts: 5,
retryFn: (content, attempt, error) => attempt * 5000,
},
})
async processOrder(content: OrderPayload) { ... }
这个 retryFn 召回接收信件内容, 当前尝试编号,
和所抛出的错误。 应返回以毫秒计的延迟
在下次重试之前。 返回值控制行为 :
- 正数 : 信件会用 TTL 发送到重试队列
- 零:电文立即被重审(重发至结束) 原始队列)
- 负数: 重试被完全跳过, 信件会去 直接到死信策略
默认 (当 retryStrategy 未指定 :
enabled: 真实maxAttempts: 5(或extraOptions.defaultMaxRetry设定时)retryFn: () => 5000
您可以设置全局默认值, 用于最大重试尝试
extraOptions.defaultMaxRetry它适用于每个不
自 己声明 retryStrategy.maxAttempts数字 :
RabbitMQModule.forRoot({
connectionString: 'amqp://...',
extraOptions: {
defaultMaxRetry: 10,
},
});
优先级( 从最高到最低): consumer.retryStrategy.maxAttempts →
extraOptions.defaultMaxRetry → 5。 。 。 .
您也可以给出一个“ 字符串” 值, 指同一类的方法, 在接口后 :
function retry(content: any, attempt: number, exception: Error): Promise<number> | number;
达到最大尝试次数后, 信件被取消并发送 转到已死字母队列.
死信战略
每个消费者都可以界定 dlqStrategy 来控制当
信件会用尽所有尝试 :
@RabbitConsumer({
queue: 'order.process',
exchangeName: 'orders',
routingKey: 'order.process',
dlqStrategy: {
suffix: '.dlq',
dlqFn: async (content) => {
await alertService.notify('Order processing failed', content);
return true;
},
},
})
async processOrder(content: OrderPayload) { ... }
这个 suffix 控制已死信件队列的名称。 默认为 .dlq, (中文(简体) ),
导致队列命名 {queue}.dlq。 。 。 .
这个 dlqFn 在向 DLQ 发送消息之前执行。 它接收
原始信件内容并返回布尔值 :
true: 消息在调用后转发到 DLQfalse: 信件被确认并删除, 它不会去 DLQ
如果回调时丢出错误, 消息会被传送到 DLQ .
喜欢 retryStrategy,你可以通过一个"字符串" 方法名称,你想要的
如果属于同一类,请打电话。 方法应执行 IDLQFn 接口 :
function (content: T): Promise<boolean> | boolean;
自定义信头元数据
每个已发布的消息自动包含以下自定义标题:
{
"x-correlation-id": "uuid (mirrors the AMQP correlationId property)",
"x-original-exchange": "exchange_name",
"x-original-routing-key": "routing.key",
"x-published-at": "2026-01-01T00:00:00.000Z"
}
这些信头保留了原始的交换和路由关键引用,它们 否则,当消息通过重试队列或 (原始内容存档于2018-09-21). DLQ.
A 类 correlationId 自动生成(或从
options.correlationId 如果你通过一个),并活下来的重试跳出:
消息失败, 并重新发布到重试队列, 所有原始的 AMQP
属性( e)correlationId, (中文(简体) ), contentType, (中文(简体) ), messageId)被保留,并且
a 一个 x-last-retry-at 印有时间戳 x-retries-count。 。 。 .
这个 originalRoutingKey 输入 MessageParams 来源于这些信头
当可用时,返回到消息当前路由键.
额外选项
邮件检查和记录
您可以通过设置查看消费者和出版商的信息 extraOptions.logType
或 RABBITMQ_LOG_TYPE 环境变量为:
all记录消费者和出版商的信息consumer——只记录消费者信息publisher——仅日志发布者信息none——没有信件记录(默认)
环境变量优先于配置值.
消费者的错误总是被记录下来,不管这种设置如何.
每个日志条目是一个结构化对象,包含绑定,持续,
correlationId,重试元数据(retryCount, (中文(简体) ), publishedAt)和——当
信件带有 W3C 微量上下文标题(traceparent编号 :tracestate,比如说.
由OpenTeleometry自动仪器注入)——a traceContext 字段,所以
日志条目可以与分布的痕迹相连接.
健康检查
您可以检查消费者和出版商的连接状况 连接 :
const rabbit = app.get(RabbitMQService);
// Check all connections (returns 0 if any connection is offline)
const status = await rabbit.checkHealth(); // 1 = online, 0 = offline
// Check a specific connection
const sharedStatus = await rabbit.checkHealth('shared-bus');
何时 withConsumers() 未导入,健康检查仅核实
出版社连接.
当地建筑
pnpm install
pnpm build