java - 在 java 9 Flow 上以一种只有一个订阅者会使用它的方式将数据发布给订阅者

Question

有没有办法以只有一个订阅者可以接收的方式向订阅者发布数据？我想要实现的是，订阅者发布者模型将作为一个队列工作，该队列具有多个读者但只有一个发布者。一旦发布者发布数据，第一个接收数据的订阅者将是唯一一个处理数据的订阅者。

提前致谢!!!

Answer 1

在 react 流中(至少，在它们的 java.util.concurrent.Flow 化身中)，订阅者只需要数据，只有发布者可以控制如何发布该数据。

Java 9 中Flow.Publisher 的唯一通用实现是 SubmissionPublisher，它遵循标准的发布/订阅方式向所有订阅者发布任何已发布的项目.我没有找到任何简单的方法来破解 SubmissionPublisher 以使其只发布给一个订阅者。

但是您可以尝试编写自己的 Flow.Publisher 实现，如下所示:

class QueueLikePublisher<T> implements Publisher<T> {
    private final ExecutorService executor = ForkJoinPool.commonPool(); // daemon-based
    private List<QueueLikeSubscription<? super T>> subscriptions = new CopyOnWriteArrayList<>();

    public synchronized void subscribe(Subscriber<? super T> subscriber) {
        // subscribing: adding a new subscription to the list
        QueueLikeSubscription<? super T> subscription = new QueueLikeSubscription<>(subscriber, executor);
        subscriptions.add(subscription);
        subscriber.onSubscribe(subscription);
    }

    public void submit(T item) {
        // we got some data: looking for non-completed and demanding
        // subscription and give it the data item

        for (QueueLikeSubscription<? super T> subscription : subscriptions) {
            if (!subscription.completed && subscription.demand > 0) {
                subscription.offer(item);
                // we just give it to one subscriber; probaly offer() call needs
                // to be wrapped in a try/catch
                break;
            }
        }
    }

    static class QueueLikeSubscription<T> implements Subscription {
        private final Subscriber<? super T> subscriber;
        private final ExecutorService executor;
        volatile int demand = 0;
        volatile boolean completed = false;

        QueueLikeSubscription(Subscriber<? super T> subscriber,
                ExecutorService executor) {
            this.subscriber = subscriber;
            this.executor = executor;
        }

        public synchronized void request(long n) {
            if (n != 0 && !completed) {
                if (n < 0) {
                    IllegalArgumentException ex = new IllegalArgumentException();
                    executor.execute(() -> subscriber.onError(ex));
                } else {
                    // just extending the demand
                    demand += n;
                }
            }
        }

        public synchronized void cancel() {
            completed = true;
        }

        Future<?> offer(T item) {
            return executor.submit(() -> {
                try {
                    subscriber.onNext(item);
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            });
        }
    }
}

它将尚未完成(例如，已取消)且需求非零的项目发布给第一个订阅者。

请注意，此代码只是用于说明想法的大纲。例如，它可能应该包含更多异常处理(例如处理 RejectedExecutionException)。