如何只从反应流中发射累积总和？

Question

我有一个用例，只有当累积“和”等于或超过给定值n时，流才应该发出。让我们以n=5的六个整数为例。

+---+------+---------+
| i | Emit |   Sum   |
+---+------+---------+
| 1 |    - | 1       |
| 2 |    - | 3       |
| 3 |    5 | 1       |
| 4 |    5 | 0       |
| 5 |    5 | 0       |
| 2 |    2 | 0 (end) |
+---+------+---------+

正如您所看到的，除非和等于或超过5，否则什么都不会发出，除了最后一个元素，它无论如何都会发出。

Answer 1

这不可能直接在flux对象上实现，但是如果您能够访问创建flux对象的资源，则可以实现解决方案。由于在stream（Flux）中，您不能访问上一个元素的，您可以在索引上为您的资源创建Flux，并直接从索引的Flux访问此资源（因为它是只读操作）。例如，如下所示：

List<Integer> list = List.of(1, 2, 3, 4, 5, 2);
AtomicReference<Integer> atomicSum = new AtomicReference<>(0);
return Flux.fromStream(IntStream.range(0, list.size() - 1).boxed())
        .flatMap(i -> {
            int sum = atomicSum.updateAndGet((integer -> integer + list.get(i)));
            if (sum >= 5) {
                atomicSum.updateAndGet(integer -> integer - 5);
                return Flux.just(5);
            }

            return (i.equals(list.size() -1))
                    ? Flux.just(list.get(i)) // emit last element even if sum was not 5
                    : Flux.empty();
        }); // emitted element's

请注意，这不是很好的做法，我不建议这样的解决方案。flux对象处理可能会在线程之间跳过，因此如果修改flux之外的对象，应该以同步的方式进行（因此使用atomicreference)。列表只用于只读操作，因此它是可以的。另外，我不知道这部分代码是否真的有效，但我想向您展示，如果您能够访问创建flux对象的资源，您将如何找到解决方案。

编辑：即使这样的解决方案也不起作用。我搞错了myslef,flux对象不会在线程之间跳过，但可能会被多个线程处理，导致单个原子引用处于无效状态。这个云仍然可以用一些同步机制来解决，比如锁，而不是原子引用，但这远远超出了一般开发人员的经验。您确定不能使用scan()函数，因为您可以提供自己的累加器函数作为参数吗？