如何创建RSocket forRequestStream并返回Pekko/Akka Sink（弃用FluxProcessor）

核心思路 #

用Reactor官方推荐的Sink.Many替代废弃的EmitterProcessor，再通过标准Reactive Streams接口将Reactor Sink转换为Pekko Sink，既满足RSocket对Flux的要求，也适配现有代码对Pekko Sink的依赖。

可行代码实现 #

import reactor.core.publisher.{Flux, Sink}
import org.apache.pekko.stream.scaladsl.Sink as PekkoSink
import io.rsocket.core.{RSocketServer, SocketAcceptor}
import io.rsocket.transport.netty.server.TcpServerTransport
import io.rsocket.Payload

def serverSink: PekkoSink[Payload, NotUsed] = {
    // 创建Reactor多播Sink，替代废弃的EmitterProcessor
    val reactorSink: Sink.Many[Payload] = Sink.many().multicast().onBackpressureBuffer()
    val flux: Flux[Payload] = reactorSink.asFlux()

    // 启动RSocket服务器，绑定请求流处理逻辑
    RSocketServer.create(
      SocketAcceptor.forRequestStream(_ => flux)
    ).bindNow(TcpServerTransport.create("localhost", 3141))

    // 将Reactor Sink转换为Pekko Sink
    PekkoSink.fromSubscriber(reactorSink.asSubscriber())
}

为什么之前的Flow.toProcessor方案失效 #

你之前用Flow[Payload].toProcessor.run()得到的是Pekko原生Processor，直接用Flux.from(processor)包装为Reactor Flux时，两者的背压处理、订阅生命周期逻辑没有对齐，导致写入Pekko Sink的数据无法被RSocket的Flux订阅者正确接收。而Reactor原生的Sink.Many能完美适配RSocket的Reactor生态，再通过标准Subscriber转换为Pekko Sink，数据流就能正常流转。

额外优化选项 #

如果需要更精细的流控制，可以调整Sink.many()的参数：

• 单订阅者场景用unicast()替代multicast()：内存开销更低
• 给onBackpressureBuffer设置容量上限：避免无限制缓存导致内存溢出

内容的提问来源于stack exchange，提问作者David Masters