Асинхронное программирование в Node.js: события, обработчики, генераторы

Неблокирующие функции

Основная «фишка» node.js, как известно — большинство функций в ней неблокирующие. Что это значит для программиста?

Операции в программе отнимают разное время в зависимости от того к чему мы обращаемся. Операции с регистрами — самые быстрые, потом идут операции с кэшами первого и второго уровней (1 и 5 наносекунд соответственно), операции с RAM (~80-90 нс), операции с жёстким диском (~14 микросекунд). Чтобы понять масштаб, можно взглянуть вот на этот gif. Огромная колонна — обращение к жёсткому диску. Если сделать zoom in, будут видны операции с памятью и кэшами. Обращение к сетевому серверу ещё дольше — когда придёт ответ, вообще непонятно.

Философия неблоокирующего ввода-вывода состоит в том что функции вообще не должны ждать окончания длительных операций ввода-вывода (диск и сеть). Для этого функция либо принимает функцию, которую надо выполнить по завершении операции (callback), либо возвращает объект — генератор событий, на который опять же вешаются функции. Пример:

// Первый способ: оставление обработчика
server.query('Query text', function(err, data) {
    // Если не передано исключение, сделать что нибудь с данными
});

// Второй способ: подключение обработчиков к генератору событий
var query_status = server.query('Query text');
query_status.addListener('success', function(data){
    // Сделать что нибудь с данными
});

query_status.addListener('error', function(error){
    // Если пришла ошибка, тоже что нибудь сделать
});

Генераторы событий дают больше возможностей — они позволяют вешать разные функции на разные события. Callback-стиль, в свою очередь, выглядит более компактно, его удобнее использовать для простых вызовов. Ранее в Node был ещё класс promise — особый случай эмиттера, у которого есть только три события: success, error, cancel. Promises давно выброшены из ядра и не применяются, поэтому я здесь их рассматривать не буду 🙂

В Node v0.2.0 вместо метода addListener рекомендуется использовать короткий синоним on():

var sys = require("sys");

var connection = my_connector(1445, 'hello'); // Получили эмиттер

connection.on('success', function(data) {
    sys.puts('Data received: ' + data); // Обработчик успешного завершения
});

connection.on('error', function(error) {
    sys.puts('Error happened: ' + error); // Обработчик ошибок
});

Более длинный addListener скорее всего будет полностью исключен в следующих версиях Node. Поэтому далее я везде буду использовать on().

В некоторых старых модулях ещё можно найти использование promises, которые сейчас вынесены в отдельный сторонний модуль.

Пишем асинхронный код

Если мы хотим написать асинхронную функцию, нам надо как можно быстрее отдать управление в основной код, вернув какое-либо значение. Например, мы пишем библиотеку для обращения к какому то сервису:

var net = require("net");

var my_connector = function(port, data, callback) {
    var server_conn = tcp.createConnection(port);
    server_conn.on('connect', function () {
        // Подключились, получили объект потока
        server_conn.write(data, 'ascii');
        //Отправили данные, ждём ответа
        server_conn.on('data', function(answer) {
            callback(false, answer);
            // Закрыли поток после получения данных
            server_conn.end();
        }
    }
    return true;
}

Здесь возврат произойдёт сразу после установки первого внутреннего callback’а — того, который ждёт подключения. Скрипт будет выполняться дальше, а переданная нами функция будет терпеливо дожидаться конца операции. Этот подход работает, пока нам нужно обработать только одно событие — завершение операции. Чтобы хотя бы сигнализировать о том была ли операция успешной, нам придётся передавать статус в обработчик, вроде этого:

var net = require("net");

var my_connector = function(port, data, callback) {
    var server_conn = net.createConnection(port);
    server_conn.on('data', function(answer) {
        callback(null, answer); // Первый параметр - null, признак успешной операции
    }

    // Соединение закрыто
    server_conn.on('end', function(haserror) {
        if (haserror) {
            callback(new Error('Connection error'), null); // Первый параметр - объект ошибки при соединении
        } else {
            callback(new Error('Connection closed'), null); // Первый параметр - объект ошибки закрытия соединения
        }
    }
    return true;
}

Это не очень красиво выглядит, но будет работать, пока не понадобится обрабатывать пять-шесть различных ситуаций. После этого код обработки разрастётся до неприличия.

В Node принято передавать обработчику первым параметром объект ошибки (либо null если операция произошла успешно), либо возвращать объект генератора событий.

Событие генерируется методом event.emit('eventname', param1, param2...);, обработчики вешаются методом addListener() с указанием названия события:

var sys = require("sys");

var connection = my_connector(1445, 'hello'); // Получили эмиттер

connection.on('success', function(data) {
    sys.puts('Data received: ' + data); // Обработчик успешного завершения
});

connection.on('error', function(error) {
    sys.puts('Error happened: ' + error); // Обработчик ошибок
});

Управление обработчиками

Кроме назначения обработчиков бывает нужно снять некоторые из них или просто узнать какие обработчики сейчас действуют (потому что node-скрипт не завершится, если есть хоть один обработчик на действующем генераторе событий).

Получить список обработчиков для события можно с помощью метода listeners:

server.on('stream', function (stream) {
  console.log('someone connected!');
});
console.log(sys.inspect(server.listeners('stream'));

Методу передаётся идентификатор события, он возвращает массив функций. Чтобы снять один из обработчиков, надо передать функцию вместе с строкой события методу removeListener:

var callback = function(stream) {
  console.log('someone connected!');
};
server.on('connection', callback);
// ...
server.removeListener('connection', callback);

Снять все обработчики с события можно методом removeAllListeners:

var callback = function(stream) {
  console.log('someone connected!');
};
server.on('stream', callback);
// ...
server.removeAllListeners('stream');

Иногда требуется снять текущий обработчик и повесить вместо него новый. Например, мы ждём от TCP-сервера приветствие, и обработчик приветствия навешивает обработчик реальных данных ответа. В таких случаях удобнее снять обработчик через this, а не лезть за ним в массив возвращаемый listeners:

stream.on('data', function(data) {
    if (data == "hello") {
        // Снять текущий обработчик с события
        stream.removeListener('data', this);
        // Заменить его другим
        stream.on('data', function(data) {
            sys.puts('Data: ' + data);
        });
    } else {
        sys.puts('Wrong header received');
    }
});