Demand và backpressure

Backpressure là việc bên tiêu thụ làm gì khi bên sản xuất nhanh hơn nó. Phần lớn framework reactive trả lời “cứ dồn bộ đệm rồi cầu may”; Combine xây câu trả lời thẳng vào protocol, và quyết định ấy giải thích vài góc kỳ quặc của nó.

struct Demand: Equatable {
    static let unlimited: Demand
    static let none: Demand
    static func max(_ value: Int) -> Demand
}

Demand cộng dồn và chỉ có thể tăng. Một subscriber đã xin .max(3) rồi xin thêm .max(2) thì có thể nhận năm giá trị. Không có cách nào giảm phần demand đang treo — bạn không rút lời xin lại được. Cách duy nhất để dừng là hủy.

Vì sao bạn chưa bao giờ phải nghĩ đến nó

Vì sink và assign xin .unlimited ngay khi đăng ký, và .unlimited vô hiệu hóa hoàn toàn cơ chế này. Mọi giá trị publisher tạo ra được đều được gửi đi ngay khi nó tồn tại.

Với một phản hồi mạng hay một cú chạm nút, như vậy là đúng và miễn phí. Nó thôi miễn phí khi bên sản xuất thật sự chạy nhanh hơn bên tiêu thụ:

• Một lượt đọc file phát ra các khối nhanh hơn tốc độ phân tích chúng
• Một cảm biến hay một socket giao dữ liệu ở tần suất cao cố định
• Timer.publish đổ vào một sink làm việc thật ở mỗi nhịp
• Một PassthroughSubject được nạp trong một vòng lặp chặt

Ở mỗi trường hợp, một sink không giới hạn sẽ dồn bộ đệm, và mặc định bộ đệm của Combine không bị chặn trên. Bộ nhớ phình ra cho tới khi thứ gì đó bị hệ thống đá đi.

buffer làm chính sách trở nên tường minh

fastPublisher
    .buffer(size: 100, prefetch: .keepFull, whenFull: .dropOldest)
    .sink { process($0) }
    .store(in: &cancellables)

Ba quyết định, và trước đây bạn vẫn đang ra cả ba một cách ngầm định:

• size — bao nhiêu giá trị được phép chờ.
• prefetch — .keepFull xin trước để luôn đầy; .byRequest chỉ xin khi phía dưới xin.
• whenFull — .dropOldest, .dropNewest, hoặc .customError.

Giá trị của toán tử này ít nằm ở việc dồn bộ đệm mà nhiều hơn ở chỗ nó buộc bạn trả lời câu “nên xảy ra chuyện gì khi ta bị tụt lại?” — câu hỏi mà mọi pipeline thời gian thực đều có, và phần lớn trả lời một cách tình cờ.

Những câu trả lời rẻ hơn, xét trước

Trước khi với tới buffer, hãy cân nhắc xem các giá trị đó có cần đến nơi hay không:

Toán tử	Hành vi
throttle(for:scheduler:latest:)	Tối đa một giá trị mỗi khoảng, cái đầu hoặc cái cuối
debounce(for:scheduler:)	Chỉ sau một quãng lặng — toán tử của ô tìm kiếm
removeDuplicates()	Bỏ các giá trị bằng nhau liên tiếp
collect(.byTime(_:_:))	Gom thành mảng theo từng khoảng
switchToLatest	Bỏ cái đang bay khi có cái mới hơn đến

Với giao diện, những thứ này gần như luôn là câu trả lời đúng. Một vị trí cuộn phát ra ở 120 Hz đổ vào một view vẽ lại ở 60 thì cần throttle, không cần bộ đệm — bỏ đi các giá trị trung gian không phải là mất dữ liệu, vì một vị trí cuộn đã cũ thì chẳng có giá trị gì.

Chỗ trừu tượng bị rò

Subject lờ đi demand. send(_:) trên một PassthroughSubject giao giá trị ngay lập tức bất kể phía dưới đã xin bao nhiêu. Subject là cánh cửa mệnh lệnh, và code mệnh lệnh thì không thương lượng — đó là thêm một lý do nữa để coi một subject nằm giữa pipeline là một mùi lạ. Nó lặng lẽ gỡ bỏ backpressure khỏi mọi thứ nằm dưới nó.

receive(on:) sinh ra một hàng đợi không chặn trên. Các giá trị vượt qua ranh giới scheduler sẽ được xếp vào hàng đợi của scheduler đó, và hàng đợi ấy không phải là cái buffer nào bạn đã cấu hình ở phía trên. Một pipeline được backpressure đúng đắn cho tới receive(on: DispatchQueue.main) vẫn có thể làm ngập hàng đợi chính.

flatMap có tham số maxPublishers, và mặc định của nó là .unlimited. Đó là giá trị mặc định đắt đỏ nhất trong cả framework:

urls.publisher
    .flatMap { url in fetch(url) }              // mọi request cùng một lúc
    .sink { … }

urls.publisher
    .flatMap(maxPublishers: .max(4)) { url in fetch(url) }   // bốn cái một lúc
    .sink { … }

Một nghìn URL qua phiên bản đầu sẽ khởi động một nghìn request đồng thời. Phiên bản sau là một giới hạn đồng thời viết gọn trong một tham số, và đó là lý do đáng biết chương này tồn tại kể cả khi bạn không bao giờ tự viết một subscriber.