java - 对 Stream 的终端调用永远不会执行

Question

我很难使用 Spring Reactor Stream API(类似于 rxjava)在我的服务中构造一个响应对象，包装由两个下游服务提供的响应。

下面是accept()我的 channel 消费者上的方法。为了保护无辜者，一些名称已更改..

@Overridepublic void accept(final NetChannel channel) {    log.info("Consuming NetChannel of FullHttpRequest");    try {        // Our initial Stream is a single HTTP request containing our XML document.        Stream requestStream = channel.in();        requestStream.filter((x) -> {return true;})                .dispatchOn(dispatcher)                .map(new FooRequestFunction())                    // 1)                .flatMap(new BarRequestStreamFunction())          // 2)                .flatMap(new DownstreamRequestZipFunction())      // 3)                .toList()                                         // 4)                .onComplete(new ResponsesConsumer(channel));      // 5)    } catch (Exception e) {        log.error("Exception thrown during Channel processing", e);    }}

So, a FooRequest wraps many BarRequests, each of which has one associated Classify request, and one associated Validate request. We want to 1) Convert to a FooRequest, 2) Convert the FooRequest to a series of BarRequests, 3) Run two downstream requests for each BarRequest, 4) Aggregate all of our BarResponse objects into an overall response, 5) send a response back out to the client.

The point at which I encounter problems is the toList() method, which never seems to execute. Every time I've attempted something that involves a Promise it always seems to break, and this has been no exception.

FooRequestFunction, BarRequestStreamFunction are fairly simple, and seem to run fine. Their method signatures are:

// FooRequestFunction
public FooRequest apply(final FullHttpRequest request);

和:

// BarRequestStreamFunction
public Stream<BarRequest> apply(FooRequest dsoRequests);

DownstreamRequestZipFunction看起来像这样:

@Overridepublic Stream apply(BarRequest t) {     Stream classifyRes = Streams            .just(t)            .flatMap(new ClassifyDownstreamRequestFunction());    Stream validateRes = Streams            .just(t)            .flatMap(new ValidateDownstreamRequestFunction());    return Streams.zip(classifyRes, validateRes, (tuple) -> {        BarResponse response = new BarResponse();        response.setClassifyRes(tuple.getT1());        response.setValidateRes(tuple.getT2());        return response;    });}

This seems to work fine, as long as both of the downstream request functions return a result.

Finally, the Consumer at the end of the chained calls has this signature:

// ResponsesConsumer
public void accept(Promise<List<BarResponse>> responses)

它所做的是 await() 响应 promise ，然后将所有这些响应聚合到单个 XML 文档中写回 channel 。我可以告诉执行永远不会达到这个方法，因为没有日志记录触发。这一切似乎都停在了 .toList() 上。

有谁知道为什么这个设置似乎执行 toList()还是之后呢？

编辑:好的，我有更多信息。在为应用程序中的每个线程提供命名约定以使调试更容易之后，我可以看到“shared-1”，运行 accept() 方法的线程进入 WAITING 状态，然后停留在那里。这可能与底层 Dispatcher 是一个环形缓冲区调度程序这一事实有关，它是单线程的。

我修改了代码，使方法略有不同，并使用多线程调度程序，并避免使用 Promise ，但我仍然有一个状态，在该状态下，链式调用集的尾部将不会执行。见下文:

@Overridepublic void accept(final NetChannel channel) {    log.info("Consuming NetChannel of FullHttpRequest");    try {        // Our initial Stream is a single HTTP request containing our XML document.        Stream requestStream = channel.in();        requestStream.filter((x) -> {return true;})                .dispatchOn(dispatcher)                .map(new FooRequestFunction())                    // 1)                .flatMap(new BarRequestStreamFunction())          // 2)                .flatMap(new DownstreamRequestZipFunction())      // 3)                .reduce(new ArrayList(), (list,resp) -> {log.info("Reducing"); list.add(resp); return list;})                                        // 4)                .consumeOn((x)->{log.info("Consume");}, (x)->{log.error("error");}, (x)->{log.info("Complete");}, dispatcher);      // 5)    } catch (Exception e) {        log.error("Exception thrown during Channel processing", e);    }}

在上面，我用对 reduce() 的调用替换了 toList() 并将所有内容折叠成一个 List<BarResponse> .我可以看到这个执行和日志记录很好。但是，无论我对最后一次调用做什么，在尝试使用 consume()、consumeOn() 等之后，它都不会执行，也不会记录您在上面看到的最终调用。

在 VisualVM 中，我可以看到调度程序线程都在与阻塞队列关联的同一个对象监视器上等待 - 换句话说，它们都在等待工作到达。这就像完全忽略尾部 consumeOn() 调用。

我在这里做错了什么？我有什么不明白的？

编辑 2:鉴于约翰在下面的回复，我怀疑问题出在服务器设置上。可能仅用于 reactor 版本 2.0.0.M2，它在主 Application 中配置类如下:

 @Bean    public NetServer httpServer(            final Environment env,            final MetricRegistry metrics,            final ChannelConsumer consumer) throws InterruptedException {        NetServer server = new TcpServerSpec(                NettyTcpServer.class)                .env(env)                .options(                        new NettyServerSocketOptions()                            .pipelineConfigurer((ChannelPipeline pipeline) -> pipeline                                .addLast(new HttpServerCodec())                                .addLast(new HttpObjectAggregator(MAX_CONTENT_LENGTH))))                .consume(consumer)                .get();        server.start().await();        return server;    }

没有为此配置调度程序，它似乎在后台使用 LMAX 干扰器，而不是 NettyEventLoopDispatcher .不清楚如何设置NettyEventLoopDispatcher并将其用作替代调度程序。

Answer 1

consumeOn() 调用是多余的，因为您已经在那个 Dispatcher 上了(除非您的“真实”代码使用与此示例不同的东西)。当您编写输出时，无论如何都会在内部切换到 NettyEventLoopDispatcher。

我做了一个快速检查以确保此流程在 TcpServer 之外工作并且按预期工作:

@Test
public void testStream() throws InterruptedException {
    Stream<String> s1 = Streams.just("Hello World!");

    s1.filter(s -> s.startsWith("Hello"))
      .dispatchOn(Environment.sharedDispatcher())
      .map(s -> s.toUpperCase())
      .flatMap(s -> Streams.just(s, s))
      .flatMap(s -> Streams.just(s, s))
      .reduce(new ArrayList<>(), (l, s) -> {
          l.add(s);
          return l;
      })
      .consume(l -> {
          System.out.println("thread: " + Thread.currentThread() + ", l: " + l);
      });

    Thread.sleep(500);
}

通过简单地注释掉 .dispatchOn() 调用来了解流程是否在 Netty 线程上运行会很有趣。