[waterflow] support configuring node concurrency

Author: loveTsong

framework/fel/java/fel-flow/src/main/java/modelengine/fel/engine/activities/AiState.java

@@ -82,6 +82,18 @@ public AiState<O, D, I, RF, F> id(String id) {
         return this;
     }
 
+    /**
+     * Sets the maximum concurrency level for this state's processing pipeline.
+     *
+     * @param concurrency The maximum number of concurrent operations allowed (must be positive).
+     * @return The current state instance for method chaining.
+     * @throws IllegalArgumentException If the concurrency value is zero or negative.
+     */
+    public AiState<O, D, I, RF, F> concurrency(int concurrency) {
+        this.state.concurrency(concurrency);
+        return this;
+    }
+
     /**
      * 获取当前节点的数据订阅者。
      *

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/states/State.java

@@ -104,6 +104,18 @@ public State<O, D, I, F> id(String id) {
         return ObjectUtils.cast(super.setId(id));
     }
 
+    /**
+     * Sets the maximum concurrency level for this state's processing pipeline.
+     *
+     * @param concurrency The maximum number of concurrent operations allowed (must be positive).
+     * @return The current state instance for method chaining.
+     * @throws IllegalArgumentException If the concurrency value is zero or negative.
+     */
+    public State<O, D, I, F> concurrency(int concurrency) {
+        ObjectUtils.<Node<I, O>>cast(this.processor).setMaxConcurrency(concurrency);
+        return this;
+    }
+
     /**
      * 跳转到指定节点，使用节点的唯一标识来标识一个节点。
      * <p>

framework/waterflow/java/waterflow-core/src/main/java/modelengine/fit/waterflow/domain/stream/nodes/To.java

@@ -100,6 +100,8 @@ public class To<I, O> extends IdGenerator implements Subscriber<I, O> {
     @Getter
     private final FlowLocks locks;
 
+    private int maxConcurrency = MAX_CONCURRENCY;
+
     // 默认自动流转过滤器是按batchID批次过滤contexts
     private final Operators.Filter<I> defaultAutoFilter = (contexts) -> {
         if (CollectionUtils.isEmpty(contexts)) {
@@ -606,7 +608,7 @@ private synchronized void updateConcurrency(int newConcurrency) {
      * @return true-已经满负载， false-未满负载
      */
     public boolean isOverLimit() {
-        return this.curConcurrency >= MAX_CONCURRENCY;
+        return this.curConcurrency >= this.maxConcurrency;
     }
 
     /**
@@ -736,6 +738,16 @@ public void emit(O data, FlowSession session) {
         this.listeners.values().forEach(listener -> listener.handle(data, session));
     }
 
+    /**
+     * Sets the maximum concurrency level for this state's processing pipeline.
+     *
+     * @param concurrency The maximum number of concurrent operations allowed (must be positive).
+     * @throws IllegalArgumentException If the concurrency value is zero or negative.
+     */
+    public void setMaxConcurrency(int concurrency) {
+        this.maxConcurrency = Validation.greaterThan(concurrency, 0, "The concurrency should greater than 0.");
+    }
+
     private FlowSession getNextSession(FlowSession session) {
         return FlowSessionRepo.getNextToSession(this.streamId, session);
     }