事件处理业务的简易组件编排框架

背景

很多入侵事件类型的处理流程非常相似，只有少部分的差异。 按目前的方式，每种入侵事件类型都要开发一遍，很多相似的模板代码，可复用性和可维护性都一般。这种情况下，可以采用业务编排的方式，来提升事件处理的可复用性和可维护性。

业务的组件编排，意味着需要将业务逻辑抽象成组件，然后将组件编排成业务流程。 基于组件编程思想。

业务的组件编排和工作流引擎有点类似。但我们的目标不是实现类似审批、采购这样的流程，而是为了让业务逻辑能够更加可复用和可扩展，提升研发效率。因此，最终目标不是实现一个工作流引擎，而是实现一个相对灵活的业务组件编排和执行框架。

目标

• 将通用和差异分离，最大程度复用已有逻辑，降低新入侵事件处理的开发量，提升其研发效率；
• 业务逻辑组件化，改动只局限在新增的组件，遵循开闭原则，提升软件可维护性。
  
  

适用场景

• 具有相似性的模板流程的事件处理。
  
  

设计思路

要设计和实现一个事件处理的业务组件编排，需要解决如下问题：

• 定义一个事件处理流程的基本概念集： 事件数据(EventData)、基于事件数据的事件处理上下文语境( FlowContext) 及构建者(FlowContextBuilder)、事件处理流程定义(EventFlow)、事件处理组件声明和定义（ ComponentProperties 和 FlowComponent）；

• 定义事件处理流程的基本模板（或者可以自定义）；

• 定义具体的业务事件处理的流程所需要的组件，并能根据指定业务类型来选择合适的流程；

• 定义具体的业务组件，并能根据组件的功能和类型来选择合适的组件来执行。

整体框图如下：

本文源代码在 “ALLIN” 下的 cc.lovesq.flows 包下。

基本概念

事件数据 IEventData

• 表示任何事件数据；不同的事件处理可以定义不同的具体类 EventData。

public interface EventData {

    /**
     * 获取事件数据
     */
    String getData();

    /**
     * 获取事件类型
     */
    String getType();
}

事件处理语境 FlowContext 和 FlowContextBuilder

• 基于 EventData 构建，用于处理事件过程中的上下文语境。

public abstract class AbstractFlowContext<E extends EventData> {

    /** 事件对象 */
    protected E eventData;

    /** 事件对象中所含的业务 DO 对象 */
    protected DetectDO detectDO;

    /** 事件对象中所含的业务 DTO 对象 */
    protected DetectDTO detectDTO;

    /** 扩展信息 */
    private Map<String, Holder> extendInfo;

    public AbstractFlowContext() {
    }

    public AbstractFlowContext(E eventData, DetectDO detectDO, DetectDTO detectDTO) {
        this.eventData = eventData;
        this.detectDO = detectDO;
        this.detectDTO = detectDTO;
    }

    public E getEventData() {
        return eventData;
    }

    public String getData() {
        return eventData.getData();
    }

    public DetectDO getDetectDO() {
        return detectDO;
    }

    public DetectDTO getDetectDTO() {
        return detectDTO;
    }

    public <T> void put(String name, Holder<T> holder) {
        extendInfo.put(name, holder);
    }

    public <T> T get(String name, Class<T> cls) {
        return (T)extendInfo.get(name).getData();
    }


}

public interface FlowContextBuilder<E extends EventData> {

    /**
     * 根据事件数据构建事件处理的上下文语境
     */
    AbstractFlowContext build(E eventData);
}

public class DefaultDetectFlowBuilder implements FlowContextBuilder<DetectEventDataWrapper> {

    @Override
    public AbstractFlowContext build(DetectEventDataWrapper eventData) {
        String detectType = eventData.getDetectType();

        Class<DetectDO> doCls = DetectEventEnum.getDoClass(detectType);
        DetectDO detectDO = parse(eventData.getData(), doCls);

        Class<DetectDTO> dtoCls = DetectEventEnum.getDtoClass(detectType);
        DetectDTO detectDTO = parse(eventData.getData(), dtoCls);

        return new DefaultDetectFlowContext(eventData, detectDO, detectDTO);
    }

    private <T> T parse(String text, Class<T> cls) {
        JSONObject jsonObject = JSONObject.fromObject(text);
        return (T)JSONObject.toBean(jsonObject, cls);
    }
}

事件处理流程定义EventFlow

• 定义处理事件数据的处理流程模板，这个流程是通过执行多个流程组件来实现的。

public interface EventFlow<E extends EventData> {

    /**
     * 处理事件的流程
     */
    void process(E eventData);
}

流程组件FlowComponent

• 在事件处理流程中处理业务逻辑的业务组件；

• 组件包含组件意图和组件业务类型，作为选择组件的依据；

• 统一用 AbstractFlowContext 作为参数；

• FlowComponent 的执行结果使用 FlowResult 来表示；FlowResult 包含一个可能的数据值，以及必要的一个枚举 FlowDecision 决定继续或结束流程。

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ComponentProperties {

    /** 组件的意图或功能 */
    String purpose();

    /** 组件的业务类型选择 */
    String biz() default "common";

}

/**
 * 事件处理流程中的一个业务组件
 */
public interface FlowComponent<I extends AbstractFlowContext> {
    FlowResult process(I flowContext);

    // 满足此条件才能执行该组件，默认可以执行

    default boolean needAccess(I flowContext) {
        return true;
    }
}

定义一个具体组件：

public interface BigDataSender<I extends AbstractFlowContext> extends FlowComponent<I> {
    void sendEventMessage(I context);

    default FlowResult process(I context) {
        sendEventMessage(context);
        return new FlowResult(null, FlowDecision.Continue);
    }
}

@ComponentProperties(purpose = "BigDataSender")
@Component
public class DefaultDetectBigDataSender implements BigDataSender<DefaultDetectFlowContext> {

    private static Log log = LogFactory.getLog(DefaultDetectBigDataSender.class);

    @Override
    public void sendEventMessage(DefaultDetectFlowContext context) {
        String bigDataType = DetectEventEnum.getBigDataType(context.getEventData().getDetectType());
        String info = String.format("send: bigDataType=%s, msg=%s", bigDataType, JSON.toJSONString(context.getEventData().getData()));
        log.info(info);
    }
}

事件处理流程

事件处理流程的基本模板

STEP1：对于某一类事件，初始化 FlowContextBuilder flowContextBuilder 和 FlowComponents ;

STEP2：使用 flowContextBuilder 将 EventData 构建成 FlowContext ;

STEP3： 循环或并发调用 FlowComponent.process 进行 事件处理 。

public abstract class AbstractEventFlow<E extends EventData> implements EventFlow<E> {

    @Resource
    protected FlowComponentFactory flowComponentFactory;
    @Resource
    protected ComponentExecutor componentExecutor;

    public AbstractEventFlow(FlowComponentFactory flowComponentFactory) {
        this.flowComponentFactory = flowComponentFactory;
    }

    @Override
    public void process(E eventData) {

        AbstractFlowContext flowContext = builder().build(eventData);
        List<FlowComponent> flowComponents = flowComponents(eventData.getType());
        componentExecutor.execMayExit(flowContext, flowComponents);
    }

    /** 可以覆写此方法来构建自己的 FlowContext */
    abstract public FlowContextBuilder builder();

    /** 可以覆写此方法来指明流程所需要的业务处理组件 */
    abstract public List<FlowComponent> flowComponents(String detectType);
}

@Component
public class DefaultDetectEventFlower extends AbstractEventFlow<DetectEventDataWrapper> {

    private FlowContextBuilder flowContextBuilder = new DefaultDetectFlowBuilder();

    @Autowired
    public DefaultDetectEventFlower(FlowComponentFactory flowComponentFactory) {
        super(flowComponentFactory);
    }

    @Override
    public FlowContextBuilder builder() {
        return flowContextBuilder;
    }

    @Override
    public List<FlowComponent> flowComponents(String detectType) {
        return flowComponentFactory.getComponents(DetectEventFlowDefinitions.getNeedComponents(detectType));
    }
}

业务的事件处理流程定义和选择

public enum DetectEventFlowDefinitions {

    bounce_detect_flow(BOUNCE_SHELL.getType(),
            Arrays.asList(DefaultNotifySender, DefaultBigDataSender, DefaultThreatInfoSender)),

    ;

    /** 入侵事件业务类型 */
    String detectType;

    /** 入侵事件类型 */
    BizEventTypeEnum bizEventTypeEnum;

    /** 入侵事件处理流程的全限定类名 */
    String eventClassName;

    /** 事件处理所需要的组件 */
    List<ComponentType> componentTypes;

    DetectEventFlowDefinitions(String detectType, List<ComponentType> componentTypes) {
        this(detectType, BizEventTypeEnum.CREATE, DefaultDetectEventFlower.class.getName(), componentTypes);
    }

    DetectEventFlowDefinitions(String detectType, String eventClassName, List<ComponentType> componentTypes) {
        this(detectType, BizEventTypeEnum.CREATE, eventClassName, componentTypes);
    }

    DetectEventFlowDefinitions(String detectType, BizEventTypeEnum bizEventTypeEnum, String eventClassName, List<ComponentType> componentTypes) {
        this.detectType = detectType;
        this.bizEventTypeEnum = bizEventTypeEnum;
        this.eventClassName = eventClassName;
        this.componentTypes = componentTypes;
    }

    private static Map<String, DetectEventFlowDefinitions> flowDefinitionsMap = new HashMap<>();
    private static Set<String> flows = new HashSet<>();

    static {
        for (DetectEventFlowDefinitions detectEventFlowDefinitions : DetectEventFlowDefinitions.values()) {
            flowDefinitionsMap.put(getKey(detectEventFlowDefinitions.detectType, detectEventFlowDefinitions.bizEventTypeEnum.getType()), detectEventFlowDefinitions);
            flows.add(detectEventFlowDefinitions.detectType);
        }
    }

    public static String getKey(String detectModelType, String eventType) {
        return detectModelType + "_" + eventType;
    }

    public static String getEventFlowClassName(String detectModelType, String eventType) {
        return flowDefinitionsMap.get(getKey(detectModelType, eventType)).eventClassName;
    }

    public static List<ComponentType> getNeedComponents(String detectType) {
        return flowDefinitionsMap.get(getKey(detectType, BizEventTypeEnum.CREATE.getType())).componentTypes;
    }

    public static List<ComponentType> getNeedComponents(String detectType, String bizEventType) {
        if (StringUtils.isBlank(bizEventType)) {
            bizEventType = BizEventTypeEnum.CREATE.getType();
        }
        return flowDefinitionsMap.get(getKey(detectType, bizEventType)).componentTypes;
    }
}

使用工厂模式来选择和获取具体的某个业务事件处理流程：

public abstract class AbstractEventFlowFactory {

    private final Logger logger = LoggerFactory.getLogger(AbstractComponentFactory.class);

    protected Map<String, EventFlow> eventFlowMap = new HashMap<>();

    private volatile boolean initialized = false;

    /*
     *  初始化事件处理流程Beans
     */
    protected void initBeans() {
        if (!initialized) {
            Map<String, EventFlow> eventFlowBeans = getEventFlowBeans();
            for (EventFlow eventFlow: eventFlowBeans.values()) {
                eventFlowMap.put(eventFlow.getClass().getName(), eventFlow);
            }
            logger.info("init-success: eventFlowMap: {}", eventFlowMap);
            initialized = true;
        }
    }

    /** 返回的事件处理流程 bean 映射 Map[BeanName,IEventFlowBean] */
    abstract public Map<String, EventFlow> getEventFlowBeans();

    /**
     * 根据指定的事件处理流程全限定性类名获取指定的 Bean 实例
     */
    public EventFlow get(String eventFlowClassName) {
        return eventFlowMap.get(eventFlowClassName);
    }

    public EventFlow getEventFlow(String bizEventType, String detectType) {
        String eventFlowClassName = DetectEventFlowDefinitions.getEventFlowClassName(detectType, bizEventType);
        return eventFlowMap.get(eventFlowClassName);
    }
}

@Component
public class DetectEventFlowFactory extends AbstractEventFlowFactory implements ApplicationContextAware {

    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @PostConstruct
    public void init() {
        initBeans();
    }

    @Override
    public Map<String, EventFlow> getEventFlowBeans() {
        return applicationContext.getBeansOfType(EventFlow.class);
    }
}

组件工厂与定位

根据组件意图和组件业务类型，使用工厂模式来选择和获取某个具体的组件：

/**
 * 可定制的组件工厂类
 */
public abstract class AbstractComponentFactory {

    private static Logger logger = LoggerFactory.getLogger(AbstractComponentFactory.class);

    Map<String, FlowComponent> componentBizMap = new HashMap<>();
    Map<String, FlowComponent> componentClassMap = new HashMap<>();

    private volatile boolean initialized = false;

    /*
     * 初始化组件Beans
     */
    protected void initBeans() {
        if (!initialized) {
            Map<String, FlowComponent> beansOfType = getComponentBeans();
            beansOfType.values().forEach(
                    component -> {
                        ComponentProperties annotations = component.getClass().getAnnotation(ComponentProperties.class);
                        if (annotations != null) {
                            componentBizMap.put(ComponentType.getKey(annotations.purpose(), annotations.biz()), component);
                        }
                        componentClassMap.put(component.getClass().getName(), component);
                    }
            );
            logger.info("init-success: ComponentBizMap: {}", componentBizMap);
            logger.info("init-success: componentClassMap: {}", componentClassMap);
            initialized = true;
        }
    }

    /**
     *  返回所有对应的组件实例映射 Map[BeanName, ComponentBean]
     */
    abstract public Map<String, FlowComponent> getComponentBeans();

    /**
     * 根据指定的组件全限定性类名来获取对应的组件实例
     * @param qualifiedClassName 组件全限定性类名
     * @return 组件实例
     */
    public FlowComponent getComponent(String qualifiedClassName) {
        return componentClassMap.get(qualifiedClassName);
    }

    /**
     * 根据指定的组件意图和业务类型来获取对应的组件实例
     * @param purpose 组件意图标识
     * @param biz 组件业务类型
     * @return 组件实例
     */
    public FlowComponent getComponent(String purpose, String biz) {
        return componentBizMap.get(ComponentType.getKey(purpose, biz));
    }

    /**
     * 根据指定的组件类型来获取对应的组件实例
     * @param componentType 组件类型
     * @return 组件实例
     */
    public FlowComponent getComponent(ComponentType componentType) {
        return componentBizMap.get(ComponentType.getKey(componentType.getPurpose(), componentType.getBiz()));
    }

    /**
     * 根据组件类型集合批量获取对应的组件实例集合
     * @param componentTypeList 组件类型集合
     * @return 组件实例集合
     */
    public List<FlowComponent> getComponents(List<ComponentType> componentTypeList) {
        return StreamUtil.map(componentTypeList, com -> getComponent(com.getPurpose(), com.getBiz()));
    }
}

@Component
public class FlowComponentFactory extends AbstractComponentFactory implements ApplicationContextAware {

    private static Log log = LogFactory.getLog(FlowComponentFactory.class);

    private ApplicationContext applicationContext;

    private volatile boolean initialized = false;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @PostConstruct
    public void init() {
        initBeans();
    }

    @Override
    public Map<String, FlowComponent> getComponentBeans() {
        return applicationContext.getBeansOfType(FlowComponent.class);
    }
}

组件执行器

@Component
public class ComponentExecutor {

    @Resource
    private FlowComponentFactory flowComponentFactory;

    // 如果要支持并发，这里可以有一个组件执行线程池

    /**
     * 执行指定流程组件集，若任一组件要结束流程，则流程结束
     */
    public <I extends AbstractFlowContext> FlowResult execMayExit(I flowContext, List<FlowComponent> flowComponents) {
        for (FlowComponent flowComp: flowComponents) {
            if (!flowComp.needAccess(flowContext)) {
                continue;
            }
            FlowResult flowResult = flowComp.process(flowContext);
            if (flowResult.getFlowDecision() == null || FlowDecision.Termination.equals(flowResult.getFlowDecision())) {
                return FlowResultHelper.terminateResult();
            }
        }
        return FlowResultHelper.continueResult();
    }

    /**
     * 单个普通业务组件执行
     */
    public <I extends AbstractFlowContext> FlowResult exec(I param, FlowComponent<I> c) {
        if (c.needAccess(param)) {
            return c.process(param);
        }
        return FlowResultHelper.continueResult();
    }

    public <I extends AbstractFlowContext> FlowResult exec(I context, String compClassName) {
        FlowComponent c = flowComponentFactory.getComponent(compClassName);
        return exec(context, c);
    }

    /**
     * 任意多个流程组件执行(提供更灵活的组件执行能力)
     * 任一组件不影响后续组件的执行(抛出异常除外)
     *
     * NOTE: 这些业务组件的参数是一样的
     */
    public <I extends AbstractFlowContext> FlowResult execBatch(I context, List<FlowComponent> components) {
        for (FlowComponent c: components) {
            if (c.needAccess(context)) {
                c.process(context);
            }
        }
        return FlowResultHelper.continueResult();
    }

}

设计说明

为什么要用泛型声明 E extends EventData 而不是直接用 EventData

使用泛型，则不同的业务可以定义不同的 EventData ， 覆写方法时，就可以用具体的 EventData 子类，而不是用 EventData （需要做类型转换）。 用泛型之后，可以处理不同类型的事件数据。

组件参数传参

有两种方式：

• 像 linux 管道那样， 上一个组件的输入作为下一个组件的输出， 这样缩小了变量和对象的作用域，能够及时回收，但是设计需要更加精巧，开发理解也不大容易；

• 使用一个 Context 或者 ThreadLocal ，简单粗暴，最大的缺点是这个类容易膨胀臃肿。 目前是先实现第二种。如果要防止膨胀，有一种办法，就是把一组相互有关联的变量组织成一个参数对象，通过参数对象去安全地获取。这样略有麻烦，但能更清楚地看到组件依赖的变量，要移除也更加容易一点。
  
  

提前结束流程

在事件处理流程的任何一步，根据新获得的信息，都可能做出判断，提前结束流程。每个组件执行之后，无论是否出错，既有可能需要结束流程，也有可能需要继续流程。此时，可以让 process 返回一个枚举值 FlowDecision ，这个枚举值来告诉主控流程，到底是结束还是继续。

更完善的组件流程

更完善的组件流程应当是：

• 一个业务流程通过若干的流程 STEP 来实现；流程 STEP 之间是有顺序的；

• 每个流程 STEP 内可能包含多个组件， 这些组件可能是相互可替换的；

• 业务编排系统应当对流程 STEP 和 特定 STEP 里的组件类型进行必要的检查。

组件执行顺序

组件执行顺序既可以用有序列表来表达，也可以用有序链表来表达。采用列表的方式，则流程的主控在 IEventFlow 的实现类里，采用链表方式表达，则流程的主控在组件里。

多个组件还可以组合成一个组件块一起执行（组合模式）；也可能在某种条件下才去执行某个组件或在组件内执行组件块。可以在 FlowComponent 里添加执行多个组件的默认执行方法 exec(FlowComponent... flowComponents)。 flowComponents 可以从组件工厂里通过组件的全限定性类名来拿到。

此外，互不依赖互不影响的不同组件块可以并发执行。 目前业务没有这种需求，限于时间因素，暂未实现。

事件处理流程配置

事件处理流程的组件配置目前是采用枚举方式，写死在枚举类 DetectAgentEventFlowDefinitions 【前置部分】和 DetectBizEventFlowDefinitions 【后置部分】里了 。采用枚举方式直观可靠（写死在代码里），但是不够灵活。如果组件集需要灵活执行，比如 A, B, C, D, E 五个组件，执行顺序是： A, (B,C 并发), C, (D,E 并发)，就难以支持了。

可以将组件执行顺序（即事件处理流程配置）改为 JSON 或 YML 形式，更加灵活。只要编写一个组件执行器（支持串行或并发地执行组件集），就可以支持。目前入侵事件处理流程暂不需要这样的功能，限于时间因素，因此暂未实现，但可以留下设计空间。

采用枚举方式的事件处理流程配置，可以作为降级兜底策略。

可嵌入

如果要从全局来改造原有流程，可能重构成本比较大。比如 AcceptFlow 部分，从全局来改造，需要从 AgentSensor 开始。实际上 AgentSensor 和 Handler 的逻辑并不多，可以从 AcceptFlow 部分嵌入这个事件处理流程的业务编排，只改动 flow.execute 的那部分，能够降低重构成本（难以重构的或者暂时没精力重构的暂时保持现状不影响），同时对于确实有模板特征的事件处理流程能够利用上业务编排的优势。这要求，事件处理流程的业务编排框架是小型的可嵌入式的，能够灵活地使用上。

小结

本文主要给出了一个事件处理流程的简易业务编排框架，适用于具有相似的模板流程的事件处理。

设计一个小型框架，能够处理一类相似业务，更能体现性价比。比如做入侵检测业务，最开始打算将入侵业务都组件化，但这样重构成本非常高；因此，后面将目标缩小为：做一个小型业务编排系统，能够将具有相似性处理的入侵事件处理流程组件化和可编排化。做设计，不能一开始就想着弄一个大而全的精巧的框架。制定合理的设计目标，更有利于达成。

此外，设计的一大目标是使团队开发成员自然地遵循设计约束。初始设计既重要也不能看得太重。如果缺乏初始设计，整个系统很快就会腐化成一坨坨无章法的逻辑堆砌；而初始设计太精巧，随着更多的成员加入，团队成员不一定能很好地遵循初始设计约束，系统也会逐渐地腐化。

有一点经验小结下：

• 设计和实现组件的基本能力，首先要有一个基本的概念设计和接口设计（概念定义及概念关联，概念对应的接口及交互）；

• 以一个典型的业务场景为例，验证事件处理的组件编排框架的可行性；

• 以一个复杂的业务场景为例，验证事件处理的组件编排框架的完备性；

• 完善组件编排框架的配置能力；

• 多多听取团队成员的建议并改进，比如成亮提出包结构重新组织下，对做共享库就很有帮助；

• 确立合理的设计目标和设计范围，尽可能达到好的ROI。

做共享库的一点经验：

• 明确和聚焦事件处理的组件编排框架的核心能力： 组件及组件序列的配置、定位、执行；

• 共享库里的类尽量做成抽象的可定制的，减少具体实现的工作量（Javadoc中可以有示例注释）；

• 共享库里的代码注释要完善，代码要简洁，单测要完善；

• 去掉业务相关的，勿以业务相关的来充实共享库。