本文目录导读：
我将为您提供一个完整的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"));
}
}
这个完整的流程编排案例具有以下特点：
- 灵活性：支持各种节点类型（开始、任务、条件、循环、并行、结束）
- 可扩展性：通过接口定义节点行为，易于添加新节点类型
- 业务解耦：流程定义与业务逻辑分离，便于维护
- 配置驱动：支持通过配置定义流程流转关系
- 监控支持：可以统计各节点执行时间
使用这个框架可以方便地实现复杂的业务流程编排，如订单处理、审批流程、数据处理等。
