Пошук уроків, статей та іншого контенту
Винесете довгі та ресурсоємні операції у фонове виконання без блокування HTTP-запитів.
HTTP-запит не повинен чекати завершення довгої операції. Наприклад:
надсилання електронного листа;
оброблення великого файлу;
генерація звіту;
синхронізація даних із зовнішнім сервісом;
створення резервної копії.
Якщо виконувати таку операцію безпосередньо в контролері, клієнт буде довго чекати на відповідь:
@Post('reports')
async createReport() {
const report = await this.generateLargeReport();
return report;
}Краще додати завдання до черги, одразу повернути клієнту відповідь, а саме завдання обробити пізніше:
HTTP-запит → додати завдання до черги → повернути відповідь
↓
обробити завданняДля черги потрібне сховище, у якому завдання очікуватимуть виконання. У NestJS для цього зручно використовувати BullMQ та Redis.
У типовій схемі є дві частини:
Producer — додає завдання до черги.
Processor — отримує завдання з черги та виконує його.
У NestJS producer зазвичай знаходиться в сервісі або контролері, а processor — в окремому класі з декоратором @Processor().
Redis зберігає:
завдання, які очікують виконання;
завдання, що виконуються;
успішно завершені завдання;
завдання з помилками;
кількість спроб виконання.
У вже створеному NestJS-проєкті встановіть пакети:
npm install @nestjs/bullmq bullmqBullMQ використовує Redis. Якщо Redis встановлений локально, запустіть його звичайним способом. Для швидкого запуску в Docker можна виконати:
docker run --rm --name nest-redis -p 6379:6379 redis:7-alpineСтворимо чергу для надсилання електронних листів.
app.module.tsimport { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';
import { EmailsController } from './emails.controller';
import { EmailsService } from './emails.service';
import { EmailProcessor } from './email.processor';
@Module({
imports: [
BullModule.forRoot({
connection: {
host: 'localhost',
port: 6379,
},
}),
BullModule.registerQueue({
name: 'emails',
}),
],
controllers: [EmailsController],
providers: [EmailsService, EmailProcessor],
})
export class AppModule {}BullModule.forRoot() налаштовує підключення до Redis.
BullModule.registerQueue() реєструє чергу з назвою emails. Цю саму назву потрібно використовувати в producer та processor.
Створимо сервіс, який додає завдання до черги.
emails.service.tsimport { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
interface WelcomeEmailData {
email: string;
}
@Injectable()
export class EmailsService {
constructor(
@InjectQueue('emails')
private readonly emailsQueue: Queue<WelcomeEmailData>,
) {}
async scheduleWelcomeEmail(email: string) {
const job = await this.emailsQueue.add(
'send-welcome-email',
{
email,
},
{
attempts: 3,
backoff: {
type: 'exponential',
delay: 1000,
},
removeOnComplete: 100,
removeOnFail: 1000,
},
);
return {
jobId: job.id,
status: 'queued',
};
}
}Метод add() не виконує надсилання листа. Він лише записує завдання в Redis.
Параметри завдання:
send-welcome-email — назва завдання;
email — дані, необхідні для його виконання;
attempts: 3 — максимум три спроби;
backoff — затримка перед повторною спробою;
removeOnComplete — скільки завершених завдань залишати;
removeOnFail — скільки невдалих завдань залишати.
Повторні спроби корисні, якщо тимчасово недоступний зовнішній сервіс надсилання листів.
Тепер додамо контролер, який прийматиме HTTP-запит і ставитиме завдання в чергу.
emails.controller.tsimport {
Body,
Controller,
HttpCode,
HttpStatus,
Post,
BadRequestException,
} from '@nestjs/common';
import { EmailsService } from './emails.service';
@Controller('emails')
export class EmailsController {
constructor(private readonly emailsService: EmailsService) {}
@Post('welcome')
@HttpCode(HttpStatus.ACCEPTED)
async createWelcomeEmail(@Body() body: { email?: string }) {
if (!body.email) {
throw new BadRequestException('Поле email є обов’язковим');
}
return this.emailsService.scheduleWelcomeEmail(body.email);
}
}Код 202 Accepted означає, що сервер прийняв запит, але робота ще не завершена.
Приклад запиту:
POST /emails/welcome
Content-Type: application/json
{
"email": "user@example.com"
}Приклад відповіді:
{
"jobId": "1",
"status": "queued"
}Клієнт отримує відповідь одразу після додавання завдання до черги. Він не чекає завершення надсилання листа.
Створимо processor. Він отримуватиме завдання з черги та виконуватиме потрібну операцію.
email.processor.tsimport { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
interface WelcomeEmailData {
email: string;
}
@Processor('emails')
export class EmailProcessor extends WorkerHost {
async process(job: Job<WelcomeEmailData>) {
if (job.name !== 'send-welcome-email') {
throw new Error(`Невідомий тип завдання: ${job.name}`);
}
console.log(`Початок надсилання листа на ${job.data.email}`);
// Імітуємо довгу асинхронну операцію
await new Promise((resolve) => {
setTimeout(resolve, 5000);
});
console.log(`Лист надіслано на ${job.data.email}`);
return {
email: job.data.email,
sent: true,
};
}
}Клас із @Processor('emails') слухає чергу emails.
Метод process() викликається для кожного нового завдання. Об’єкт job містить:
job.name — назву завдання;
job.data — передані дані;
job.id — ідентифікатор;
іншу службову інформацію про виконання.
Запустіть застосунок:
npm run start:devПісля POST-запиту контролер одразу поверне статус queued, а через п’ять секунд у консолі processor з’явиться повідомлення про завершення операції.
Нижче наведено основні файли разом. Цей приклад можна використати у звичайному NestJS-проєкті.
app.module.tsimport { Module } from '@nestjs/common';
import { BullModule } from '@nestjs/bullmq';
import { EmailsController } from './emails.controller';
import { EmailsService } from './emails.service';
import { EmailProcessor } from './email.processor';
@Module({
imports: [
BullModule.forRoot({
connection: {
host: 'localhost',
port: 6379,
},
}),
BullModule.registerQueue({
name: 'emails',
}),
],
controllers: [EmailsController],
providers: [EmailsService, EmailProcessor],
})
export class AppModule {}emails.service.tsimport { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
interface WelcomeEmailData {
email: string;
}
@Injectable()
export class EmailsService {
constructor(
@InjectQueue('emails')
private readonly emailsQueue: Queue<WelcomeEmailData>,
) {}
async scheduleWelcomeEmail(email: string) {
const job = await this.emailsQueue.add('send-welcome-email', {
email,
});
return {
jobId: job.id,
status: 'queued',
};
}
}emails.controller.tsimport {
BadRequestException,
Body,
Controller,
HttpCode,
HttpStatus,
Post,
} from '@nestjs/common';
import { EmailsService } from './emails.service';
@Controller('emails')
export class EmailsController {
constructor(private readonly emailsService: EmailsService) {}
@Post('welcome')
@HttpCode(HttpStatus.ACCEPTED)
async createWelcomeEmail(@Body() body: { email?: string }) {
if (!body.email) {
throw new BadRequestException('Поле email є обов’язковим');
}
return this.emailsService.scheduleWelcomeEmail(body.email);
}
}email.processor.tsimport { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
interface WelcomeEmailData {
email: string;
}
@Processor('emails')
export class EmailProcessor extends WorkerHost {
async process(job: Job<WelcomeEmailData>) {
console.log(`Оброблення завдання ${job.id}`);
// Тут може бути виклик поштового сервісу
await new Promise((resolve) => {
setTimeout(resolve, 5000);
});
console.log(`Завдання ${job.id} завершено`);
return {
email: job.data.email,
sent: true,
};
}
}У наведеному прикладі producer і processor працюють в одному NestJS-процесі. Це вже дозволяє не чекати завершення завдання в HTTP-обробнику.
Однак асинхронність не робить синхронні обчислення неблокувальними. Наприклад, такий код все одно блокуватиме event loop:
function expensiveCalculation() {
const end = Date.now() + 5000;
while (Date.now() < end) {
// Синхронне обчислення блокує потік Node.js
}
}Тому для операцій, які активно використовують процесор, processor варто запускати в окремому worker-процесі. Тоді навіть важке обчислення не блокуватиме HTTP-сервер.
Операції введення-виведення, наприклад запити до бази даних або зовнішнього HTTP-сервісу, зазвичай виконують асинхронно через await.
Якщо process() викине помилку, BullMQ позначить завдання як невдале. Якщо в завданні налаштовано attempts, воно буде повторене.
@Processor('emails')
export class EmailProcessor extends WorkerHost {
async process(job: Job<WelcomeEmailData>) {
try {
await this.sendEmail(job.data.email);
return {
sent: true,
};
} catch (error) {
console.error(`Помилка для завдання ${job.id}`);
// Помилка передається BullMQ для повторної спроби
throw error;
}
}
private async sendEmail(email: string) {
// Виклик зовнішнього сервісу надсилання листів
console.log(`Надсилання листа на ${email}`);
}
}Не слід повертати успішну відповідь із processor, якщо операція фактично завершилася помилкою. У такому разі потрібно викинути помилку, щоб черга могла застосувати налаштовану політику повторних спроб.
Фонове виконання підходить, якщо клієнту не потрібно отримати результат операції в межах поточного HTTP-запиту.
Наприклад:
POST /reports
→ звіт поставлено в чергу
→ відповідь 202 Accepted
Фонове завдання:
→ звіт згенеровано
→ файл збереженоУ відповіді можна повертати jobId, щоб пізніше додати окремий ендпойнт для перевірки стану завдання. Саме завдання при цьому залишається незалежним від життєвого циклу HTTP-запиту.
@Post()
async create() {
await this.longOperation();
return { done: true };
}Такий контролер утримує HTTP-запит до завершення операції. Замість цього додайте завдання до черги та поверніть 202 Accepted.
Якщо Redis не запущений або вказано неправильний порт, producer не зможе додати завдання, а processor не зможе його отримати.
Перевірте:
чи запущений Redis;
чи правильні host і port;
чи не зайнятий потрібний порт іншою програмою.
Назви в усіх місцях повинні збігатися:
BullModule.registerQueue({
name: 'emails',
});@InjectQueue('emails')@Processor('emails')Якщо хоча б одна назва відрізняється, processor не отримуватиме завдання з потрібної черги.
providersProcessor повинен бути зареєстрований у модулі:
@Module({
providers: [EmailProcessor],
})
export class AppModule {}Без цього NestJS не створить worker для оброблення завдань.
Тимчасова помилка зовнішнього сервісу не обов’язково означає, що завдання потрібно остаточно скасувати. Для таких операцій налаштовуйте attempts і backoff.
Черга відокремлює HTTP-відповідь від моменту завершення завдання, але processor у тому самому процесі все ще може заблокувати event loop синхронними обчисленнями. Для CPU-ресурсних операцій використовуйте окремий worker-процес.
Довгі операції не варто виконувати безпосередньо в HTTP-контролері.
BullMQ використовує Redis для зберігання та керування чергами.
Producer додає завдання до черги через queue.add().
Processor обробляє завдання через @Processor() і process().
Для прийнятого, але ще не завершеного запиту використовуйте статус 202 Accepted.
attempts і backoff допомагають автоматично повторювати невдалі завдання.
Синхронні CPU-ресурсні операції можуть блокувати Node.js навіть у processor, тому їх слід запускати в окремому процесі.