Какие знаешь методы управления асинхронными операциями в Node.JS?

Методы управления асинхронными операциями в Node.js

В Node.js асинхронность — это сердце платформы. За более чем 10 лет я видел эволюцию от callback'ов к Promises и async/await. Расскажу обо всех методах с примерами из реальных проектов.

1. Callbacks — фундамент (старый, но всё ещё используется)

Callback — это функция, которая вызывается после завершения асинхронной операции.

// Node.js fs API — callback style
const fs = require('fs');

fs.readFile('./data.json', 'utf8', (error, data) => {
  if (error) {
    console.error('Ошибка при чтении файла:', error);
    return;
  }
  console.log('Содержание файла:', data);
});

// Callback hell (пирамида ада)
fs.readFile('./file1.json', 'utf8', (err1, data1) => {
  if (err1) throw err1;
  
  fs.readFile('./file2.json', 'utf8', (err2, data2) => {
    if (err2) throw err2;
    
    fs.readFile('./file3.json', 'utf8', (err3, data3) => {
      if (err3) throw err3;
      console.log(data1, data2, data3);
    });
  });
});
// ❌ Нечитаемо, сложно отладить

Проблемы:

• Callback hell (pyramid of doom)
• Сложная обработка ошибок
• Трудно отследить поток выполнения

2. Promises — значительное улучшение

Promise — объект, который представляет значение, которое может быть доступно сейчас, в будущем или никогда.

const fs = require('fs').promises;  // API на основе Promise

const readFiles = () => {
  return fs.readFile('./file1.json', 'utf8')
    .then(data1 => {
      console.log('File 1:', data1);
      return fs.readFile('./file2.json', 'utf8');
    })
    .then(data2 => {
      console.log('File 2:', data2);
      return fs.readFile('./file3.json', 'utf8');
    })
    .then(data3 => {
      console.log('File 3:', data3);
      return { file1: data1, file2: data2, file3: data3 };
    })
    .catch(error => {
      console.error('Ошибка:', error);
    });
};

readFiles();

Преимущества:

• Чайнинг (.then().then())
• Единая обработка ошибок (.catch())
• Более читаемо

Три состояния Promise:

const promise = new Promise((resolve, reject) => {
  // pending (ожидание)
  setTimeout(() => {
    // fulfilled (выполнено)
    resolve('success');
    // или rejected (отклонено)
    // reject(new Error('failed'));
  }, 1000);
});

promise
  .then(result => console.log(result))      // 'success'
  .catch(error => console.error(error))     // обработка ошибок
  .finally(() => console.log('done'));      // выполнится в любом случае

3. Async/Await — синтаксический сахар (рекомендуется)

Async/await — это современный способ работы с Promise'ами. Код выглядит как синхронный.

const fs = require('fs').promises;

// Ключевое слово async — функция возвращает Promise
async function readAllFiles() {
  try {
    // await — ждём результата
    const data1 = await fs.readFile('./file1.json', 'utf8');
    console.log('File 1:', data1);
    
    const data2 = await fs.readFile('./file2.json', 'utf8');
    console.log('File 2:', data2);
    
    const data3 = await fs.readFile('./file3.json', 'utf8');
    console.log('File 3:', data3);
    
    return { data1, data2, data3 };
  } catch (error) {
    console.error('Ошибка:', error);
    throw error;  // пробросить ошибку выше
  } finally {
    console.log('Операция завершена');
  }
}

// Использование
await readAllFiles();

Параллельное выполнение с async/await:

async function parallelRead() {
  try {
    // Последовательное выполнение (медленно)
    const data1 = await fs.readFile('./file1.json', 'utf8');
    const data2 = await fs.readFile('./file2.json', 'utf8');
    const data3 = await fs.readFile('./file3.json', 'utf8');
    // общее время: sum(все времена)
    
    // Параллельное выполнение (быстро)
    const [file1, file2, file3] = await Promise.all([
      fs.readFile('./file1.json', 'utf8'),
      fs.readFile('./file2.json', 'utf8'),
      fs.readFile('./file3.json', 'utf8')
    ]);
    // общее время: max(одного времени)
    
    return { file1, file2, file3 };
  } catch (error) {
    console.error('Ошибка:', error);
  }
}

4. Event Emitter — для событий

EventEmitter используется для асинхронной работы с событиями в Node.js.

const EventEmitter = require('events');

class DataProcessor extends EventEmitter {
  async process(data) {
    try {
      this.emit('start', { message: 'Processing started' });
      
      // долгая операция
      await this.longOperation(data);
      
      this.emit('complete', { result: 'success' });
    } catch (error) {
      this.emit('error', error);
    }
  }
  
  async longOperation(data) {
    return new Promise(resolve => {
      setTimeout(() => {
        this.emit('progress', { percent: 50 });
        resolve();
      }, 1000);
    });
  }
}

const processor = new DataProcessor();

processor.on('start', (data) => console.log('Started:', data));
processor.on('progress', (data) => console.log('Progress:', data.percent));
processor.on('complete', (data) => console.log('Done:', data));
processor.on('error', (error) => console.error('Error:', error));

await processor.process('some data');

5. Streams — для больших объёмов данных

Streams позволяют обрабатывать данные порциями (chunks), не загружая всё в память.

const fs = require('fs');

// Чтение файла потоком (не загружает всё в память)
const readStream = fs.createReadStream('./large-file.json', {
  highWaterMark: 64 * 1024  // 64KB chunks
});

readStream.on('data', (chunk) => {
  console.log('Получена порция:', chunk.length, 'bytes');
  // обработка chunk
});

readStream.on('end', () => {
  console.log('Файл полностью прочитан');
});

readStream.on('error', (error) => {
  console.error('Ошибка:', error);
});

// Pipe — подключить потоки
const readStream2 = fs.createReadStream('./source.json');
const writeStream = fs.createWriteStream('./copy.json');

readStream2.pipe(writeStream);

// Transform stream — преобразовать данные
const { Transform } = require('stream');

const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

readStream2.pipe(upperCaseTransform).pipe(writeStream);

6. Worker Threads — для CPU-intensive операций

Worker Threads позволяют выполнять тяжёлые вычисления в отдельных потоках, не блокируя основной.

const { Worker } = require('worker_threads');
const path = require('path');

async function calculateInWorker(data) {
  return new Promise((resolve, reject) => {
    const worker = new Worker(path.join(__dirname, './worker.js'));
    
    worker.on('message', (result) => {
      resolve(result);
      worker.terminate();
    });
    
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0) {
        reject(new Error(`Worker stopped with exit code ${code}`));
      }
    });
    
    // отправить данные в worker
    worker.postMessage({ data, iterations: 1000000 });
  });
}

// worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', (message) => {
  const { data, iterations } = message;
  
  // тяжёлые вычисления
  let result = 0;
  for (let i = 0; i < iterations; i++) {
    result += Math.sqrt(data * i);
  }
  
  // отправить результат обратно
  parentPort.postMessage({ result });
});

// Использование
const result = await calculateInWorker(100);
console.log('Результат:', result);

7. Queue и Bull — для управления очередями

Bull — популярная библиотека для обработки очередей асинхронных задач.

const Queue = require('bull');
const redis = require('redis');

// Создание очереди
const emailQueue = new Queue('email-sending', {
  redis: { host: 'localhost', port: 6379 }
});

// Обработка задач из очереди
emailQueue.process(async (job) => {
  const { email, subject, body } = job.data;
  
  // Отправка email
  await sendEmail(email, subject, body);
  
  return { success: true, emailSent: email };
});

// Добавление задачи в очередь
async function queueEmailJob(email, subject, body) {
  const job = await emailQueue.add(
    { email, subject, body },
    {
      attempts: 3,           // 3 попытки
      backoff: { type: 'exponential', delay: 2000 },
      removeOnComplete: true // удалить после успеха
    }
  );
  
  return job.id;
}

// Слушание событий очереди
emailQueue.on('completed', (job) => {
  console.log(`Job ${job.id} completed`);
});

emailQueue.on('failed', (job, error) => {
  console.error(`Job ${job.id} failed:`, error.message);
});

// Использование
await queueEmailJob('user@example.com', 'Welcome', 'Hello!');

8. RxJS — для реактивного программирования

RxJS используется для работы с потоками событий (observables).

const { interval, fromEvent, BehaviorSubject } = require('rxjs');
const { map, filter, take } = require('rxjs/operators');

// Обычное значение
const counter$ = new BehaviorSubject(0);
counter$.subscribe(value => console.log('Counter:', value));
counter$.next(1);  // Counter: 1
counter$.next(2);  // Counter: 2

// Observable от события
const button = document.getElementById('myButton');
const click$ = fromEvent(button, 'click');

click$
  .pipe(
    map(() => 1),
    filter((_, i) => i % 2 === 0),  // каждый второй
    take(5)  // только 5 событий
  )
  .subscribe(() => console.log('Button clicked'));

// Интервал
const tick$ = interval(1000);
tick$
  .pipe(take(5))
  .subscribe(value => console.log('Tick:', value));

9. Async Control Flow — управление потоком

// Последовательное выполнение
async function sequential() {
  const result1 = await operation1();
  const result2 = await operation2(result1);  // зависит от result1
  const result3 = await operation3(result2);
  return result3;
}

// Параллельное выполнение
async function parallel() {
  const [r1, r2, r3] = await Promise.all([
    operation1(),
    operation2(),
    operation3()
  ]);
  return { r1, r2, r3 };
}

// Гибридное выполнение
async function hybrid() {
  // 1. Параллельно
  const [r1, r2] = await Promise.all([
    operation1(),
    operation2()
  ]);
  
  // 2. Затем последовательно
  const r3 = await operation3(r1, r2);
  
  return r3;
}

// С timeout
async function withTimeout() {
  const timeout = new Promise((_, reject) =>
    setTimeout(() => reject(new Error('Timeout')), 5000)
  );
  
  try {
    const result = await Promise.race([
      operation1(),
      timeout
    ]);
    return result;
  } catch (error) {
    console.error('Operation timed out');
  }
}

10. Сравнение методов

11. Лучшие практики

// ✅ ХОРОШО: async/await, правильная обработка ошибок
async function bestPractice() {
  try {
    const user = await getUserFromDB(userId);
    const posts = await getPostsFromDB(user.id);
    
    // Параллельные независимые запросы
    const [comments, likes] = await Promise.all([
      getComments(posts[0].id),
      getLikes(posts[0].id)
    ]);
    
    return { user, posts, comments, likes };
  } catch (error) {
    console.error('Failed to load dashboard:', error);
    throw error;  // пробросить для обработки выше
  }
}

// ❌ ПЛОХО: слишком последовательно
async function slowApproach() {
  const user = await getUserFromDB(userId);
  const posts = await getPostsFromDB(user.id);
  const comments = await getComments(posts[0].id);  // ждём
  const likes = await getLikes(posts[0].id);        // ждём
  return { user, posts, comments, likes };
}

Заключение

В Node.js есть много способов управления асинхронностью. Мой выбор зависит от задачи:

1. Обычные асинхронные операции → async/await (основной выбор)
2. События и потоки → EventEmitter + Streams
3. Фоновые задачи → Bull/RabbitMQ
4. Тяжёлые вычисления → Worker Threads
5. Сложная реактивность → RxJS

На базе 10+ лет опыта рекомендую: async/await для большинства случаев. Это современно, читаемо и безопасно.