弹簧反应堆项目的反应流背压(Reactive stream backpressure with spring reactor project)

 我有研究和阅读文件，他们不是很容易理解。 我想要实现的是以下功能： 
 
 我正在使用Spring Reactor项目并使用eventBus。 我的活动巴士正在向模块A投掷活动 
 
 模块A应接收事件并插入到将保存唯一值的热流中。 每250个Milisecons溪流应拉动所有的价值，并在他们身上进行冥想......等等。 
 
 例如：eventBus正在抛出事件，其编号为1,2,3,2,3,2 
 
 流应该得到并保持唯一值 - > 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) ，例如它的工作..虽然这不是最终的解决方案，它的一个开始。 

I have research and read documents by they are not very understandable. What i am trying to achieve is the following functionality:
 
I am using Spring Reactor project and using the eventBus. My event bus is throwing event to module A.
 
Module A should receive the event and insert into Hot Stream that will hold unique values. Every 250 Milisecons the stream should pull all value and make calulcation on them.. and so on.
 
For example: The eventBus is throwing event with number: 1,2,3,2,3,2
 
The Stream should get and hold unique values -> 1,2,3 After 250 miliseconds the stream should print the number and empty values
 
Anyone has an idea how to start? I tried the examples but nothing really works and i guess i don't understand something. Anyone has an example?
 
Tnx
 
EDIT:
 
When trying to do the next i always get exception:
 
        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);
        }
 
The exception:
 
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/:?]
 
As you can see i cannot proceed trying without passing this issue.
 
BY THE WAY: This is happeing only when i use buffer(1, TimeUnit.SECONDS) If i use buffer(50) for example it works.. Although this is not the final solution its a start. 

原文：https://stackoverflow.com/questions/37394746