Flux 中的流数据计算

Question

以下代码说明了我在 Spring Reactive 项目中需要的逻辑：

输入：

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()));

排序整数的数据源是来自 DB 的 Flux，一旦有新的合适的数据写入 DB，它将持续获取数据。结果应该是一个流，通过RSocket（通过请求流通信方式）发送到另一个节点。

在 Reactor 上进行了一些在线搜索，包括一些教程，我仍然无法弄清楚如何以 Flux 方式编写逻辑。我遇到的困难是那些在循环之外定义的数据计算。

我应该如何在 Reactor 中处理它？

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)))

除此之外：concatWithValues() / windowUntil() / flatMap() 位用于处理最后一个元素 - 可能有一种更简洁的方法来实现这一点，如果我想到它，我会编辑答案。

Answer 2

我认为 scan 在这里绝对是合适的工具，结合有状态类，虽然我的方法与 Michaels 略有不同。

累加器：

class UpbondAccumulator{

    final Integer period;
    Integer upbond;
    Integer count;
    Boolean first;
    Queue<Integer> results;

    UpbondAccumulator(Integer period){
        this.period = period;
        this.count = 0;
        this.upbond = 0;
        this.results = new ConcurrentLinkedQueue<>();
        this.first = true;
    }

    //Logic is inside accumulator, since accumulator is the only the only thing 
    //that needs it. Allows reuse of accumulator w/o code repetition
    public UpbondAccumulator process(Integer in){
        //If impossible value
        //Add current count to queue and return
        //You will have to determine what is impossible
        //Since we concat this value on the end of flux
        //It will signify the end of processing
        //And emit the last count 
        if(in<0){
            results.add(count);
            return this;
        }
        //If first value
        //Do stuff outside loop
        if(this.first) {
            upbond = in + period;
            first=false;
        }
        //Same as your loop
        if(in <= upbond)
            count++;
        else {
            results.add(count);
            count = 1;
            upbond += period;
        }
        //Return accumulator
        //This could be put elsewhere since it isn't
        //Immediately obvious that `process` should return
        //the object but is simpler for example
        return this;
    }

    public Mono<Integer> getResult() {
        //Return mono empty if queue is empty
        //Otherwise return queued result
         return Mono.justOrEmpty(results.poll());
    }
}

用法：

    dbFlux
            //Concat with impossible value
            .concatWithValues(-1)
            //Create accumulator, process value and return
            .scan(new UpbondAccumulator(period), UpbondAccumulator::process)
            //Get results, note if there are no results, this will be empty
            //meaning it isn't passed on in chain
            .flatMap(UpbondAccumulator::getResult)

---

以下 Michael 的评论是一种不可变的方法

累加器：

public class UpbondAccumulator{

    public static UpbondState process(int period,Integer in,UpbondState previous){

        Integer upbond = previous.getUpbond().orElse(in + period);
        int count = previous.getCount();

        if(in<0) return new UpbondState(upbond, count, count);

        if(in <= upbond) return new UpbondState(upbond,count + 1 , null);

        return new UpbondState(upbond + period, 1, count);
    }
}

状态对象：

public class UpbondState {
    private final Integer upbond;
    private final int count;
    private final Integer result;

    public UpbondState() {
        this.count = 0;
        this.upbond = null;
        this.result = null;
    }

    public UpbondState(Integer upbond, int count,Integer result) {
        this.upbond = upbond;
        this.count = count;
        this.result = result;
    }

    public int getCount() { return count; }
    public Optional<Integer> getUpbond() { return Optional.ofNullable(upbond); }
    public Integer getResult() { return result; }
    public boolean hasResult() { return result!=null; }
}

用法：

    dbFlux
            .concatWithValues(-1)
            .scan(new UpbondState(), 
                    (prev, in) -> UpbondAccumulator.process(period,in,prev))
            //Could be switched for Optional, but would mean one more map
            //+ I personally think makes logic less clear in this scenario
            .filter(UpbondState::hasResult)
            .map(UpbondState::getResult)