调用publishOn时如何获取线程信息日志

Question

我正在使用反应流中的调度程序，并使用 publishOn 方法在此 Flux 上使用 Flux 和调度程序，如下所示：

    System.out.println("*********Calling Concurrency************");
    List<Integer> elements = new ArrayList<>();
    Flux.range(1, 1000)
      .log()
      .map(i -> i * 2)
      .publishOn(Schedulers.parallel())
      //.subscribeOn(Schedulers.parallel())
      .subscribe(elements::add);
    System.out.println("-------------------------------------");

为此我得到了以下信息日志：

*********Calling Concurrency************
[info] | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
[info] | request(256)
[info] | onNext(1)
[info] | onNext(2)
[info] | onNext(3)
[info] | onNext(4)
[info] | onNext(5)
[info] | onNext(6)
[info] | onNext(7)
[info] | onNext(8)
[info] | onNext(9)
.....
.....
[info] | onNext(444)
[info] | onNext(445)
[info] | onNext(446)
[info] | onNext(447)
[info] | onNext(448)
-------------------------------------
[info] | request(192)
.....
.....
[info] | onNext(999)
[info] | onNext(1000)
[info] | onComplete()
[info] | request(192)

并且没有关于线程执行和处理的信息。另外，有时需要 192 个元素，有时需要 256 个元素。

这是我正在使用的依赖项：

<dependency> 
     <groupId>com.googlecode.slf4j-maven-plugin-log</groupId>
     <artifactId>slf4j-maven-plugin-log</artifactId> 

     <version>1.0.0</version> 
</dependency>

如何获取当前thread/parallel线程执行的日志信息？请提出建议。

Answer 1

要获取所有必需的信息，只需在 .publishOn()

之后输入 .log()

    Flux.range(1, 1000)
        .map(i -> i * 2)
        .publishOn(Schedulers.parallel())
        //.subscribeOn(Schedulers.parallel())
        .log()
        .blockLast();

最终输出如下：

        12:13:21.964 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
        12:13:21.997 [main] INFO reactor.Flux.PublishOn.1 - | onSubscribe([Fuseable] FluxPublishOn.PublishOnSubscriber)
        12:13:21.999 [main] INFO reactor.Flux.PublishOn.1 - | request(unbounded)
        12:13:22.002 [parallel-1] INFO reactor.Flux.PublishOn.1 - | onNext(2)
        12:13:22.002 [parallel-1] INFO reactor.Flux.PublishOn.1 - | onNext(4)
        12:13:22.002 [parallel-1] INFO reactor.Flux.PublishOn.1 - | onNext(6)
        12:13:22.002 [parallel-1] INFO reactor.Flux.PublishOn.1 - | onNext(8)
        12:13:22.002 [parallel-1] INFO reactor.Flux.PublishOn.1 - | onNext(10)
        12:13:22.002 [parallel-1] INFO reactor.Flux.PublishOn.1 - | onNext(12)

所以在那种情况下，有关 Thread 的信息将在那里

Note: Output might varying depends on used Logger library

Answer 2

这看起来像是您这边的记录器配置错误。

Reactor 会发现您使用的是 SLF4J，但您仍然需要正确配置的日志记录实现，例如 Logback，以及将记录线程名称的相关附加程序。

如果您根本不想使用日志记录框架，可以调用 Loggers.useConsoleLoggers()，它将以包括当前线程名称的简化格式打印在控制台上。不过，我建议只针对 1 次运行（如调试会话）执行此操作，而不是在生产代码中...

调用publishOn时如何获取线程信息日志

How to get thread information logs when calling publishOn

project-reactor

publisher

reactive-streams