Пошук уроків, статей та іншого контенту
Налаштуєте підтвердження, ручне споживання й повторну доставку повідомлень у брокерах.
Брокер повідомлень не повинен вважати повідомлення обробленим лише тому, що передав його застосунку. Після отримання повідомлення застосунок має повідомити брокеру, що:
повідомлення успішно оброблено;
повідомлення потрібно доставити повторно;
повідомлення не можна обробити й повторна доставка не потрібна.
У RabbitMQ це називається підтвердженням повідомлення:
ack — позитивне підтвердження;
nack — негативне підтвердження;
requeue — повернення повідомлення в чергу для повторної доставки.
Якщо застосунок аварійно завершиться до підтвердження, RabbitMQ зможе повторно доставити повідомлення. Це захищає повідомлення від втрати.
Підтвердження отримання не означає підтвердження успішного виклику обробника. Підтверджувати повідомлення потрібно після завершення необхідної бізнес-операції.
Для RabbitMQ у NestJS потрібно вимкнути автоматичне підтвердження:
noAck: falseПісля цього обробник отримує доступ до:
каналу RabbitMQ;
оригінального повідомлення;
метаданих повідомлення.
Для цього використовується RmqContext.
import { RmqContext } from '@nestjs/microservices';
const channel = context.getChannelRef();
const message = context.getMessage();Метод getData() повертає дані повідомлення, а getMessage() — об’єкт RabbitMQ, потрібний для підтвердження.
Нижче наведено обробник події orders.created. Повідомлення підтверджується тільки після успішної обробки.
import {
Controller,
Logger,
} from '@nestjs/common';
import {
Ctx,
EventPattern,
Payload,
RmqContext,
} from '@nestjs/microservices';
interface OrderCreatedEvent {
orderId: string;
userId: string;
total: number;
}
@Controller()
export class OrdersController {
private readonly logger = new Logger(OrdersController.name);
@EventPattern('orders.created')
async handleOrderCreated(
@Payload() event: OrderCreatedEvent,
@Ctx() context: RmqContext,
): Promise<void> {
const channel = context.getChannelRef();
const message = context.getMessage();
try {
this.logger.log(`Обробка замовлення ${event.orderId}`);
await this.processOrder(event);
// Підтверджуємо повідомлення лише після успішної обробки.
channel.ack(message);
this.logger.log(`Замовлення ${event.orderId} оброблено`);
} catch (error) {
this.logger.error(
`Не вдалося обробити замовлення ${event.orderId}`,
error instanceof Error ? error.stack : undefined,
);
// true означає повернути повідомлення до черги.
channel.nack(message, false, true);
}
}
private async processOrder(
event: OrderCreatedEvent,
): Promise<void> {
// Тут може бути запис у базу даних або виклик іншого сервісу.
await new Promise((resolve) => setTimeout(resolve, 100));
if (event.total <= 0) {
throw new Error('Сума замовлення має бути більшою за нуль');
}
}
}Другий аргумент false у ack і nack означає, що підтверджується лише одне повідомлення:
channel.ack(message, false);
channel.nack(message, false, true);У більшості обробників потрібно підтверджувати саме одне повідомлення. Підтвердження кількох повідомлень одночасно може бути небезпечним, якщо частина з них ще не оброблена.
Параметр noAck: false потрібно вказати в конфігурації RabbitMQ-транспорту.
import { NestFactory } from '@nestjs/core';
import {
MicroserviceOptions,
Transport,
} from '@nestjs/microservices';
import { AppModule } from './app.module';
async function bootstrap(): Promise<void> {
const app = await NestFactory.createMicroservice<MicroserviceOptions>(
AppModule,
{
transport: Transport.RMQ,
options: {
urls: ['amqp://localhost:5672'],
queue: 'orders',
queueOptions: {
durable: true,
},
noAck: false,
prefetchCount: 10,
},
},
);
await app.listen();
}
bootstrap();У цьому прикладі:
noAck: false вмикає ручні підтвердження;
durable: true робить чергу стійкою до перезапуску брокера;
prefetchCount: 10 обмежує кількість непідтверджених повідомлень, які RabbitMQ передає одному споживачу.
prefetchCount не замінює підтвердження. Він лише контролює, скільки повідомлень можуть одночасно перебувати в стані очікування.
ackПісля успішного завершення операції викликається:
channel.ack(message);Після цього RabbitMQ видаляє повідомлення з черги. Якщо застосунок завершиться після ack, це повідомлення вже не буде повторно доставлене.
Тому порядок операцій має значення:
await saveOrderToDatabase(event);
channel.ack(message);Неправильний порядок може призвести до втрати повідомлення:
channel.ack(message);
await saveOrderToDatabase(event);Якщо процес завершиться між ack і записом у базу даних, брокер уже видалить повідомлення, а бізнес-операція не завершиться.
nackМетод nack має такий вигляд:
channel.nack(message, allUpTo, requeue);Для одного повідомлення з повторною доставкою:
channel.nack(message, false, true);Параметри означають:
message — повідомлення RabbitMQ;
false — не застосовувати операцію до попередніх повідомлень;
true — повернути повідомлення в чергу.
Якщо повторна доставка не потрібна:
channel.nack(message, false, false);У такому випадку повідомлення буде відкинуте. Якщо для черги налаштована dead-letter exchange, RabbitMQ може передати його до dead-letter черги.
Повторна доставка відбувається, коли повідомлення залишилося непідтвердженим або було повернуте в чергу через nack(..., true).
RabbitMQ передає ознаку повторної доставки в полі:
message.fields.redeliveredУ NestJS її можна перевірити через RmqContext:
import {
Controller,
Logger,
} from '@nestjs/common';
import {
Ctx,
EventPattern,
Payload,
RmqContext,
} from '@nestjs/microservices';
@Controller()
export class PaymentsController {
private readonly logger = new Logger(PaymentsController.name);
@EventPattern('payments.created')
async handlePayment(
@Payload() payment: { paymentId: string },
@Ctx() context: RmqContext,
): Promise<void> {
const channel = context.getChannelRef();
const message = context.getMessage();
const isRedelivered = message.fields.redelivered;
if (isRedelivered) {
this.logger.warn(
`Повторна доставка платежу ${payment.paymentId}`,
);
}
try {
await this.processPayment(payment);
// Підтвердження після успішної обробки.
channel.ack(message);
} catch (error) {
if (isRedelivered) {
this.logger.error(
`Повторна обробка платежу теж завершилася помилкою`,
);
// Не створюємо нескінченний цикл повторних доставок.
channel.nack(message, false, false);
return;
}
// Перша помилка: дозволяємо повторну доставку.
channel.nack(message, false, true);
}
}
private async processPayment(
payment: { paymentId: string },
): Promise<void> {
this.logger.log(`Платіж ${payment.paymentId} обробляється`);
}
}Перевірка redelivered може бути корисною, але вона не є повноцінним лічильником спроб. Значення true лише повідомляє, що повідомлення вже доставлялося раніше.
Такий код може створити нескінченний цикл:
catch {
channel.nack(message, false, true);
}Якщо повідомлення завжди спричиняє помилку, воно буде:
отримане споживачем;
повернуте в чергу;
знову отримане;
знову повернуте в чергу.
У результаті одне проблемне повідомлення може постійно займати споживача.
Практичні стратегії:
повторити доставку обмежену кількість разів;
після кількох невдалих спроб відкинути повідомлення;
направити невдале повідомлення до dead-letter черги;
для неповторюваних помилок одразу використати nack(message, false, false).
Вибір стратегії залежить від типу помилки. Тимчасова недоступність зовнішнього сервісу може виправдати повторну доставку. Некоректний формат повідомлення зазвичай не зміниться після повторної спроби.
Не слід підтверджувати повідомлення в блоці finally:
try {
await processMessage();
} finally {
channel.ack(message);
}У такому разі повідомлення буде підтверджене навіть після помилки.
Безпечніший варіант:
try {
await processMessage();
channel.ack(message);
} catch {
channel.nack(message, false, true);
}Також не потрібно підтверджувати одне повідомлення двічі:
channel.ack(message);
channel.ack(message);Після першого підтвердження RabbitMQ вже вважає повідомлення завершеним. Повторна операція може призвести до помилки каналу.
Повторна доставка означає, що один і той самий бізнес-запит може бути оброблений більше одного разу. Наприклад, застосунок може:
записати замовлення в базу даних;
завершитися до виклику ack;
отримати те саме повідомлення після перезапуску;
спробувати записати замовлення повторно.
Тому операції споживача мають бути ідемпотентними: повторний виклик не повинен створювати некоректний результат.
Типовий підхід:
передавати у повідомленні стабільний ідентифікатор події;
зберігати оброблені ідентифікатори;
перевіряти, чи подія вже оброблялася;
викликати ack, якщо повторне отримання не потребує нової операції.
Наприклад, перевірка може мати такий вигляд:
const alreadyProcessed = await eventStore.has(eventId);
if (alreadyProcessed) {
// Повторне повідомлення вже було оброблене.
channel.ack(message);
return;
}
await processEvent(event);
await eventStore.save(eventId);
// Фіксуємо успіх лише після завершення обробки.
channel.ack(message);Важливо, щоб запис про обробку події та основна бізнес-операція узгоджувалися між собою. Інакше повторна доставка все одно може спричинити дублювання.
У NestJS із RabbitMQ Nest створює споживача й викликає метод, позначений @EventPattern() або @MessagePattern().
Ручне підтвердження означає не ручне створення channel.consume(), а ручне керування моментом підтвердження:
@EventPattern('orders.created')
async handle(
@Payload() data: unknown,
@Ctx() context: RmqContext,
): Promise<void> {
const channel = context.getChannelRef();
const message = context.getMessage();
await doWork(data);
channel.ack(message);
}Якщо вказати noAck: true, брокер автоматично вважатиме повідомлення підтвердженим під час доставки. У такому режимі ручний виклик ack не потрібен і не повинен використовуватися.
Для надійного споживача послідовність дій така:
Отримати дані через @Payload().
Отримати RmqContext.
Взяти канал і оригінальне повідомлення.
Виконати бізнес-операцію.
Викликати ack після успіху.
Викликати nack у разі помилки.
Вирішити, чи потрібна повторна доставка.
Не допускати нескінченних повторних спроб.
noAck: trueЯкщо noAck: true, повідомлення підтверджуються автоматично. Виклик channel.ack(message) не забезпечить ручний контроль.
Для ручного підтвердження потрібно:
noAck: falseack викликається до бізнес-операціїПовідомлення не можна підтверджувати до завершення обробки. Інакше аварія між ack і бізнес-операцією призведе до втрати повідомлення.
nack(..., true) без обмеження спроб може створити нескінченний цикл повторної доставки.
ack або nackЯкщо повідомлення залишається непідтвердженим, воно може блокувати доступну кількість повідомлень згідно з prefetchCount. Обробник має завершити повідомлення одним із чітких сценаріїв:
ack;
nack із повторною доставкою;
nack без повторної доставки.
Повторна доставка є нормальною поведінкою брокера. Критичні операції потрібно проєктувати так, щоб повторна обробка не псувала дані.
noAck: false вмикає ручне підтвердження RabbitMQ-повідомлень у NestJS.
RmqContext надає доступ до каналу та оригінального повідомлення.
channel.ack(message) потрібно викликати після успішної обробки.
channel.nack(message, false, true) повертає повідомлення в чергу.
channel.nack(message, false, false) відкидає повідомлення без повторної доставки.
Повторна доставка може спричинити дублювання, тому обробники мають бути ідемпотентними.
Безконтрольне використання requeue: true може створити нескінченний цикл помилок.
prefetchCount обмежує кількість непідтверджених повідомлень, але не замінює ack і nack.