流量中的流数据计算

Question

var period = 3;
int [] inArr = {2, 4, 6, 7, 9, 11, 13, 16, 17, 18, 20, 22 };

var upbond = inArr[0] + period;
var count =0;
List<Integer> result = new ArrayList();
for(int a: inArr){
  if(a <= upbond){
    count++;
  }else{
    result.add(count);
    count = 1;
    upbond += period;
  }
}
result.add(count);
System.out.println(Arrays.toString(result.toArray()));

Answer 1

scan()变体允许您使用单独类型的累加器。

我将使用一个单独的state类来解决这个问题：

public class State {
    private int count;
    private Optional<Integer> upbond;
    private Optional<Integer> result;

    public State() {
        this.count = 0;
        this.upbond = Optional.empty();
        this.result = Optional.empty();
    }

    public State(int count, int upbond) {
        this.count = count;
        this.upbond = Optional.of(upbond);
        this.result = Optional.empty();
    }

    public State(int count, int upbond, int result) {
        this.count = count;
        this.upbond = Optional.of(upbond);
        this.result = Optional.of(result);
    }

    public int getCount() {
        return count;
    }

    public Optional<Integer> getUpbond() {
        return upbond;
    }

    public Optional<Integer> getResult() {
        return result;
    }
}

...然后使用scan()逐个元素构建状态：

sourceFlux
        .concatWithValues(0)
        .scan(new State(), (state, a) ->
                a <= state.getUpbond().orElse(a + period) ?
                        new State(state.getCount() + 1, state.getUpbond().orElse(a + period)) :
                        new State(1, state.getUpbond().orElse(a + period) + period, state.getCount())

        )
        .windowUntil(s -> s.getResult().isPresent())
        .flatMap(f -> f.reduce((s1, s2) -> s1.getResult().isPresent()?s1:s2).map(s -> s.getResult().orElse(s.getCount() - 1)))