用户指南 04 · 扩展开发

四个扩展点：拦截器、事件监听、动态参与者、动态决策。本文给出各语言完整示例。

1. 扩展点总览

扩展点	时机	场景
FlowInterceptor	节点执行前后	日志、统计、条件拦截
ProcessEventListener	流程生命周期	通知（钉钉/企微/短信）、同步业务状态
IAssignmentHandler（Registry）	创建任务时	参与者按业务规则计算
IDecisionHandler（Registry）	决策节点	表达式表达不了的复杂路由

2. 动态参与者（最常见的扩展）

场景：审批人不是固定的，而是"按部门经理"算出来的。

2.1 实现处理器

// Java：实现 IAssignmentHandler 并注册
public class DeptLeaderHandler implements IAssignmentHandler {
    @Override
    public List<String> assign(ProcessInstance instance) {
        String deptId = (String) instance.getVariables().get("u_deptId");
        return List.of(deptService.findLeader(deptId).getUserId());
    }
}

# Python
class DeptLeaderHandler(IAssignmentHandler):
    async def assign(self, node, instance):
        dept_id = instance.variables.get("u_deptId")
        return [dept_service.find_leader(dept_id).user_id]

// Go
type DeptLeaderHandler struct{ DeptSvc *DeptService }
func (h *DeptLeaderHandler) Assign(node *model.FlowNode, inst *model.ProcessInstance) []string {
    deptID, _ := inst.Variables["u_deptId"].(string)
    return []string{h.DeptSvc.FindLeader(deptID).UserID}
}

// Node.js
class DeptLeaderHandler implements IAssignmentHandler {
  async assign(node: FlowNode, inst: ProcessInstance): Promise<string[]> {
    const deptId = inst.variables['u_deptId'] as string
    return [await deptService.findLeader(deptId).then(u => u.userId)]
  }
}

2.2 注册

engine.getRegistry().register("deptLeaderHandler", new DeptLeaderHandler());

engine.ext.registry.register_assignment("deptLeaderHandler", DeptLeaderHandler())

engine.SetRegistry(&engine.HandlerRegistry{})
engine.GetRegistry().RegisterAssignment("deptLeaderHandler", &DeptLeaderHandler{})

const reg = new HandlerRegistry()
reg.registerAssignment('deptLeaderHandler', new DeptLeaderHandler())
engine.setRegistry(reg)

2.3 流程定义引用

{
  "id": "task1",
  "type": "snaker:task",
  "properties": { "assignmentHandler": "deptLeaderHandler" }
}

引擎解析顺序：assignee 优先；为空时查 Registry 的 assignmentHandler。

3. 动态决策

场景：分支条件需要查库（如"该客户是否黑名单"），表达式写不了。

public class RiskDecisionHandler implements IDecisionHandler {
    @Override
    public String decide(ProcessInstance instance, Dict vars) {
        String customerId = vars.getStr("customerId");
        return riskService.isBlack(customerId) ? "edge_black" : "edge_normal";
    }
}
// 注册 + 节点 properties.decisionHandler = "riskDecisionHandler"

返回值为目标边的 id（edges[].id），引擎据此跳转。

4. 拦截器

场景：审批节点前置校验——"客户未实名认证则不给组长建任务"。

public class RealnameInterceptor implements FlowInterceptor {
    @Override
    public boolean preHandle(NodeModel node, ProcessInstance instance) {
        if ("verify".equals(node.getName())) {
            return "1".equals(instance.getVariables().get("realnameVerified"));
        }
        return true;
    }
    @Override public void postHandle(NodeModel node, ProcessInstance instance) {}
    @Override public int getOrder() { return 10; }
}

规则：

• preHandle 返回 false → 跳过该节点（不创建任务），流程暂停
• order 越小越先执行；post 按倒序（洋葱模型）
• 拦截器在节点级生效（可配置到流程级，见流程定义 preInterceptors）

5. 事件监听

场景：任务完成发钉钉通知。

async def on_event(evt):
    if evt.type == EventType.TASK_COMPLETE:
        await dingtalk.send(f"任务完成：{evt.taskName}（实例 {evt.instanceId}）")
    elif evt.type == EventType.PROCESS_FINISH:
        await dingtalk.send(f"流程完成：{evt.instanceId}")

engine.ext.event_listener = on_event   # Python
# engine.setExtensions({ listeners: [onEvent] })   # Node
# engine.SetExtensions(&engine.Extensions{Listeners: [...]})  # Go

事件清单：PROCESS_START / PROCESS_FINISH / PROCESS_REJECT / TASK_COMPLETE

注意：事件只带 instanceId/taskId，需要业务数据时自己查仓储（刻意设计，见设计原理 04）。

6. 四版源码位置

扩展	Java	Go	Node.js	Python
拦截器接口	interceptor/FlowInterceptor.java	engine/interceptor.go	src/extensions.ts	jeeflow/extensions.py
事件	event/ProcessEventListener.java	engine/interceptor.go	src/extensions.ts	jeeflow/extensions.py
参与者处理器	interceptor/AssignmentHandler.java	engine/registry.go	src/registry.ts	jeeflow/extensions.py
决策处理器	interceptor/DecisionHandler.java	engine/registry.go	src/registry.ts	jeeflow/extensions.py
注册表	ServiceContext	engine/registry.go	src/registry.ts	jeeflow/extensions.py