归档

• GitHub: JDK-反应流（响应式流）

使用示例

• https://github.com/zengxf/small-frame-demo/blob/master/multi-thread/reactive-test/reactor-demo/src/main/java/cn/zxf/reactor_demo/jdk/PubSubTest.java

JDK 版本

openjdk version "17" 2021-09-14
OpenJDK Runtime Environment (build 17+35-2724)
OpenJDK 64-Bit Server VM (build 17+35-2724, mixed mode, sharing)

原理

关键类

• java.util.concurrent.SubmissionPublisher

// 提交式-发布者
public class SubmissionPublisher<T> implements Publisher<T>, AutoCloseable {BufferedSubscription<T> clients;    // 客户端（BufferedSubscriptions）链表final ReentrantLock lock;           // 锁定以排除多个源volatile boolean closed;            // 运行状态，仅在锁内更新boolean subscribed; // 在第一次调用订阅时设置 true，以初始化可能的拥有者Thread owner;       // 第一个要订阅的调用者线程，如果线程发生更改则为 nullvolatile Throwable closedException; // closeExceptionally 中的异常final Executor executor;            // 用于构造 BufferedSubscriptions 的参数final BiConsumer<? super Subscriber<? super T>, ? super Throwable> onNextHandler;final int maxBufferCapacity;// 调用入口 ref: sign_demo_001public SubmissionPublisher() {this(ASYNC_POOL, Flow.defaultBufferSize(), null);   // ref: sign_cm_002}// sign_cm_002public SubmissionPublisher(Executor executor, int maxBufferCapacity,BiConsumer<? super Subscriber<? super T>, ? super Throwable> handler) {... // 参数校验this.lock = new ReentrantLock();this.executor = executor;   // def: ForkJoinPoolthis.onNextHandler = handler;this.maxBufferCapacity = roundCapacity(maxBufferCapacity);  // def: 256}
}

订阅

• java.util.concurrent.SubmissionPublisher

    // 订阅（添加订阅者）。调用入口 ref: sign_demo_010public void subscribe(Subscriber<? super T> subscriber) {if (subscriber == null) throw new NullPointerException();ReentrantLock lock = this.lock;int max = maxBufferCapacity;// INITIAL_CAPACITY = 32, 默认情况下计算完数组长度为 32Object[] array = new Object[max < INITIAL_CAPACITY ? max : INITIAL_CAPACITY]; // 初始化订阅关系，ref: sign_c_110 | sign_cm_110BufferedSubscription<T> subscription = new BufferedSubscription<T>(subscriber, executor, onNextHandler, array, max);lock.lock();try {if (!subscribed) {subscribed = true;owner = Thread.currentThread();}for (BufferedSubscription<T> b = clients, pred = null;;) {if (b == null) {    // 还没有初始化链Throwable ex;subscription.onSubscribe();if ((ex = closedException) != null)subscription.onError(ex);   // 有异常else if (closed)subscription.onComplete();  // 已关闭else if (pred == null)clients = subscription;     // 初始化链elsepred.next = subscription;   // 加入链break;}BufferedSubscription<T> next = b.next;if (b.isClosed()) {   // removeb.next = null;    // detachif (pred == null)clients = next;elsepred.next = next;}else if (subscriber.equals(b.subscriber)) {// 不能重复添加b.onError(new IllegalStateException("Duplicate subscribe"));break;}elsepred = b;   // 方便后面的加入链b = next;       // 方便遍历链}} finally {lock.unlock();}}

• java.util.concurrent.SubmissionPublisher.BufferedSubscription

    // sign_c_110 订阅关系static final class BufferedSubscription<T> implements Subscription, ForkJoinPool.ManagedBlocker {long timeout;                      // Long.MAX_VALUE if untimed waitint head;                          // next position to takeint tail;                          // next position to putfinal int maxCapacity;             // max buffer sizevolatile int ctl;                  // atomic run state flagsObject[] array;                    // bufferfinal Subscriber<? super T> subscriber;final BiConsumer<? super Subscriber<? super T>, ? super Throwable> onNextHandler;Executor executor;                 // null on errorBufferedSubscription<T> next;      // 组装链 // sign_cm_110BufferedSubscription(Subscriber<? super T> subscriber,   // 自定义的订阅者Executor executor,      // ForkJoinPoolBiConsumer<? super Subscriber<? super T>, ? super Throwable> onNextHandler, // nullObject[] array,         // len: 32int maxBufferCapacity   // 256) {this.subscriber = subscriber;this.executor = executor;this.onNextHandler = onNextHandler;this.array = array;this.maxCapacity = maxBufferCapacity;}}

提交数据

• java.util.concurrent.SubmissionPublisher

    // 提交数据。调用入口 ref: sign_demo_020public int submit(T item) {return doOffer(item, Long.MAX_VALUE, null); // ref: sign_m_210}// sign_m_210private int doOffer(T item, long nanos, BiPredicate<Subscriber<? super T>, ? super T> onDrop) {if (item == null) throw new NullPointerException();int lag = 0;boolean complete, unowned;ReentrantLock lock = this.lock;lock.lock();try {Thread t = Thread.currentThread(), o;BufferedSubscription<T> b = clients;if ((unowned = ((o = owner) != t)) && o != null)owner = null;                     // disable biasif (b == null)complete = closed;else {  // 有订阅者才做处理complete = false;boolean cleanMe = false;BufferedSubscription<T> retries = null, rtail = null, next;do {next = b.next;int stat = b.offer(item, unowned);  // 依次发给订阅者句柄，ref: sign_m_220...} while ((b = next) != null);   // 遍历链...}} finally {lock.unlock();}...}

• java.util.concurrent.SubmissionPublisher.BufferedSubscription

        // sign_m_220 添加到队列final int offer(T item, boolean unowned) {Object[] a;int stat = 0, cap = ((a = array) == null) ? 0 : a.length;int t = tail, i = t & (cap - 1), n = t + 1 - head;if (cap > 0) {boolean added;if (n >= cap && cap < maxCapacity)  // resize (扩容)added = growAndOffer(item, a, t);else if (n >= cap || unowned)       // need volatile CAS (CAS 替换)added = QA.compareAndSet(a, i, null, item);else {                              // can use release mode (设置值)QA.setRelease(a, i, item); added = true;}if (added) {        // 添加成功tail = t + 1;   // 改下标（可循环使用数组）stat = n;}}return startOnOffer(stat);  // 尝试启动，ref: sign_m_221}/*** sign_m_221 尝试在添加后启动消费者任务*/final int startOnOffer(int stat) {int c; // start or keep alive if requests exist and not activeif (((c = ctl) & (REQS | ACTIVE)) == REQS &&((c = getAndBitwiseOrCtl(RUN | ACTIVE)) & (RUN | CLOSED)) == 0)tryStart(); // 尝试启动，ref: sign_m_222...return stat;}// sign_m_222 尝试启动消费者任务final void tryStart() {try {Executor e;ConsumerTask<T> task = new ConsumerTask<T>(this);   // ref: sign_c_230if ((e = executor) != null)     // skip if disabled on errore.execute(task);            // 执行体，ref: sign_c_230} ... // catch}// sign_m_225 消费final void consume() {Subscriber<? super T> s;if ((s = subscriber) != null) {          // hoist checkssubscribeOnOpen(s); // 没开启，则开启并回调 Subscriber #onSubscribe() 方法，ref: sign_demo_110long d = demand;for (int h = head, t = tail;;) {int c, taken; boolean empty;if (((c = ctl) & ERROR) != 0) {closeOnError(s, null);      // 有异常，回调 Subscriber #onError() 方法，ref: sign_demo_130break;}else if ((taken = takeItems(s, d, h)) > 0) {    // 获取队列元素并处理，ref: sign_m_226head = h += taken;d = subtractDemand(taken);}...else if (t == (t = tail)) {      // stability checkif ((empty = (t == h)) && (c & COMPLETE) != 0) {closeOnComplete(s);      // 已完成，回调 Subscriber #onComplete() 方法，ref: sign_demo_140break;}...}}}}// sign_m_226 获取队列元素并处理final int takeItems(Subscriber<? super T> s, long d, int h) {Object[] a;int k = 0, cap;if ((a = array) != null && (cap = a.length) > 0) {int m = cap - 1, b = (m >>> 3) + 1;int n = (d < (long)b) ? (int)d : b;for (; k < n; ++h, ++k) {Object x = QA.getAndSet(a, h & m, null); // 获取元素...else if (!consumeNext(s, x)) // 通知订阅者，回调 Subscriber #onNext() 方法，ref: sign_demo_120break;}}return k;}

• java.util.concurrent.SubmissionPublisher.ConsumerTask

    // sign_c_230 消费任务（通知订阅者）static final class ConsumerTask<T> extends ForkJoinTask<Void>implements Runnable, CompletableFuture.AsynchronousCompletionTask {final BufferedSubscription<T> consumer;ConsumerTask(BufferedSubscription<T> consumer) {this.consumer = consumer;}... // 其他方法// sign_c_230 执行体public final void run() { consumer.consume(); } // 消费，ref: sign_m_225}

关闭

• 简单，略

背压

• 看代码或调试时，没发现 publisher 暂停的代码，可用 JConsole 查看线程栈

...
java.base@17/jdk.internal.misc.Unsafe.park(Native Method)
... .locks.LockSupport.park(LockSupport.java:211)
... .SubmissionPublisher$BufferedSubscription.block(SubmissionPublisher.java:1495) // ref: sign_m_321
... .ForkJoinPool.unmanagedBlock(ForkJoinPool.java:3463)
... .ForkJoinPool.managedBlock(ForkJoinPool.java:3434)
... .SubmissionPublisher$BufferedSubscription.awaitSpace(SubmissionPublisher.java:1462) // ref: sign_m_320
...

• java.util.concurrent.SubmissionPublisher

    // 提交数据。调用入口 ref: sign_demo_020private int doOffer(T item, long nanos, BiPredicate<Subscriber<? super T>, ? super T> onDrop) {int lag = 0;...try {Thread t = Thread.currentThread(), o;if ((unowned = ((o = owner) != t)) && o != null)...if (retries != null || cleanMe)lag = retryOffer(item, nanos, onDrop, retries, lag, cleanMe);   // 背压处理，ref: sign_m_310} ... // finally...}// sign_m_310 背压处理private int retryOffer(T item, long nanos,BiPredicate<Subscriber<? super T>, ? super T> onDrop,BufferedSubscription<T> retries, int lag,boolean cleanMe) {for (BufferedSubscription<T> r = retries; r != null;) {BufferedSubscription<T> nextRetry = r.nextRetry;r.nextRetry = null;if (nanos > 0L)r.awaitSpace(nanos);    // 等待，ref: sign_m_320...}...return lag;}

• java.util.concurrent.SubmissionPublisher.BufferedSubscription

    static final class BufferedSubscription<T> implements Subscription, ForkJoinPool.ManagedBlocker {// sign_m_320 帮助或阻止直到超时、关闭或空间可用final void awaitSpace(long nanos) {if (!isReleasable()) {ForkJoinPool.helpAsyncBlocker(executor, this);if (!isReleasable()) {timeout = nanos;try {ForkJoinPool.managedBlock(this);    // 最终会调用 block() 进行阻塞，ref: sign_m_321} ... // catch}}}// sign_m_321 阻塞实现 (实现 ManagedBlocker 方法)@Overridepublic final boolean block() {...while (!isReleasable()) {...else if (waiter == null)waiter = Thread.currentThread();    // 记录当前线程...elseLockSupport.park(this); // 阻塞}...}// 在订阅者获取元素时，进行通知，继 sign_m_226final int takeItems(Subscriber<? super T> s, long d, int h) {...if (waiting != 0)signalWaiter(); // 通知发布者，ref: sign_m_325...}// sign_m_325 通知发布者final void signalWaiter() {Thread w;waiting = 0;if ((w = waiter) != null)LockSupport.unpark(w);  // 唤醒发布者线程}}