java - 如何创建 Reactor Netty 热流

Question

我正在尝试找到一种创建热流的方法，我可以在其中以一种方法插入数据，而订阅者可以通过另一种方法获取数据。我已成功使用 a WorkQueueProcessor，但我不确定这是否是正确的做法。是否可以使用 Flux.create 做同样的事情？这是我的工作片段：

1. 称呼connect();

2. 向服务器发送字节数据，客户端将收到tcp服务器的响应并workQueueProcessor发出数据。

   @Component
   @RequiredArgsConstructor
   public class TcpCli {
   
       @Setter
       private Connection connection;
       private NettyOutbound out;
   
       //Creation of Work Queue Processor, can a Flux.create here can do the same job ?
       private WorkQueueProcessor<String> workQueueProcessor = WorkQueueProcessor.<String>builder().build();
       public Mono<? extends Connection> connect() {
           return TcpClient.create()
               .host(tcpConfig.getHost())
               .port(tcpConfig.getPort())
               .handle(this::handleConnection)
               .connect();
       }
       public Mono<String> sendData(ByteArray data) {
           out.sendByteArray(Mono.just(data)).then().subscribe();
   
           //Get emitted data from workQueueProcessor
           return workQueueProcessor.next();
       }
       private Publisher<Void> handleConnection(NettyInbound in, NettyOutbound out) {
           this.out = out;
           in.receive().asString()
               .log("In received")
               .subscribe(str -> {
                   LOGGER.info(String.format("Inbound: %s", str));
   
                   //Emit data to workQueueProcessor
                   workQueueProcessor.onNext(str);
               });
           return out
               .neverComplete()    //keep connection alive
               .log("Never close");
       }
   }
   

标签： javaspringspring-bootproject-reactorreactor-netty

Answer 1

你能解释一下@Setter 在这里做了什么吗？我尝试相同的代码，它总是给出

@Setter
private Connection connection;
private NettyOutbound out;