具有匿名线程的反应器调度程序答案

【问题标题】：Reactor Scheduler with anonymous thread具有匿名线程的反应器调度程序
【发布时间】：2019-01-20 09:35:48
【问题描述】：

我正在测试 reactor 的工作原理，创建的代码与 reactor 文档中的代码非常相似。

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

@SpringBootTest
@RunWith(SpringRunner.class)
public class ReactorApplicationTests {

  @Test
  public void publishOnThreadTest() {
    Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);

    final Mono<String> mono = Mono.just("Publish on test: \n")
            .map(msg -> msg + "before: " + Thread.currentThread() )
            .publishOn(s)
            .map(msg -> msg + "\nafter: " + Thread.currentThread());

    new Thread(() -> mono.subscribe(System.out::println)).start();
  }
}

我无法让它运行，我做错了什么？只需订阅它就可以工作，但我想看看使用的线程并玩一下它。

【问题讨论】：

1. “我无法让它运行”——会发生什么？ 2. 添加导入语句，以便我们可以重现行为。
我不能运行这个我的意思是没有打印任何结果。它适用于上述订阅在新线程之外。

标签： java multithreading project-reactor reactor

【解决方案1】：

您的测试程序什么也不打印的原因是它退出得太早了。它应该等到 substriber 的方法被调用：

@Test
public void publishOnThreadTest() throws InterruptedException {
    Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);
    CountDownLatch latch = new CountDownLatch(1);

    final Mono<String> mono = Mono.just("Publish on test: \n")
            .map(msg -> msg + "before: " + Thread.currentThread() )
            .publishOn(s)
            .map(msg -> msg + "\nafter: " + Thread.currentThread());

    new Thread(() -> mono.subscribe((String str) ->{
        System.out.println(str);
        latch.countDown();
    })).start();

    latch.await();
}

【讨论】：