Java Flow: ให้ Consumer กำหนดจังหวะด้วย Backpressure

ถ้า producer ส่งข้อมูลเร็วกว่าที่ consumer ทำไหว เรามีสามปลายทาง: memory โต, ข้อมูลตก หรือทั้งระบบช้าลงครับ Backpressure คือให้ปลายทางบอกต้นทางว่าพร้อมรับเท่าไร เหมือนร้านกาแฟรับบัตรคิวตามจำนวนแก้วที่ชงทัน แทนรับออเดอร์ไม่จำกัดแล้วกองเต็มเคาน์เตอร์

Java 9 เพิ่ม Flow เป็นชุด interface ตาม Reactive Streams: Publisher ส่ง, Subscriber รับ, Subscription เชื่อมสองฝ่าย และ Processor เป็นทั้งรับกับส่ง จุดสำคัญอยู่ที่ subscription.request(n) ไม่ใช่ชื่อคลาส

WHY: Demand เป็นส่วนหนึ่งของ Protocol

Subscriber ต้องเรียก request ด้วยจำนวนบวกก่อน Publisher ส่ง onNext ถ้าขอทีละหนึ่ง มันควบคุม memory ง่ายและเหมาะกับงานหนัก แต่ throughput อาจต่ำจากการประสานบ่อย ถ้าขอเป็น batch จะเร็วขึ้นแต่ต้องเผื่อ buffer และ failure ของทั้ง batch

SubmissionPublisher เป็น implementation สำหรับ async publication ไม่ใช่ message broker ถ้าปิด process ข้อมูลที่ยังไม่ durable หายได้ และ submit อาจ block เมื่อ subscriber ช้า จึงอย่านำไปแทน Kafka เพียงเพราะ API คล้ายกัน

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

public class Main {
    static final class OneByOneSubscriber implements Flow.Subscriber<Integer> {
        private final List<Integer> values = new ArrayList<>();
        private final CountDownLatch done = new CountDownLatch(1);
        private Flow.Subscription subscription;

        public void onSubscribe(Flow.Subscription value) {
            subscription = value;
            subscription.request(1);
        }
        public void onNext(Integer value) {
            values.add(value);
            subscription.request(1);
        }
        public void onError(Throwable error) {
            done.countDown();
        }
        public void onComplete() {
            done.countDown();
        }
    }

    public static void main(String[] args) throws Exception {
        OneByOneSubscriber subscriber = new OneByOneSubscriber();
        try (SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>()) {
            publisher.subscribe(subscriber);
            publisher.submit(2);
            publisher.submit(3);
        }

        assert subscriber.done.await(2, TimeUnit.SECONDS) : "subscriber timed out";
        assert subscriber.values.equals(List.of(2, 3)) : subscriber.values;
        System.out.println("flow tests passed");
    }
}

HOW: กำหนด Error และ Cancellation ก่อนเชื่อมของจริง

onNext ของ subscriber เดียวกันถูกเรียงตาม happens-before แต่ไม่ได้แปลว่า thread เดิมเสมอ state ภายในจึงต้องออกแบบให้รองรับ executor ที่เลือก อย่า block นานใน callback; ส่งต่อไป worker ที่มี bounded queue หรือเลือก executor ให้ตรง workload

เมื่อ onNext ล้ม ห้ามปล่อย exception หลุดแบบไม่รู้สถานะ ให้ cancel subscription และส่ง error เข้าระบบสังเกตการณ์ onError กับ onComplete เป็น terminal signal หลังได้รับแล้วห้ามคาดว่าจะมี item ต่อ และ subscriber ต้องรับมือ onSubscribe เพียงครั้งเดียวตามสัญญา

วัด outstanding demand, lag, dropped item และเวลาต่อ onNext ถ้าใช้ offer พร้อม drop handler ต้องประกาศ semantics ว่าข้อมูลใดทิ้งได้ งานการเงินหรือ audit มักต้องใช้ durable broker และ idempotent consumer แทน in-process publisher

Recap

• request(n) คือสัญญาควบคุม demand
• batch ใหญ่เพิ่ม throughput แต่ใช้ buffer มากขึ้น
• SubmissionPublisher ไม่ใช่ durable broker
• callback ต้องไม่ block โดยไม่มีขอบเขต
• ออกแบบ cancel, terminal signal และ metric ก่อน production

เริ่มจาก request(1) ให้ถูกต้องก่อน แล้วเพิ่ม batch จากผลวัดแทนการเดาจังหวะครับ