Async Pattern [4/4] - Pub-Sub Pattern trong JavaScript / TypeScript

1. Pub-Sub Pattern là gì?

Pub-Sub (Publisher-Subscriber) Pattern là một mẫu thiết kế hành vi trong đó publishers (nhà phát hành) không gửi thông báo trực tiếp đến subscribers (người đăng ký) cụ thể. Thay vào đó, các thông báo được phân loại thành các kênh (channels) hoặc chủ đề (topics) mà không cần biết có những subscribers nào đang lắng nghe.

Điểm khác biệt chính giữa Pub-Sub và Observer Pattern là:

Trong Observer Pattern, subject (chủ thể) biết rõ về observers (người quan sát) của nó
Trong Pub-Sub Pattern, publishers và subscribers không biết về sự tồn tại của nhau, chúng giao tiếp thông qua một trung gian (message broker/event bus)


  graph LR
    A[Publisher] -->|Publish| B[Event Channel/Broker]
    B -->|Notify| C[Subscriber 1]
    B -->|Notify| D[Subscriber 2]
    B -->|Notify| E[Subscriber 3]

    style A fill:#f9f,stroke:#333,stroke-width:2px
    style B fill:#bbf,stroke:#333,stroke-width:2px
    style C fill:#dfd,stroke:#333,stroke-width:1px
    style D fill:#dfd,stroke:#333,stroke-width:1px
    style E fill:#dfd,stroke:#333,stroke-width:1px

2. Cấu trúc của Pub-Sub Pattern

Publisher: Đối tượng tạo ra thông báo và gửi đến kênh
Subscriber: Đối tượng đăng ký với kênh để nhận thông báo
Message Broker/Event Bus: Trung gian quản lý việc đăng ký và phân phối thông báo
Channel/Topic: Phân loại thông báo theo chủ đề

3. Triển khai Pub-Sub Pattern trong JavaScript

3.1. Triển khai cơ bản

class PubSub {
  constructor() {
    this.channels = {};
  }

  // Đăng ký subscriber với một kênh
  subscribe(channel, subscriber) {
    if (!this.channels[channel]) {
      this.channels[channel] = [];
    }

    this.channels[channel].push(subscriber);
    return () => this.unsubscribe(channel, subscriber);
  }

  // Hủy đăng ký subscriber từ một kênh
  unsubscribe(channel, subscriber) {
    if (!this.channels[channel]) return;

    this.channels[channel] = this.channels[channel].filter(
      (sub) => sub !== subscriber
    );
  }

  // Publish thông báo đến một kênh
  publish(channel, data) {
    if (!this.channels[channel]) return;

    this.channels[channel].forEach((subscriber) => {
      subscriber(data);
    });
  }
}

// Sử dụng
const pubsub = new PubSub();

// Tạo subscribers
const userSubscriber = (data) => {
  console.log(`User event received: ${JSON.stringify(data)}`);
};

const logSubscriber = (data) => {
  console.log(`Log event: ${JSON.stringify(data)}`);
};

// Đăng ký subscribers với các kênh
pubsub.subscribe("user:login", userSubscriber);
pubsub.subscribe("user:logout", userSubscriber);
pubsub.subscribe("log", logSubscriber);

// Publish thông báo
pubsub.publish("user:login", { userId: 123, username: "john_doe" });
pubsub.publish("log", { level: "info", message: "User logged in" });
pubsub.publish("user:logout", { userId: 123, username: "john_doe" });

3.2. Triển khai với TypeScript

type Subscriber<T> = (data: T) => void;
type Unsubscribe = () => void;

class PubSub<T = any> {
  private channels: Record<string, Subscriber<T>[]> = {};

  subscribe(channel: string, subscriber: Subscriber<T>): Unsubscribe {
    if (!this.channels[channel]) {
      this.channels[channel] = [];
    }

    this.channels[channel].push(subscriber);

    // Trả về hàm để hủy đăng ký
    return () => this.unsubscribe(channel, subscriber);
  }

  unsubscribe(channel: string, subscriber: Subscriber<T>): void {
    if (!this.channels[channel]) return;

    this.channels[channel] = this.channels[channel].filter(
      (sub) => sub !== subscriber
    );
  }

  publish(channel: string, data: T): void {
    if (!this.channels[channel]) return;

    this.channels[channel].forEach((subscriber) => {
      subscriber(data);
    });
  }
}

// Định nghĩa các kiểu dữ liệu cho các kênh
interface UserEvent {
  userId: number;
  username: string;
}

interface LogEvent {
  level: "info" | "warning" | "error";
  message: string;
}

// Sử dụng với kiểu dữ liệu cụ thể
const userPubSub = new PubSub<UserEvent>();
const logPubSub = new PubSub<LogEvent>();

// Đăng ký subscribers
const userLoginSubscriber = (data: UserEvent) => {
  console.log(`User logged in: ${data.username}`);
};

const logSubscriber = (data: LogEvent) => {
  console.log(`[${data.level.toUpperCase()}] ${data.message}`);
};

// Đăng ký và lưu hàm hủy đăng ký
const unsubscribeUserLogin = userPubSub.subscribe("login", userLoginSubscriber);
logPubSub.subscribe("system", logSubscriber);

// Publish thông báo
userPubSub.publish("login", { userId: 123, username: "john_doe" });
logPubSub.publish("system", { level: "info", message: "System started" });

// Hủy đăng ký
unsubscribeUserLogin();

4. Ví dụ thực tế: Hệ thống thông báo trong ứng dụng

Dưới đây là một ví dụ về hệ thống thông báo trong ứng dụng sử dụng Pub-Sub Pattern:

// Định nghĩa kiểu dữ liệu cho các thông báo
interface Notification {
  id: string;
  type: "info" | "success" | "warning" | "error";
  message: string;
  timestamp: Date;
}

// Hệ thống thông báo sử dụng Pub-Sub
class NotificationSystem {
  private pubsub = new PubSub<Notification>();

  // Đăng ký nhận thông báo
  subscribe(
    type: "info" | "success" | "warning" | "error" | "all",
    callback: (notification: Notification) => void
  ): Unsubscribe {
    return this.pubsub.subscribe(type, callback);
  }

  // Gửi thông báo
  notify(
    type: "info" | "success" | "warning" | "error",
    message: string
  ): void {
    const notification: Notification = {
      id: this.generateId(),
      type,
      message,
      timestamp: new Date(),
    };

    // Publish thông báo đến kênh cụ thể
    this.pubsub.publish(type, notification);

    // Publish thông báo đến kênh 'all'
    this.pubsub.publish("all", notification);
  }

  private generateId(): string {
    return Math.random().toString(36).substring(2, 9);
  }
}

// Sử dụng hệ thống thông báo
const notificationSystem = new NotificationSystem();

// UI Component hiển thị thông báo
class NotificationUI {
  constructor(private container: HTMLElement) {
    // Đăng ký nhận tất cả thông báo
    notificationSystem.subscribe("all", this.displayNotification.bind(this));
  }

  private displayNotification(notification: Notification): void {
    const element = document.createElement("div");
    element.className = `notification ${notification.type}`;
    element.innerHTML = `
      <strong>${notification.type.toUpperCase()}</strong>: 
      ${notification.message}
      <small>${notification.timestamp.toLocaleTimeString()}</small>
    `;

    this.container.appendChild(element);

    // Tự động xóa thông báo sau 5 giây
    setTimeout(() => {
      element.classList.add("fade-out");
      setTimeout(() => element.remove(), 500);
    }, 5000);
  }
}

// Logger ghi lại thông báo lỗi
class ErrorLogger {
  constructor() {
    // Chỉ đăng ký nhận thông báo lỗi
    notificationSystem.subscribe("error", this.logError.bind(this));
  }

  private logError(notification: Notification): void {
    console.error(
      `ERROR [${notification.timestamp.toISOString()}]: ${notification.message}`
    );
    // Trong thực tế, có thể gửi lỗi đến hệ thống giám sát
  }
}

// Khởi tạo các thành phần
const notificationUI = new NotificationUI(
  document.getElementById("notifications") as HTMLElement
);
const errorLogger = new ErrorLogger();

// Tạo thông báo
notificationSystem.notify("info", "Ứng dụng đã khởi động");
notificationSystem.notify("success", "Dữ liệu đã được lưu thành công");
notificationSystem.notify("warning", "Kết nối mạng không ổn định");
notificationSystem.notify("error", "Không thể kết nối đến máy chủ");

5. Pub-Sub trong các framework và thư viện

5.1. Pub-Sub trong Node.js với EventEmitter

Node.js có sẵn module events với class EventEmitter hoạt động tương tự như Pub-Sub:

const EventEmitter = require("events");

class OrderSystem extends EventEmitter {
  placeOrder(product, user) {
    // Xử lý đặt hàng
    console.log(`Order placed: ${product} for ${user}`);

    // Emit sự kiện với dữ liệu
    this.emit("order_placed", { product, user, date: new Date() });
  }

  shipOrder(product, user) {
    // Xử lý vận chuyển
    console.log(`Order shipped: ${product} to ${user}`);

    // Emit sự kiện với dữ liệu
    this.emit("order_shipped", { product, user, date: new Date() });
  }
}

const orderSystem = new OrderSystem();

// Đăng ký listeners (subscribers)
orderSystem.on("order_placed", (data) => {
  console.log(`Notification: New order at ${data.date}`);
  // Gửi email xác nhận đơn hàng
});

orderSystem.on("order_shipped", (data) => {
  console.log(`Notification: Order for ${data.user} has been shipped`);
  // Gửi thông báo vận chuyển
});

// Thêm nhiều listeners cho cùng một sự kiện
orderSystem.on("order_placed", (data) => {
  console.log(`Inventory update: Reduce stock for ${data.product}`);
  // Cập nhật kho hàng
});

// Sử dụng
orderSystem.placeOrder("Laptop", "[email protected]");
orderSystem.shipOrder("Laptop", "[email protected]");

5.2. Pub-Sub trong React với Context API

React Context API có thể được sử dụng để triển khai Pub-Sub Pattern:

import React, { createContext, useContext, useState, useCallback } from "react";

// Tạo context
const PubSubContext = createContext(null);

// Provider component
export const PubSubProvider = ({ children }) => {
  const [subscribers, setSubscribers] = useState({});

  const subscribe = useCallback((channel, callback) => {
    setSubscribers((prev) => {
      const channelSubscribers = prev[channel] || [];
      return {
        ...prev,
        [channel]: [...channelSubscribers, callback],
      };
    });

    // Trả về hàm hủy đăng ký
    return () => {
      setSubscribers((prev) => {
        const channelSubscribers = prev[channel] || [];
        return {
          ...prev,
          [channel]: channelSubscribers.filter((cb) => cb !== callback),
        };
      });
    };
  }, []);

  const publish = useCallback(
    (channel, data) => {
      const channelSubscribers = subscribers[channel] || [];
      channelSubscribers.forEach((callback) => callback(data));
    },
    [subscribers]
  );

  return (
    <PubSubContext.Provider value={{ subscribe, publish }}>
      {children}
    </PubSubContext.Provider>
  );
};

// Hook để sử dụng PubSub
export const usePubSub = () => {
  const context = useContext(PubSubContext);
  if (!context) {
    throw new Error("usePubSub must be used within a PubSubProvider");
  }
  return context;
};

// Sử dụng trong component
const NotificationButton = () => {
  const { publish } = usePubSub();

  const handleClick = () => {
    publish("notification", {
      type: "success",
      message: "Operation completed successfully",
    });
  };

  return <button onClick={handleClick}>Notify Success</button>;
};

const NotificationList = () => {
  const { subscribe } = usePubSub();
  const [notifications, setNotifications] = useState([]);

  useEffect(() => {
    // Đăng ký nhận thông báo
    const unsubscribe = subscribe("notification", (data) => {
      setNotifications((prev) => [...prev, data]);
    });

    // Hủy đăng ký khi component unmount
    return unsubscribe;
  }, [subscribe]);

  return (
    <div className="notification-list">
      {notifications.map((notification, index) => (
        <div key={index} className={`notification ${notification.type}`}>
          {notification.message}
        </div>
      ))}
    </div>
  );
};

// App component
const App = () => {
  return (
    <PubSubProvider>
      <NotificationButton />
      <NotificationList />
    </PubSubProvider>
  );
};

5.5 Pub-Sub hiện đại bằng native API

Trước khi cài một lib pub-sub, hãy biết JS đã có 4 API native giải quyết phần lớn use case:

5.5.1 `EventTarget` — chuẩn DOM, chạy cả Node 15+

EventTarget giờ là global trong Node — có thể tạo event bus không cần dependency:

class TypedBus<Events extends Record<string, unknown>> extends EventTarget {
  emit<K extends keyof Events & string>(type: K, detail: Events[K]) {
    this.dispatchEvent(new CustomEvent(type, { detail }));
  }
  on<K extends keyof Events & string>(
    type: K,
    handler: (detail: Events[K]) => void,
    options?: AddEventListenerOptions
  ) {
    const wrapped = (e: Event) => handler((e as CustomEvent<Events[K]>).detail);
    this.addEventListener(type, wrapped, options);
    return () => this.removeEventListener(type, wrapped, options);
  }
}

type AppEvents = {
  "user.login": { userId: string };
  "order.created": { orderId: number; total: number };
};

const bus = new TypedBus<AppEvents>();
const ctrl = new AbortController();

// `AbortSignal` unsubscribe — chuẩn với fetch, addEventListener
bus.on("order.created", (o) => console.log("new order", o.orderId), {
  signal: ctrl.signal,
});

bus.emit("order.created", { orderId: 1, total: 99 });
ctrl.abort();

Ưu điểm: zero dependency, tuân thủ contract đã quen, tự hỗ trợ once, capture, signal.

5.5.2 `BroadcastChannel` — pub-sub giữa các tab/window/worker cùng origin

// Tab A
const chan = new BroadcastChannel("app");
chan.postMessage({ type: "logout" });

// Tab B, Tab C, SharedWorker…
const chan = new BroadcastChannel("app");
chan.addEventListener("message", (e) => {
  if (e.data.type === "logout") redirect("/login");
});

Dùng khi muốn logout đồng bộ, sync cart, cập nhật theme — không cần server round-trip.

5.5.3 `MessageChannel` + `MessagePort` — pub-sub point-to-point qua iframe/worker

const { port1, port2 } = new MessageChannel();
// Gửi port2 qua postMessage cho iframe/worker
iframe.contentWindow.postMessage("hi", "*", [port2]);

port1.onmessage = (e) => console.log("from iframe:", e.data);
port1.postMessage({ type: "init", config: { theme: "dark" } });

5.5.4 `WeakRef` cho bus sống lâu

Nếu event bus là singleton chạy xuyên vòng đời app, subscriber là component ngắn → lưu subscriber bằng WeakRef để không giữ mất GC:

class WeakPubSub<T> {
  private topics = new Map<string, Set<WeakRef<(p: T) => void>>>();
  private reg = new FinalizationRegistry<{ topic: string; ref: WeakRef<any> }>(
    ({ topic, ref }) => this.topics.get(topic)?.delete(ref)
  );

  subscribe(topic: string, fn: (p: T) => void) {
    const ref = new WeakRef(fn);
    if (!this.topics.has(topic)) this.topics.set(topic, new Set());
    this.topics.get(topic)!.add(ref);
    this.reg.register(fn, { topic, ref });
  }

  publish(topic: string, payload: T) {
    const set = this.topics.get(topic);
    if (!set) return;
    for (const ref of set) {
      const fn = ref.deref();
      if (fn) fn(payload);
      else set.delete(ref);
    }
  }
}

Như đã nói ở bài Observer: WeakRef là optimization. Với mọi logic nghiệp vụ, hãy expose unsubscribe tường minh.

5.5.5 Bảng quyết định

Tình huống	Dùng
Pub-sub trong 1 tab, đơn giản	`EventTarget` + `CustomEvent`
Đồng bộ giữa các tab/window	`BroadcastChannel`
Giao tiếp với iframe / Worker	`MessageChannel` / `postMessage`
Worker pool cùng page	`SharedWorker` + `BroadcastChannel`
Cần pattern operator (debounce, map, merge)	RxJS
Khối lượng message lớn + retry + ack	Server broker (NATS, Kafka, Redis Streams) qua WebSocket/SSE

6. Ưu điểm và nhược điểm

Ưu điểm:

Loose coupling: Publishers và subscribers không biết về sự tồn tại của nhau
Khả năng mở rộng: Dễ dàng thêm publishers và subscribers mới mà không ảnh hưởng đến hệ thống
Linh hoạt: Một subscriber có thể đăng ký nhiều kênh, một publisher có thể gửi thông báo đến nhiều kênh
Phân tán: Có thể triển khai trên nhiều hệ thống phân tán

Nhược điểm:

Khó debug: Khó theo dõi luồng dữ liệu khi hệ thống phức tạp
Overhead: Có thể tạo ra overhead khi có nhiều subscribers và publishers
Khó đảm bảo thứ tự: Không đảm bảo thứ tự xử lý các thông báo
Memory leaks: Nếu không hủy đăng ký đúng cách, có thể gây ra memory leaks

7. So sánh với Observer Pattern

Tiêu chí	Observer Pattern	Pub-Sub Pattern
Coupling	Tight coupling (Subject biết về Observers)	Loose coupling (Publishers và Subscribers không biết về nhau)
Trung gian	Không có trung gian	Có trung gian (Message Broker/Event Bus)
Phạm vi	Thường trong một ứng dụng	Có thể mở rộng qua nhiều ứng dụng/hệ thống
Triển khai	Đơn giản hơn	Phức tạp hơn do có trung gian
Use cases	UI events, Object state changes	Distributed systems, Cross-component communication

8. Khi nào nên sử dụng Pub-Sub Pattern?

Khi cần giao tiếp giữa các thành phần không liên quan trực tiếp
Trong hệ thống phân tán với nhiều microservices
Khi cần xử lý sự kiện bất đồng bộ
Khi cần giảm sự phụ thuộc giữa các thành phần
Trong các ứng dụng real-time cần phản ứng với nhiều loại sự kiện

9. Kết luận

Pub-Sub Pattern là một mẫu thiết kế mạnh mẽ cho phép xây dựng các hệ thống có khả năng mở rộng cao với các thành phần tách biệt. Mặc dù có một số nhược điểm như khó debug và overhead, pattern này vẫn là lựa chọn tốt cho các ứng dụng hiện đại cần xử lý nhiều sự kiện và có nhiều thành phần giao tiếp với nhau.

Trong JavaScript và TypeScript, Pub-Sub Pattern có thể được triển khai dễ dàng và được hỗ trợ bởi nhiều thư viện và framework. Hiểu rõ sự khác biệt giữa Pub-Sub và Observer Pattern sẽ giúp bạn chọn đúng mẫu thiết kế cho từng trường hợp sử dụng cụ thể.