EventEmitter trong Node.js: on, emit, once và sự kiện tự tạo

EventEmitter là lớp trong module events của Node, cho phép một đối tượng phát ra các sự kiện có tên. Bạn đăng ký một listener bằng emitter.on('name', fn) và kích hoạt nó bằng emitter.emit('name', ...args); mọi listener của tên đó đều chạy, theo thứ tự được thêm vào, với các đối số bạn truyền vào. Máy chủ HTTP, stream và process đều là EventEmitter, nên vài phương thức này dùng được ở mọi nơi trong Node.

const EventEmitter = require('events');

const shop = new EventEmitter();

shop.on('order', (item, qty) => {
    console.log('order received:', qty, 'x', item);
});
shop.on('order', (item) => {
    console.log('update stock for', item);
});

console.log('before emit');
shop.emit('order', 'coffee', 2);
console.log('after emit');

Cả hai listener đều chạy giữa before emit và after emit: emit gọi chúng một cách đồng bộ và chỉ trả về khi chúng đã chạy xong. Đổi chỗ hai lời gọi shop.on thì hai dòng ở giữa cũng đổi chỗ theo.

on, once và off

on giữ listener cho đến khi bạn gỡ nó ra, once gỡ listener sau lần gọi đầu tiên, và off (giống removeListener) gỡ một listener. Để gỡ một listener, bạn cần đúng hàm đã thêm vào, nên hãy đặt tên cho hàm thay vì truyền trực tiếp một arrow function.

const EventEmitter = require('events');
const door = new EventEmitter();

function logOpen(who) {
    console.log('opened by', who);
}

door.on('open', logOpen);
door.once('open', () => console.log('first open ever'));

door.emit('open', 'Ada');
door.emit('open', 'Linus');

door.off('open', logOpen);
console.log('listeners left:', door.listenerCount('open'));
console.log('emit returned', door.emit('open', 'Grace'));

first open ever được in một lần, sau Ada. Khi logOpen đã bị gỡ thì không còn ai lắng nghe, nên lệnh emit cuối cùng trả về false; door.removeAllListeners('open') sẽ xóa cả hai listener chỉ trong một lời gọi.

Sự kiện error

'error' là sự kiện đặc biệt. Nếu bạn emit nó mà không ai lắng nghe, emit sẽ ném lỗi đó ra, và trong một chương trình thực tế, lỗi không được bắt sẽ làm tiến trình crash. Hãy luôn thêm listener 'error' cho các emitter có thể thất bại, như stream và máy chủ.

const EventEmitter = require('events');
const job = new EventEmitter();

try {
    job.emit('error', new Error('disk full'));
} catch (err) {
    console.log('emit threw:', err.message);
}

job.on('error', (err) => {
    console.log('handled:', err.message);
});
job.emit('error', new Error('disk full'));
console.log('still running');

Khi không có listener, emit ném lỗi và catch bắt được nó; khi đã thêm listener, cùng lệnh emit đó chỉ gọi listener và chương trình tiếp tục chạy. Mọi tên sự kiện khác không có listener đều bị bỏ qua một cách im lặng.

Kế thừa EventEmitter trong lớp của bạn

Cách dùng thông thường là cho lớp của bạn kế thừa EventEmitter, gọi super() trong hàm khởi tạo và gọi this.emit khi có điều gì xảy ra. Nơi gọi sau đó đăng ký lắng nghe đối tượng của bạn như với mọi đối tượng Node khác.

const EventEmitter = require('events');

class Upload extends EventEmitter {
    constructor(name, chunks) {
        super();
        this.name = name;
        this.chunks = chunks;
    }

    start() {
        for (let i = 1; i <= this.chunks; i++) {
            this.emit('progress', Math.round((i / this.chunks) * 100));
        }
        this.emit('done', this.name);
    }
}

const up = new Upload('photo.png', 4);
up.on('progress', (pct) => console.log(pct + '%'));
up.once('done', (name) => console.log(name, 'uploaded'));
up.start();

Kết quả in ra là 25%, 50%, 75%, 100% rồi photo.png uploaded. Chuyển up.start() lên trên hai dòng up.on thì không có gì được in: các sự kiện được phát ra trước khi có ai lắng nghe.

Cái bẫy tương tự xảy ra khi emit từ hàm khởi tạo. Các thư viện cần emit ngay lập tức sẽ bọc lời gọi trong process.nextTick, vốn chạy sau các dòng on của nơi gọi (xem trang event loop).

Chờ một sự kiện với events.once

events.once(emitter, name) trả về một promise được resolve với các đối số của sự kiện, dưới dạng mảng, ở lần tiếp theo sự kiện được phát ra. Nó biến một sự kiện xảy ra một lần thành thứ bạn có thể await, và sẽ reject nếu emitter phát ra 'error' trước.

import { EventEmitter, once } from 'events';

const db = new EventEmitter();

setTimeout(() => db.emit('ready', 'v1.4', 3), 100);

console.log('waiting for ready');
const [version, tables] = await once(db, 'ready');
console.log('ready:', version, 'with', tables, 'tables');

await tạm dừng cho đến khi timer phát ra ready, rồi mảng được destructure thành hai đối số. Với một sự kiện được phát ra nhiều lần, events.on(emitter, name) cho bạn một async iterator để lặp qua bằng for await.

Những lỗi thường gặp

Thêm listener bên trong một hàm chạy thường xuyên. Gọi on trong mọi request handler sẽ làm chồng chất listener và tốn bộ nhớ. Hãy thêm một lần, hoặc dùng once.

Gỡ một hàm viết trực tiếp. off('data', () => ...) không gỡ được gì, vì arrow function đó là một hàm mới. Hãy giữ tham chiếu tới hàm bạn đã thêm.

Nghĩ rằng listener sẽ chạy sau. emit là đồng bộ, nên một listener chạy chậm sẽ giữ chân đoạn code đã emit. Đánh dấu listener là async cũng không giúp gì, vì code của nó vẫn chạy trên cùng luồng cho đến await đầu tiên. Hãy chia công việc nặng thành nhiều phần với setImmediate ở giữa, hoặc chuyển nó sang một worker thread.