Reactor Core项目中替代FluxProcessor实现自定义Subscriber的技术方案

在Reactor Core项目中，开发者经常需要处理自定义Subscriber的实现需求。本文探讨如何在不使用已废弃的FluxProcessor情况下，实现一个既能作为Publisher又能作为Subscriber的组件。

背景与挑战

在响应式编程中，有时需要创建一个既能接收数据又能发布数据的中间组件。传统做法是使用Processor接口，但随着Reactor的发展，Processor已被标记为废弃，推荐使用Sinks API替代。

传统Processor方案

在旧版本中，可以简单地创建一个EmitterProcessor实例：

val processor : Processor[Payload, Payload] = EmitterProcessor.create(1)

这种方案虽然简单直接，但基于已被废弃的API，不推荐在新项目中使用。

Sinks API替代方案

Reactor Core推荐使用Sinks API作为Processor的替代方案。核心思路是利用Sinks.ManyWithUpstream的subscribeTo方法：

Sinks.ManyWithUpstream<Integer> adapter = Sinks.unsafe().many().multicast().onBackpressureBuffer();
Publisher<Integer> upstream = ...;
Disposable disposable = adapter.subscribeTo(upstream);

与Pekko集成的实践

在与Pekko框架集成时，需要特别注意响应式流的生命周期管理。一个可行的解决方案是：

def serverSink : Sink[Payload, NotUsed] = {
    val sink : Sink[Payload, Publisher[Payload]] = Sink.asPublisher(true)
    sink.mapMaterializedValue { pub =>
      pub.subscribe(new Subscriber[Payload] {
        override def onComplete(): Unit = ()
        override def onError(t: Throwable): Unit = ()
        override def onNext(t: Payload): Unit = ()
        override def onSubscribe(s: Subscription): Unit = s.request(Long.MaxValue)
      })
      RSocketServer.create(
        SocketAcceptor.forRequestStream(payload =>
          Flux.from(pub)
      ).bindNow(TcpServerTransport.create("localhost", 3141))
      NotUsed.notUsed()
    }
}

关键注意事项

1. 必须添加一个虚拟Subscriber来激活Publisher，否则数据流不会启动
2. 使用Sink.asPublisher(true)启用fanout模式，支持多个订阅者
3. 注意资源清理，确保在不再需要时释放所有资源

结论

虽然Reactor Core不再直接支持将Sink作为Subscriber使用，但通过合理的架构设计和API组合，仍然可以实现相同的功能。在与Pekko等框架集成时，需要特别注意响应式流的激活机制和生命周期管理。