如何在 spring boot 中实现对一个数据对象的流式处理？

[zhijiansha]

如何在 spring 中实现对同一个数据对象，实现流式处理？ 即，定义某一接口，同时定义多个处理类来实现这个接口，在处理数据时，依次（并行）调用这些处理类，在处理类内部，根据业务需求，判断是否需要进行处理，不处理的直接 return 。

现在的方式是使用策略模式，通过 Autowired 自动注入这些处理类，放入 map ，根据业务类型取出不同的处理类进行处理。

不知这是否是在 spring boot 中的最佳实现方式？ 如果同时需要多个处理类进行处理（如上所述），是否只能在自动注入时放入一个 List 中，进行遍历处理？

public interface Strategy {
   Integer getBizCode();
   public int doOperation(int num1, int num2);
}



@Component
public class OperationAdd implements Strategy{
    @Override
    public Integer getBizCode() {
        return 1;
    }

   @Override
   public int doOperation(int num1, int num2) {
      return num1 + num2;
   }
}

@Component
public class OperationSubtract implements Strategy{

    @Override
    public Integer getBizCode() {
        return 2;
    }

   @Override
   public int doOperation(int num1, int num2) {
      return num1 - num2;
   }
}



public class StrategyPatternDemo {

    HashMap<Integer, Strategy> strategyMap = new HashMap<>();

    @Autowired
    public StrategyPatternDemo(List<Strategy> strategies){
        for (Strategy strategy : strategies) {
            strategyMap.put(strategy.getCode(),strategy);
        }
    }

   public void test(int type, int num1, int num2) {
        Strategy strategy = strategyMap.get(type);
        int result = strategy.doOperation(num1, num2);
        System.out.println(result);
   }
}

[yjll9102]

看看是否可以使用责任链来处理

[shaneflane]

https://github.com/dromara/liteflow

[damai0419]

感觉也是责任链模式，听起来比较符合。

[zhijiansha]

@yjll9102 #1
@damai0419 #3
感谢，好像对于依次调用的场景是可以的，那如果对于没有前后逻辑的，可以并行处理的场景，是否有其他的方式呢？

[damai0419]

@zhijiansha
包装一下逻辑，内部用异步多线程处理？
或者直接用 CompletableFuture 处理？

[zmal]

数据结构可以选择一个支持并发的队列。

你可以说一下具体的场景，感觉像是一个 A-B 问题。

[vvtf]

```java

interface Rule<T extends S> {

T run(T... args);

default boolean enabled() {
return true;
}

default boolean async() {
return false;
}

}

@Order(0)
@Component
class RuleA implements Rule<SomeA> {

T run(T... args) {
// TODO
}

boolean async() {return true;}

}

@Component
class RuleChain {
@Autowired
private List<Rule> rules;

static ExecutorService pool = xxxx;

public void run(T... args) {

rules.stream().filter(Rule::enabled).filter(Rule::async).forEach(rule -> pool.execute(rule.run()););
rules.stream().filter(Rule::enabled).filter(rule -> !rule.async()).forEach(Rule::run);

}

}

```

[zhijiansha]

@damai0419 #5
@zmal #6
抱歉，可能我的描述有些问题，我现在面临的场景，是没有先后逻辑的，可以乱序执行，而责任链模式，好像是必须要指定 next ？比如有 A 、B 、C 三个处理类，需要执行情况 可能是 AB 、AC 、ABC 这种，我现在能想到的就是把这些处理类都放进一个 list 中，通过 for 循环，传入复制的数据对象进行处理

[zhijiansha]

@vvtf #7 感谢感谢

[zmal]

如果你的执行函数可以不区分前后，那还叫流式处理吗？
真实场景到底是什么？建议讲清楚。

[genesys]

“ 依次（并行）调用这些处理类” 到底是依次还是并行，差别很大的。
如果是并行的话，首先要选择如何实现并行。具体并行下执行的 task ，也许你可以考虑策略模式之类的。

[zhijiansha]

@zmal #10
@genesys #11
抱歉，标题和内容可能表述有误，我想请教 依次调用、并行调用这两种情况下的最佳实现方式。
#1 、#3 提到的责任链能解决依次调用
#7 提供的代码 demo 可以解决 并行调度
感谢

[genesys]

@zhijiansha 没关系。多提问挺好，这也是对自己提问能力的锻炼，提问能力是工作中很重要的一项技能。

[kytrun]

如果不需要拿到每一步的结果，直接 @ Async 完事儿

[0xLittleFi]

可以搜下开源的 easyflow 是咋实现的

[zhangleshiye]

@zhijiansha 根据 8 楼的信息 ， 我感觉楼主是不是想要响应式编程的那种链式编程？类似
rxjava observable.zip<A,B,C>(getDataA(),getDataB) .(new subscribtion(A a,B ,b){
return ConvertC(a,b)
})
这样？
好久没用 rxjava 了 , 大概意思是这样..