Backpressure: Điều tiết luồng trong hệ thống bất đồng bộ

Một API đọc 40.000 event mỗi giây nhưng dependency chỉ chịu được 6.000. Ban đầu hệ thống trông nhanh vì producer không đợi. Sau đó memory tăng, garbage collection chiếm CPU, timeout kích hoạt retry, còn message cũ nằm trước công việc mới. Async không xóa capacity limit; nó chỉ giấu giới hạn trong backlog.

Backpressure là cơ chế để consumer buộc upstream giảm tốc, tạm dừng hoặc chọn overload policy tường minh. Thiết kế tốt giới hạn cả việc đang chạy lẫn đang chờ, truyền demand cho producer và định nghĩa chuyện gì xảy ra khi không thể giảm đầu vào.

Áp lực bắt đầu từ chênh lệch tốc độ #

Xét một stage có tốc độ đến lambda item mỗi giây, tốc độ phục vụ mu và số item trung bình trong stage là L. Nếu lambda luôn lớn hơn mu, không implementation queue nào làm hệ thống ổn định được. Trong khoảng quá tải t giây, backlog tăng xấp xỉ:

Với hệ thống ổn định ở trạng thái đều, định luật Little liên hệ concurrency, throughput và thời gian trong hệ thống:

• L = số item trung bình đang phục vụ cộng với đang chờ
• lambda = throughput hoàn tất mỗi giây
• W = số giây trung bình một item ở trong stage

Nếu dependency hoàn tất 200 request mỗi giây và mỗi request ở đó 0,25 giây, concurrency kỳ vọng là $L = 200 \times 0.25 = 50$. Đo được concurrency 500 với cùng throughput không có nghĩa là thêm capacity. Khoảng 450 request đang chờ, sắp timeout hoặc tranh chấp tài nguyên.

Mức sử dụng cho thêm một tín hiệu cảnh báo:

Khi rho gần 1, biến thiên nhỏ trong service time cũng làm queueing delay tăng mạnh. 100% utilization không chừa chỗ cho burst, request chậm hay retry. Capacity planning cần latency budget và headroom, không chỉ throughput trung bình.

Push, pull và tín hiệu nhu cầu #

Trong giao diện push, producer quyết định lúc emit. DOM event, EventEmitter, webhook và nhiều callback hoạt động như vậy. Nếu consumer làm việc async bên trong listener, emitter thường không await. Hàng nghìn promise có thể active trước khi promise đầu tiên hoàn tất.

Trong giao diện pull, consumer yêu cầu giá trị tiếp theo. Async iterator chỉ tiến khi code gọi và await next(). Nhờ đó nhu cầu trở thành một phần của protocol:

async function* readPages(
  fetchPage: (cursor?: string) => Promise<{
    items: string[];
    nextCursor?: string;
  }>,
): AsyncGenerator<string> {
  let cursor: string | undefined;

  do {
    const page = await fetchPage(cursor);
    for (const item of page.items) yield item;
    cursor = page.nextCursor;
  } while (cursor);
}

for await (const item of readPages(fetchPage)) {
  await persist(item); // Chỉ yêu cầu item kế tiếp sau khi hoàn tất.
}

Pull chưa chắc bounded: page có thể rất lớn, hàm có thể prefetch hoặc consumer khởi động việc mà không await. Mapper dưới đây giữ tối đa concurrency operation active và dừng pull khi cửa sổ đầy:

type Tracked<Result> = Promise<{
  tracked: Tracked<Result>;
  result: Result;
}>;

export async function* mapConcurrent<Input, Result>(
  source: AsyncIterable<Input>,
  concurrency: number,
  map: (input: Input) => Promise<Result>,
): AsyncGenerator<Result> {
  if (!Number.isInteger(concurrency) || concurrency < 1) {
    throw new RangeError('concurrency must be a positive integer');
  }

  const active = new Set<Tracked<Result>>();

  for await (const input of source) {
    let tracked!: Tracked<Result>;
    tracked = map(input).then((result) => ({ tracked, result }));
    active.add(tracked);

    if (active.size >= concurrency) {
      const completed = await Promise.race(active);
      active.delete(completed.tracked);
      yield completed.result;
    }
  }

  while (active.size > 0) {
    const completed = await Promise.race(active);
    active.delete(completed.tracked);
    yield completed.result;
  }
}

Mapper emit theo thứ tự hoàn tất. Giữ thứ tự input có thể gây head-of-line blocking. Bản production cũng nên nhận AbortSignal, settle task còn lại sau lỗi và quy định lỗi có dừng toàn stream hay không.

Demand có thể là iterator pull, Reactive Streams credit, HTTP/2 hay QUIC window, TCP receive window, stream return value hoặc broker prefetch. Contract chung: downstream cấp quyền gửi hữu hạn.

Queue hữu hạn, watermark và Node streams #

Bounded queue biến capacity thành invariant. Khi đầy, push phải đợi, reject hoặc áp dụng loss policy. Queue sau block producer đến khi consumer tạo chỗ; mỗi caller giữ đúng một item trong lúc đợi admission.

export class BoundedAsyncQueue<Item> implements AsyncIterable<Item> {
  readonly #items: Item[] = [];
  readonly #readers: Array<(result: IteratorResult<Item>) => void> = [];
  readonly #spaceWaiters: Array<() => void> = [];
  #closed = false;

  constructor(readonly capacity: number) {
    if (!Number.isInteger(capacity) || capacity < 1) {
      throw new RangeError('capacity must be a positive integer');
    }
  }

  async push(item: Item, signal?: AbortSignal): Promise<void> {
    while (this.#items.length >= this.capacity && !this.#closed) {
      await this.#waitForSpace(signal);
    }
    if (this.#closed) throw new Error('queue is closed');

    const reader = this.#readers.shift();
    if (reader) reader({ value: item, done: false });
    else this.#items.push(item);
  }

  close(): void {
    if (this.#closed) return;
    this.#closed = true;
    for (const reader of this.#readers.splice(0)) {
      reader({ value: undefined, done: true });
    }
    for (const wake of this.#spaceWaiters.splice(0)) wake();
  }

  async next(): Promise<IteratorResult<Item>> {
    if (this.#items.length > 0) {
      const value = this.#items.shift() as Item;
      this.#spaceWaiters.shift()?.();
      return { value, done: false };
    }
    if (this.#closed) return { value: undefined, done: true };
    return new Promise((resolve) => this.#readers.push(resolve));
  }

  [Symbol.asyncIterator](): AsyncIterator<Item> {
    return { next: () => this.next() };
  }

  #waitForSpace(signal?: AbortSignal): Promise<void> {
    if (signal?.aborted) return Promise.reject(signal.reason);

    return new Promise((resolve, reject) => {
      const cleanup = () => signal?.removeEventListener('abort', onAbort);
      const wake = () => {
        cleanup();
        resolve();
      };
      const onAbort = () => {
        const index = this.#spaceWaiters.indexOf(wake);
        if (index >= 0) this.#spaceWaiters.splice(index, 1);
        cleanup();
        reject(signal?.reason ?? new Error('aborted'));
      };

      this.#spaceWaiters.push(wake);
      signal?.addEventListener('abort', onAbort, { once: true });
    });
  }
}

Một ngưỡng duy nhất có thể làm producer dao động giữa chạy và dừng. Watermark thêm hysteresis: pause ở high watermark, resume dưới low watermark. Queue 1.000 item có thể pause ở 800 và resume ở 300. Capacity 1.000 vẫn chứa race; high watermark là trigger, không phải hard limit.

Node streams đã triển khai quá trình thương lượng này. Writable.write() trả về false khi buffer nội bộ chạm highWaterMark; producer thủ công phải đợi event drain. pipeline() nối đường pause, error, cleanup và completion cho chuỗi stream:

import { createWriteStream } from 'node:fs';
import { Readable, Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';

async function* records(): AsyncGenerator<{ id: string; score: number }> {
  for await (const row of readDatabaseCursor()) yield row;
}

const toJsonLines = new Transform({
  objectMode: true,
  writableHighWaterMark: 32,
  transform(record: { id: string; score: number }, _encoding, callback) {
    callback(null, `${JSON.stringify(record)}\n`);
  },
});

await pipeline(
  Readable.from(records(), { objectMode: true, highWaterMark: 32 }),
  toJsonLines,
  createWriteStream('scores.ndjson', { highWaterMark: 64 * 1024 }),
);

highWaterMark không phải trần memory toàn cục. Mỗi stage có buffer; object mode đếm object thay vì byte; transform và kernel còn giữ dữ liệu. Hãy giới hạn payload lớn theo byte.

Batching và chính sách quá tải #

Batching chia đều fixed cost của round trip, commit và acknowledgement. Hãy flush khi chạm maxItems hoặc maxWaitMs: chỉ đợi batch đầy gây latency vô hạn khi ít tải; chỉ dùng timer làm burst kém hiệu quả. Với payload biến thiên, byte size quan trọng hơn item count.

Backpressure hoạt động khi producer có thể chậm lại. Public traffic, UDP telemetry, webhook sender và record đã publish lên broker có thể không nghe lệnh pause cục bộ. Hệ thống nhận lúc đó cần overload policy:

Nên load shed trước parsing đắt đỏ hoặc dependency call. Admission control xét capacity, concurrency, deadline, priority và tenant budget. Request còn 20 ms không nên vào queue đợi 500 ms. Per-tenant limit ngăn một nguồn chiếm mọi slot.

Retry cũng là producer. Nếu mỗi lần lỗi được retry tối đa r lần và từng attempt có xác suất lỗi độc lập p, hệ số attempt kỳ vọng là:

• A = số attempt kỳ vọng trên mỗi operation gốc
• p = xác suất một attempt thất bại
• r = số lần retry tối đa

Trong outage, p gần 1 nên A gần r + 1. Ba retry có thể gần nhân bốn áp lực. Hãy giới hạn attempt, dùng backoff có jitter, tôn trọng Retry-After, áp deadline tổng và retry budget. Probe half-open của circuit breaker cũng cần demand hữu hạn.

Hệ thống queue chỉ chuyển áp lực, không xóa nó #

Broker bền tách availability của producer khỏi availability của consumer. Nó có thể hấp thụ burst hữu hạn và replay sau deployment. Nó không tạo thêm consumer capacity. Nếu ingress luôn lớn hơn processing, retention sẽ tăng đến khi chạm giới hạn byte, tuổi dữ liệu, partition hoặc chi phí.

Demand control riêng của broker rất quan trọng. Prefetch kiểu RabbitMQ giới hạn delivery chưa ack trên mỗi consumer. Prefetch bằng 1 bảo vệ worker chậm nhưng có thể dùng concurrency không hết; prefetch cực lớn chuyển backlog từ broker vào memory worker và có thể phân phối bất công. Đặt nó gần mức concurrency hữu ích của worker rồi đo.

Kafka consumer pull record, nhưng max.poll.records không phải giới hạn processing hoàn chỉnh. Nếu code poll tiếp trước khi xong batch trước hoặc tạo một promise cho từng record, lượng việc trong memory vẫn tăng. Pause assigned partition khi local capacity đầy, tiếp tục poll đủ thường xuyên để giữ group membership khi protocol yêu cầu, resume dưới low watermark và chỉ commit offset theo delivery semantics mong muốn. Ordering theo partition có nghĩa concurrency thường phải nhận biết partition.

Cloud queue có visibility timeout. Xử lý quá lâu làm message xuất hiện lại. Hãy gia hạn visibility cho task dài, dùng handler idempotent, giới hạn receive attempt và chuyển poison message sang dead-letter queue. Theo dõi age/count của dead letter và redrive có kiểm soát để không dội tải vào service vừa hồi phục.

Pressure end-to-end phải đi qua mọi boundary. Giới hạn consumer nhưng cho HTTP ingress enqueue vô hạn chỉ dời điểm lỗi. Hãy công bố backlog tối đa theo item, byte hoặc tuổi và để producer nhìn thấy hậu quả khi chạm giới hạn.

Đo lường và kiểm thử đường quá tải #

Chỉ throughput sẽ che mất pressure. Quan sát từng stage và tách công việc đang đợi khỏi đang active:

• queue depth theo item và byte, cùng tuổi của item cũ nhất
• tốc độ arrival, admission, completion, rejection, drop, retry và dead-letter
• active concurrency và concurrency limit đã cấu hình
• service time tách khỏi queue wait và end-to-end latency
• lần chuyển high/low watermark và thời gian ở trạng thái pause
• timeout budget còn lại khi công việc bắt đầu
• event-loop delay, heap usage, garbage-collection pause, socket count và broker lag

Depth là trạng thái; đạo hàm cho biết hướng. Depth 10.000 đang giảm có thể khỏe hơn depth 100 tăng liên tục. Tuổi item cũ nhất thường phản ánh tác hại tốt hơn count. Alert theo tăng trưởng và tuổi so với SLO, không theo một ngưỡng chung.

Kiểm thử bằng dependency mà test có thể khóa completion. Lấp queue đến high watermark và assert upstream ngừng pull; giải phóng một operation và assert active work không vượt giới hạn. Kiểm tra resume, cancellation, shutdown và lỗi không để producer đợi mãi.

Sau đó chạy load test với arrival dạng burst, service time biến thiên và long-tail, payload quá cỡ, dependency brownout, retryable error, broker redelivery và consumer restart. Theo dõi phương trình bảo toàn trong một khoảng test:

Mọi nhãn đều dùng ASCII và mỗi term phải được đếm đúng một lần. Nếu phương trình không cân bằng, công việc có thể bị duplicate, leak hoặc mất im lặng. Cũng cần kiểm tra recovery: sau khi dependency khỏe lại, backlog age phải hội tụ về 0 mà không có retry surge gây sụp đổ lần nữa.

Những điểm cần nhớ #

• Backpressure là demand protocol end-to-end, không phải queue lớn hơn. Nó làm capacity hữu hạn của downstream hiện rõ với upstream.
• Service ổn định đòi hỏi arrival rate dài hạn thấp hơn service rate. Định luật Little nối throughput, latency và in-flight work; utilization cao giải thích queueing delay tăng gắt.
• Pull interface, bounded concurrency, bounded queue và high/low watermark bổ sung cho nhau. Không cơ chế nào đủ nếu vẫn còn hidden buffer không giới hạn.
• Node streams truyền pressure qua write(), drain và pipeline(), nhưng từng stage và payload size vẫn góp vào memory bound thực tế.
• Batching tăng hiệu quả; admission control, shedding, deadline và fairness định nghĩa hành vi khi producer không thể chậm lại.
• Retry khuếch đại pressure trong lúc lỗi. Giới hạn nó bằng jittered backoff, deadline tổng, retry budget và concurrency limit.
• Broker di chuyển và lưu bền backlog. Prefetch, polling, visibility timeout, offset rule, idempotency và vận hành dead-letter quyết định backlog có được kiểm soát hay không.
• Đo queue age, wait time, rate, retry và saturation; sau đó kiểm thử overload lẫn recovery như hành vi hạng nhất.