Reactor Core中自定义Subscriber导致Flux.collectList阻塞问题分析

问题现象描述

在使用Reactor Core框架时，开发者尝试实现自定义的org.reactivestreams.Processor接口，并在独立线程中生产数据。当调用Flux.collectList().block()方法时，程序会出现阻塞现象，无法正常返回结果。而使用Flux.subscribe()或Flux.toIterable()方法则能正常工作。

问题重现示例

开发者提供了两种重现方式：

1. 第一种方式通过自定义Processor实现，在独立线程中推送数据，然后尝试阻塞式获取结果列表，最终抛出IllegalStateException超时异常。

2. 第二种方式展示了更简洁的代码示例：

class AdapterProcessor implements Processor<String, String> {
    // 实现Processor接口方法
    // ...
}

class CustomRunnable implements Runnable {
    private final Subscriber<String> subscriber;
    // 在run方法中直接调用subscriber的onNext和onComplete
    // ...
}

@Test
void reproCase() {
    var function = new AdapterProcessor();
    var runnable = new CustomRunnable(function);
    var thread = new Thread(runnable);

    Flux.from(function)
            .map(Integer::parseInt)
            .map(Integer::longValue)
            .doFirst(thread::start)
            .collectList()
            .block(Duration.of(10, ChronoUnit.SECONDS));
}

问题根本原因

经过分析，这个问题源于对Reactive Streams规范的违反，具体表现在：

1. 违反订阅顺序：示例代码在订阅过程完成前就开始推送数据，违反了Reactive Streams规范中的规则1.9。

2. 忽略背压机制：自定义实现没有正确处理请求信号(request signal)，直接推送数据而不考虑下游的处理能力。

3. 线程安全问题：在多线程环境下没有正确处理订阅和事件发布的同步问题。

解决方案

开发者发现使用Reactor Core提供的FluxSink可以解决这个问题。正确的做法应该是：

1. 使用Reactor提供的抽象：将实现改为同时实现Subscriber和Consumer<FluxSink>接口。

2. 使用Flux.create工厂方法：通过Reactor提供的创建方式构建Flux，而不是直接实现底层接口。

3. 遵循响应式编程规范：确保在订阅完成后再开始数据推送，并正确处理背压信号。

最佳实践建议

1. 优先使用Reactor提供的高级API：如Flux.create、Flux.generate等，而不是直接实现底层接口。

2. 理解响应式流规范：在自定义实现前，确保充分理解Reactive Streams规范的所有规则。

3. 正确处理多线程场景：在涉及多线程的场景下，特别注意订阅和数据推送的顺序和同步问题。

4. 利用现有工具：Reactor Core提供了丰富的操作符和工具类，大多数情况下不需要自定义底层实现。

总结

这个问题展示了在响应式编程中直接实现底层接口可能带来的复杂性。Reactor Core框架提供了更高层次的抽象来简化开发，开发者应该优先使用这些抽象，而不是直接操作底层接口。理解并遵循Reactive Streams规范对于构建正确的响应式应用至关重要。