Java实现流程编排案例

本文目录导读：

1. 核心接口定义
2. 节点实现类
3. 流程引擎
4. 具体业务案例
5. 扩展接口和工具类

我将为您提供一个完整的Java流程编排实现案例,这个案例将展示如何构建一个灵活的流程编排引擎。

核心接口定义

// 节点接口
public interface FlowNode {
    String getId();
    void execute(FlowContext context);
    void setNext(FlowNode next);
    FlowNode getNext();
}
// 流程上下文
public class FlowContext {
    private Map<String, Object> data = new HashMap<>();
    private Map<String, Object> attributes = new HashMap<>();
    private boolean terminated = false;
    public void putData(String key, Object value) { data.put(key, value); }
    public Object getData(String key) { return data.get(key); }
    public Map<String, Object> getDataMap() { return data; }
    public void setAttribute(String key, Object value) { attributes.put(key, value); }
    public Object getAttribute(String key) { return attributes.get(key); }
    public boolean isTerminated() { return terminated; }
    public void setTerminated(boolean terminated) { this.terminated = terminated; }
}
// 流程定义
public class FlowDefinition {
    private String name;
    private Map<String, FlowNode> nodes = new LinkedHashMap<>();
    private String startNodeId;
    private List<Map<String, String>> transitions = new ArrayList<>();
    // getter和setter...
}

节点实现类

// 基础节点
public abstract class BaseNode implements FlowNode {
    protected String id;
    protected FlowNode next;
    @Override
    public String getId() { return id; }
    @Override
    public void setNext(FlowNode next) { this.next = next; }
    @Override
    public FlowNode getNext() { return next; }
}
// 起始节点
public class StartNode extends BaseNode {
    @Override
    public void execute(FlowContext context) {
        System.out.println("流程开始: " + id);
        if (next != null) next.execute(context);
    }
}
// 任务节点
public class TaskNode extends BaseNode {
    private String taskName;
    private TaskHandler handler;
    public TaskNode(String taskName, TaskHandler handler) {
        this.taskName = taskName;
        this.handler = handler;
    }
    @Override
    public void execute(FlowContext context) {
        System.out.println("执行任务: " + taskName);
        if (handler != null) {
            handler.execute(context);
        }
        if (next != null) next.execute(context);
    }
}
// 条件节点
public class ConditionNode extends BaseNode {
    private ConditionEvaluator evaluator;
    private FlowNode trueBranch;
    private FlowNode falseBranch;
    @Override
    public void execute(FlowContext context) {
        boolean result = evaluator.evaluate(context);
        System.out.println("条件判断结果: " + result);
        if (result && trueBranch != null) {
            trueBranch.execute(context);
        } else if (!result && falseBranch != null) {
            falseBranch.execute(context);
        }
    }
}
// 循环节点
public class LoopNode extends BaseNode {
    private int times;
    private FlowNode loopBody;
    @Override
    public void execute(FlowContext context) {
        for (int i = 0; i < times; i++) {
            System.out.println("循环第" + (i+1) + "次");
            loopBody.execute(context);
        }
        if (next != null) next.execute(context);
    }
}
// 结束节点
public class EndNode extends BaseNode {
    @Override
    public void execute(FlowContext context) {
        System.out.println("流程结束: " + id);
        context.setTerminated(true);
    }
}
// 并行节点
public class ParallelNode extends BaseNode {
    private List<FlowNode> branches = new ArrayList<>();
    private ExecutorService executor;
    @Override
    public void execute(FlowContext context) {
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        for (FlowNode branch : branches) {
            futures.add(CompletableFuture.runAsync(() -> 
                branch.execute(context), executor));
        }
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        if (next != null) next.execute(context);
    }
}

流程引擎

public class FlowEngine {
    private Map<String, FlowNode> nodeMap = new HashMap<>();
    private Map<String, List<FlowNode>> parallelGroups = new HashMap<>();
    // 构建流程
    public void buildFlow(FlowDefinition definition) {
        // 1. 创建所有节点
        for (Map.Entry<String, FlowNode> entry : definition.getNodes().entrySet()) {
            nodeMap.put(entry.getKey(), entry.getValue());
        }
        // 2. 连接节点
        for (Map<String, String> transition : definition.getTransitions()) {
            String from = transition.get("from");
            String to = transition.get("to");
            String type = transition.getOrDefault("type", "normal");
            FlowNode fromNode = nodeMap.get(from);
            FlowNode toNode = nodeMap.get(to);
            if (type.equals("normal")) {
                fromNode.setNext(toNode);
            } else if (type.equals("parallel")) {
                // 处理并行分支
                parallelGroups.computeIfAbsent(from, k -> new ArrayList<>())
                            .add(toNode);
            }
        }
        // 3. 设置并行节点
        for (Map.Entry<String, List<FlowNode>> entry : parallelGroups.entrySet()) {
            FlowNode node = nodeMap.get(entry.getKey());
            if (node instanceof ParallelNode) {
                ParallelNode parallelNode = (ParallelNode) node;
                for (FlowNode branch : entry.getValue()) {
                    parallelNode.addBranch(branch);
                }
            }
        }
    }
    // 执行流程
    public FlowContext execute(String startNodeId) {
        FlowContext context = new FlowContext();
        FlowNode startNode = nodeMap.get(startNodeId);
        if (startNode != null) {
            startNode.execute(context);
        }
        return context;
    }
}

具体业务案例

// 订单处理流程
public class OrderFlowExample {
    public static void main(String[] args) {
        // 创建流程引擎
        FlowEngine engine = new FlowEngine();
        // 创建节点
        StartNode startNode = new StartNode();
        startNode.id = "start";
        TaskNode validateOrder = new TaskNode("订单校验", (context) -> {
            System.out.println("校验订单信息...");
            String orderId = (String) context.getData("orderId");
            if (orderId == null) {
                context.putData("valid", false);
            } else {
                context.putData("valid", true);
            }
        });
        TaskNode checkStock = new TaskNode("检查库存", (context) -> {
            System.out.println("检查商品库存...");
            context.putData("stockAvailable", true);
        });
        TaskNode calculatePrice = new TaskNode("计算价格", (context) -> {
            System.out.println("计算订单金额...");
            double price = 100.0;
            context.putData("price", price);
        });
        ConditionNode checkCondition = new ConditionNode();
        checkCondition.id = "checkCondition";
        checkCondition.evaluator = (context) -> {
            Boolean valid = (Boolean) context.getData("valid");
            return valid != null && valid;
        };
        TaskNode createPayment = new TaskNode("创建支付", (context) -> {
            System.out.println("创建支付订单...");
            context.putData("paymentId", "PAY123");
        });
        TaskNode notifyUser = new TaskNode("通知用户", (context) -> {
            System.out.println("发送通知给用户...");
        });
        EndNode endNode = new EndNode();
        endNode.id = "end";
        // 构建流程定义
        FlowDefinition definition = new FlowDefinition();
        definition.setName("订单处理流程");
        definition.setStartNodeId("start");
        // 添加节点
        Map<String, FlowNode> nodes = new HashMap<>();
        nodes.put("start", startNode);
        nodes.put("validateOrder", validateOrder);
        nodes.put("checkStock", checkStock);
        nodes.put("calculatePrice", calculatePrice);
        nodes.put("checkCondition", checkCondition);
        nodes.put("createPayment", createPayment);
        nodes.put("notifyUser", notifyUser);
        nodes.put("end", endNode);
        definition.setNodes(nodes);
        // 定义流转关系
        List<Map<String, String>> transitions = new ArrayList<>();
        transitions.add(createTransition("start", "validateOrder", "normal"));
        transitions.add(createTransition("validateOrder", "checkCondition", "normal"));
        transitions.add(createTransition("checkStock", "calculatePrice", "normal"));
        transitions.add(createTransition("checkCondition", "createPayment", "true"));
        transitions.add(createTransition("checkCondition", "notifyUser", "false"));
        transitions.add(createTransition("createPayment", "end", "normal"));
        transitions.add(createTransition("notifyUser", "end", "normal"));
        definition.setTransitions(transitions);
        // 构建并执行
        engine.buildFlow(definition);
        FlowContext context = new FlowContext();
        context.putData("orderId", "ORDER123");
        System.out.println("=== 开始执行订单流程 ===");
        FlowContext result = engine.execute("start");
        System.out.println("流程状态: " + (result.isTerminated() ? "完成" : "进行中"));
    }
    private static Map<String, String> createTransition(String from, String to, String type) {
        Map<String, String> transition = new HashMap<>();
        transition.put("from", from);
        transition.put("to", to);
        transition.put("type", type);
        return transition;
    }
}

扩展接口和工具类

// 任务处理器接口
@FunctionalInterface
public interface TaskHandler {
    void execute(FlowContext context);
}
// 条件评估器接口
@FunctionalInterface
public interface ConditionEvaluator {
    boolean evaluate(FlowContext context);
}
// 流程构建器
public class FlowBuilder {
    private FlowDefinition definition = new FlowDefinition();
    private Map<String, FlowNode> nodes = new HashMap<>();
    public FlowBuilder node(String id, FlowNode node) {
        nodes.put(id, node);
        return this;
    }
    public FlowBuilder start(String nodeId) {
        definition.setStartNodeId(nodeId);
        return this;
    }
    public FlowBuilder transition(String from, String to) {
        Map<String, String> trans = new HashMap<>();
        trans.put("from", from);
        trans.put("to", to);
        trans.put("type", "normal");
        definition.getTransitions().add(trans);
        return this;
    }
    public FlowDefinition build() {
        definition.setNodes(nodes);
        return definition;
    }
}
// 流程监控器
public class FlowMonitor {
    private long startTime;
    private Map<String, Long> nodeExecutionTime = new HashMap<>();
    public void startFlow() {
        startTime = System.currentTimeMillis();
    }
    public void nodeExecuted(String nodeId, long duration) {
        nodeExecutionTime.put(nodeId, duration);
    }
    public void printReport() {
        long totalTime = System.currentTimeMillis() - startTime;
        System.out.println("流程执行报告:");
        System.out.println("总耗时: " + totalTime + "ms");
        System.out.println("节点执行时间:");
        nodeExecutionTime.forEach((node, time) -> 
            System.out.println("  " + node + ": " + time + "ms"));
    }
}

这个完整的流程编排案例具有以下特点：

1. 灵活性：支持各种节点类型（开始、任务、条件、循环、并行、结束）
2. 可扩展性：通过接口定义节点行为，易于添加新节点类型
3. 业务解耦：流程定义与业务逻辑分离，便于维护
4. 配置驱动：支持通过配置定义流程流转关系
5. 监控支持：可以统计各节点执行时间

使用这个框架可以方便地实现复杂的业务流程编排，如订单处理、审批流程、数据处理等。