Главный вызов при проектировании систем на Node.js под высокой нагрузкой — строгие ограничения однопоточного цикла событий (Event Loop) и лимит кучи V8 (по умолчанию около 1.4–2 ГБ). Попытка прочитать 5-гигабайтный файл в оперативную память через fs.readFile() немедленно приводит к фатальному аварийному завершению процесса с ошибкой JavaScript heap out of memory. Индустриальное решение строится на трех столпах: потоковая передача данных (Streams) с гарантией обратного давления (Backpressure), вынос тяжелых асинхронных задач в распределенные очереди BullMQ и защита от деградации СУБД с помощью вероятностного кэширования XFetch в Redis.
Без правильной архитектуры потоков и очередей бэкенд неизбежно сталкивается со следующими проблемами:
Архитектурная схема разделяет быстрый синхронный I/O от тяжелой отложенной обработки:
[ HTTP Client: Загрузка 2 ГБ файла ]
│
▼ (Chunk-by-chunk HTTP Stream)
┌─────────────────────────────────────────────────────────┐
│ Node.js stream.pipeline with Automatic Backpressure │
│ Readable (HTTP) ──> Transform (Gzip/Decrypt) ──> Disk │
└───────────────────────────────┬─────────────────────────┘
│ Файл безопасно сохранен
▼
┌─────────────────────────────────────────────────────────┐
│ BullMQ Producer (Быстрый ответ клиенту: 202 Accepted) │
│ Добавление задачи { fileId, userId } в очередь Redis │
└───────────────────────────────┬─────────────────────────┘
│
▼
┌───────────────────────────┐
│ Redis 7+ Data Store │
│ (Sorted Sets / Streams) │
└─────────────┬─────────────┘
│
┌───────────────────────┴───────────────────────┐
▼ ▼
┌───────────────────────────────┐ ┌───────────────────────────────┐
│ BullMQ Sandboxed Worker #1 │ │ BullMQ Sandboxed Worker #2 │
│ - Изолированный Worker Thread │ │ - Обработка строк данных │
│ - Тяжелый CPU-парсинг │ │ - Сохранение в PostgreSQL │
│ - Отправка прогресса (0-100%) │ │ - Обновление Redis XFetch кэша│
└───────────────────────────────┘ └───────────────────────────────┘Создание отказоустойчивого клиента Redis с автореконнектом, экспоненциальной задержкой и разделением pub/sub соединений.
Использование stream.pipeline для чтения и трансформации больших файлов без утечек памяти и с гарантией Backpressure.
Создание очереди с настройками экспоненциальных повторов (Backoff), лимитов конкурентности и очереди битых задач (Dead Letter Queue).
Реализация обработчика в отдельном Worker Thread для изоляции тяжелых вычислений от основного потока Event Loop.
Реализация сервиса кэширования с ранним упреждающим пересчетом для полного исключения эффекта Cache Stampede.
Почему в высоконагруженных приложениях категорически запрещено использовать readable.pipe(writable) и требуется использовать stream.pipeline()?
Файл читается чанками фиксированного размера (по 64 КБ). Если потребитель занят записью на диск, чтение из сокета временно приостанавливается.
Расход оперативной памяти остается константным (менее 50 МБ) даже при обработке 10-гигабайтного файла.
Сервер регистрирует метаданные файла в очереди BullMQ с уникальным jobId для предотвращения дублирования одинаковых задач.
Задача сразу получает статус 'waiting' в Redis Sorted Set.
Изолированный воркер забирает задачу из Redis через атомарную операцию BRPOPLPUSH, выполняет парсинг и публикует прогресс.
Падение потока воркера не приводит к падению главного HTTP-сервера.
Агрегированные данные кэшируются в Redis. При высокой частоте чтений алгоритм XFetch пересчитывает кэш в фоне за несколько секунд до истечения TTL.
Полное отсутствие просадок производительности при инвалидации горячих ключей.
Ниже представлены исходные файлы высоконагруженного сервиса.
import { Redis, RedisOptions } from 'ioredis';
const redisOptions: RedisOptions = {
host: process.env.REDIS_HOST ?? '127.0.0.1',
port: Number(process.env.REDIS_PORT ?? 6379),
password: process.env.REDIS_PASSWORD ?? undefined,
maxRetriesPerRequest: null, // Обязательное требование BullMQ для блокирующих команд
enableReadyCheck: false,
retryStrategy(times: number) {
const delay = Math.min(times * 100, 3000);
return delay;
},
};
export const redisClient = new Redis(redisOptions);
redisClient.on('connect', () => {
console.info('[Redis] Connection established successfully');
});
redisClient.on('error', (err: unknown) => {
const message = err instanceof Error ? err.message : String(err);
console.error('[Redis] Connection error:', message);
});
export async function closeRedisConnection(): Promise<void> {
await redisClient.quit();
console.info('[Redis] Connection gracefully closed');
}import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { Transform, TransformCallback } from 'node:stream';
import { createGzip } from 'node:zlib';
/**
* Потоковый трансформер для нормализации строк и очистки данных без расхода RAM
*/
class LineSanitizerTransform extends Transform {
private leftover: string = '';
constructor() {
super({ readableObjectMode: false, writableObjectMode: false });
}
public _transform(chunk: Buffer, _encoding: BufferEncoding, callback: TransformCallback): void {
const data = this.leftover + chunk.toString('utf8');
const lines = data.split('\n');
// Сохраняем незавершенную строку для следующего чанка
this.leftover = lines.pop() ?? '';
for (const line of lines) {
if (line.trim().length > 0) {
// Пример трансформации: удаление конфиденциальных данных на лету
const sanitizedLine = line.replace(/\b\d{4}-\d{4}-\d{4}-\d{4}\b/g, '[MASKED_PAN]');
this.push(sanitizedLine + '\n');
}
}
callback();
}
public _flush(callback: TransformCallback): void {
if (this.leftover.trim().length > 0) {
this.push(this.leftover + '\n');
}
callback();
}
}
/**
* Обрабатывает многогигабайтный файл с постоянным расходом памяти менее 50 МБ
*/
export async function processLargeDatasetFile(
inputFilePath: string,
outputGzipPath: string
): Promise<void> {
const readStream = createReadStream(inputFilePath, { highWaterMark: 64 * 1024 }); // 64 КБ буфер
const sanitizer = new LineSanitizerTransform();
const gzipStream = createGzip();
const writeStream = createWriteStream(outputGzipPath);
// stream.pipeline гарантирует Backpressure и безопасное закрытие всех дескрипторов
await pipeline(readStream, sanitizer, gzipStream, writeStream);
}import { Redis } from 'ioredis';
interface CacheRecord<T> {
value: T;
delta: number; // Время расчета значения в миллисекундах
expiry: number; // Абсолютный timestamp истечения срока годности
}
export class ProbabilisticCacheService {
constructor(
private readonly redis: Redis,
private readonly beta: number = 1.0 // Коэффициент агрессивности раннего пересчета (>= 1.0)
) {}
/**
* Получает значение из кэша либо упреждающе пересчитывает его по алгоритму XFetch
*/
public async getOrCompute<T>(
key: string,
ttlSeconds: number,
computeFn: () => Promise<T>
): Promise<T> {
const rawData = await this.redis.get(key);
if (rawData !== null) {
try {
const record: CacheRecord<T> = JSON.parse(rawData);
const now = Date.now();
const ttlRemainingMs = record.expiry - now;
// Формула XFetch: -beta * delta * ln(random())
// Если оставшееся время меньше этого порога, инициируем ранний асинхронный пересчет
const earlyExpirationThreshold = -this.beta * record.delta * Math.log(Math.random());
if (ttlRemainingMs > earlyExpirationThreshold) {
// Кэш свеж и нет необходимости в упреждающем пересчете
return record.value;
}
// Запускаем фоновое обновление без блокировки текущего запроса
void this.recomputeAndSet(key, ttlSeconds, computeFn);
// Мгновенно возвращаем старое значение клиенту
return record.value;
} catch (err: unknown) {
// В случае битого JSON пересчитываем значение синхронно
}
}
return this.recomputeAndSet(key, ttlSeconds, computeFn);
}
private async recomputeAndSet<T>(
key: string,
ttlSeconds: number,
computeFn: () => Promise<T>
): Promise<T> {
const startTime = Date.now();
const value = await computeFn();
const delta = Date.now() - startTime;
const expiry = Date.now() + ttlSeconds * 1000;
const record: CacheRecord<T> = { value, delta, expiry };
// Устанавливаем ключ с запасом в Redis (+60 секунд для страховки от race conditions)
await this.redis.set(key, JSON.stringify(record), 'EX', ttlSeconds + 60);
return value;
}
}import { Queue, QueueEvents } from 'bullmq';
import { redisClient } from '@/infra/redis';
export interface ReportJobPayload {
reportId: string;
userId: string;
dateRange: { start: string; end: string };
}
export const REPORT_QUEUE_NAME = 'report-generation-queue';
// Инициализация очереди
export const reportQueue = new Queue<ReportJobPayload>(REPORT_QUEUE_NAME, {
connection: redisClient,
defaultJobOptions: {
attempts: 3,
backoff: {
type: 'exponential',
delay: 2000, // Первый повтор через 2с, затем 4с, затем 8с
},
removeOnComplete: { age: 3600, count: 1000 },
removeOnFail: { age: 86400 * 7 }, // Храним упавшие задачи 7 дней для аудита
},
});
// Мониторинг событий очереди
export const queueEvents = new QueueEvents(REPORT_QUEUE_NAME, { connection: redisClient });
queueEvents.on('failed', ({ jobId, failedReason }) => {
console.error(`[Queue] Job ${jobId} permanently failed: ${failedReason}`);
});
export async function addReportJob(payload: ReportJobPayload): Promise<string> {
const job = await reportQueue.add('generate-monthly-report', payload, {
jobId: `report_${payload.reportId}`, // Дедупликация: предотвращает дублирование задачи
});
return job.id!;
}import { Worker, Job } from 'bullmq';
import { redisClient } from '@/infra/redis';
import { REPORT_QUEUE_NAME, ReportJobPayload } from '@/queues/report-queue';
export function startReportWorker(): Worker<ReportJobPayload> {
const worker = new Worker<ReportJobPayload>(
REPORT_QUEUE_NAME,
async (job: Job<ReportJobPayload>) => {
console.info(`[Worker] Starting processing job ${job.id} for user ${job.data.userId}`);
// Эмуляция пошаговой генерации отчета с обновлением прогресса
for (let step = 1; step <= 5; step++) {
await new Promise((resolve) => setTimeout(resolve, 800));
await job.updateProgress(step * 20);
}
console.info(`[Worker] Job ${job.id} completed successfully`);
return { downloadUrl: `https://storage.example.com/reports/${job.data.reportId}.pdf` };
},
{
connection: redisClient,
concurrency: 5, // Не более 5 параллельных задач на один процесс воркера
stalledInterval: 30000, // Проверка зависших задач каждые 30 секунд
maxStalledCount: 1, // Если воркер завис и не отправил heartbeat, задача передается другому
}
);
worker.on('completed', (job: Job<ReportJobPayload>) => {
console.info(`[Worker] Job ${job.id} has finished processing.`);
});
worker.on('error', (err: Error) => {
console.error('[Worker] Fatal error in worker pool:', err);
});
return worker;
}Конструкция readable.pipe(writable) не закрывает файловые дескрипторы при возникновении ошибки посередине конвейера. В результате сокеты зависают в состоянии CLOSE_WAIT, а дескрипторы файлов утекают вплоть до превышения лимита ОС EMFILE: too many open files. Всегда используйте stream.pipeline() с промисами из node:stream/promises.
Некоторые разработчики считают, что создание воркера BullMQ автоматически запускает код в отдельном потоке. Это не так: стандартный воркер выполняется в том же цикле событий Node.js! Если задача содержит расчет хешей bcrypt в синхронном цикле, весь HTTP-сервер заморозится. Для CPU-интенсивных задач передавайте путь к файлу обработчика строкой (Sandboxed Process), либо используйте нативный worker_threads.
Команда KEYS * выполняет синхронный полный перебор всех ключей в однопоточном движке Redis. Если база содержит более 500 000 ключей, Redis заблокируется на несколько секунд, вызывая тайм-ауты во всех микросервисах компании. Для итеративного поиска ключей используйте исключительно SCAN с курсором.
BullMQ использует блокирующие команды Redis (BRPOPLPUSH, BLMOVE) для ожидания новых задач без расхода CPU. Если в опциях ioredis флаг maxRetriesPerRequest не установлен строго в null, клиент выбросит исключение при блокирующем ожидании, разрушив процесс воркера.
В сервисе используется стандартное кэширование в Redis с жестким временем жизни ключа EX 300 (5 минут). Ключ запрашивается 10 000 раз в секунду. Что произойдет ровно в момент наступления 300-й секунды?