带有弹簧反应器项目的反应流背压答案

【问题标题】：Reactive stream backpressure with spring reactor project带有弹簧反应器项目的反应流背压
【发布时间】：2016-09-20 13:20:12
【问题描述】：

我研究并阅读了他们不是很容易理解的文件。我想要实现的是以下功能：

我正在使用 Spring Reactor 项目并使用 eventBus。我的事件总线正在向模块 A 抛出事件。

模块 A 应该接收事件并插入到将保存唯一值的 Hot Stream。每 250 Milisecons 流应该提取所有值并对它们进行计算......等等。

例如： eventBus 正在抛出事件，编号为：1,2,3,2,3,2

Stream 应该获取并保存唯一值 -> 1,2,3 250 毫秒后，流应该打印数字和空值

有人知道如何开始吗？我尝试了这些示例，但没有任何效果，我想我不明白。谁有例子？

Tnx

编辑：

当尝试做下一个时，我总是得到异常：

        Stream<List<Integer>> s = Streams.wrap(p).buffer(1, TimeUnit.SECONDS);

        s.consume(i -> System.out.println(Thread.currentThread() + " data=" + i));

        for (int i = 0; i < 10000; i++) {
            p.onNext(i);
        }

例外：

java.lang.IllegalStateException: The environment has not been initialized yet
    at reactor.Environment.get(Environment.java:156) ~[reactor-core-2.0.7.RELEASE.jar:?]
    at reactor.Environment.timer(Environment.java:184) ~[reactor-core-2.0.7.RELEASE.jar:?]
    at reactor.rx.Stream.getTimer(Stream.java:3052) ~[reactor-stream-2.0.7.RELEASE.jar:?]
    at reactor.rx.Stream.buffer(Stream.java:2246) ~[reactor-stream-2.0.7.RELEASE.jar:?]
    at com.ta.ng.server.controllers.user.UserController.getUsersByOrgId(UserController.java:70) ~[classes/:?]

如您所见，如果不通过此问题，我将无法继续尝试。

顺便说一句：仅当我使用buffer(1, TimeUnit.SECONDS) 时才会发生这种情况，例如，如果我使用buffer(50)，它可以工作。虽然这不是最终解决方案，但它是一个开始。

【问题讨论】：

标签： spring project-reactor

【解决方案1】：

好吧，在再次阅读文档后，我错过了这个：

static {
        Environment.initialize();
    }

这解决了问题。天呐

【讨论】：