简介面向金融实时风控与大数据流计算学习者这份完整项目包以Storm与Esper架构为核心研发证券异常交易行为监控系统重点支持金证交易系统指令的拦截、复制与解析。压缩包共45个文件包含24个Java源码文件、8张运行截图、2个Excel表格如epl_opensource.xlsx、properties/XML配置文件、SQL脚本及使用说明书整包仅3.02MB分类清晰、便于按需查阅。当前已有75人学习下载项目源码已通过测试并可正常运行且曾获评审导师认可、答辩分达95适合计算机、人工智能、物联网等专业学生用于毕业设计、课程设计或初期项目演示。配套文档和授权码可以帮助快速理解流式计算与CEP规则在证券风控中的实践既能直接部署运行也能够在此基础上扩展新功能是掌握Storm/Esper架构应用的良好素材。1. 把规则引擎放进 Storm而不是写进交易接口一个常见的反直觉结论是异常交易监控系统最大的风险不是规则漏报而是你为了加规则把交易接口的延迟打上去了。很多人接到“行为监控”需求后第一反应是在金证柜台程序里加 if-else结果每次升级都要交易系统配合测试一次违规判断的 Full GC 就能让报单链路抖动几百毫秒。标题里的“基于 StormEsper 架构”给出的是一条被反复验证过的路线由 Storm 负责指令流的分发、容错和并行计算由 Esper 在内存里完成规则引擎计算两者合起来与交易系统做旁路对接。金证交易系统的指令拦截、复制与解析是这个链路的数据入口它决定了你看到的是“事后账本”还是“盘中信号”。这篇文章按“拦截输入 → 规则计算 → 流式拓扑 → 回放自检”的顺序把每个环节的关键参数和踩坑点讲清楚。2. 金证交易系统的指令拦截、复制与解析三个必须分清的落点2.1 先定位“指令”在交易链路里的位置在证券交易系统里一条“指令”通常指从柜台客户端、量化终端或周边系统提交给金证核心的委托、撤单、查询请求。要拦截它不能只盯着数据库里的 order 表因为 DB 记录是在指令生效之后才写入的监控在盘中拦截要的是“生效前”的数据。金证交易系统一般提供柜台中间件、内存对象或数据库适配器指令在网络上会经过一个明确的报盘接入节点。要在不影响主流程的前提下拿到指令常见做法是把监控做成独立进程只读取金证的指令流不反写任何字段。这个环节需要提前定三件事。第一拦截点决定了数据完整性如果只拿委托不拿撤单频繁撤单类规则全部失效。第二复制指令的通道必须有持久化能力否则监控进程重启会丢一段数据盘中告警就出现盲区。第三解析器必须容错不能因为一条字段缺失就把整个进程拖死。把这三件事写进设计文档比先写规则更有优先级。2.2 旁路镜像与主动 Hook两种复制方式怎么选指令复制的实现方式大致分两类复制方式接入难度对交易系统影响典型场景主要风险旁路镜像低无侵入金证已开放消息中间件或日志总线总线积压、消息顺序被打乱主动 Hook中有侵入需要在交易接口统一入口做控制回调超时、异常传染到主链路在已上线的生产环境里我更倾向旁路镜像。金证的接口调用链很长主动 Hook 如果包在事务里回调里的一次 GC 或一次网络超时都可能变成交易延迟。如果项目要求必须主动 Hook那就用单独的线程池把事件抛出去并且用信号量做限流。下面是一个典型的拦截器写法// 指令拦截器只负责把原始指令复制到监控通道不做规则判断 public class OrderCommandInterceptor { private final BlockingQueueCommandEvent queue; private final Semaphore semaphore new Semaphore(2000); public boolean intercept(CommandContext ctx) { // 只处理委托和撤单指令查询类指令不进监控通道 if (!ORDER.equals(ctx.getCommandType()) !CANCEL.equals(ctx.getCommandType())) { return true; } if (!semaphore.tryAcquire()) { // 监控通道阻塞过快直接放行保证交易链路不受影响 return true; } try { byte[] snapshot ctx.getRawMessage().toByteArray(); CommandEvent event new CommandEvent(ctx.getRequestId(), snapshot); return queue.offer(event, 10, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return true; } finally { semaphore.release(); } } }这里的核心逻辑是“失败放行”。Semaphore(2000)限制并发拷贝数防止交易高峰期监控线程把 CPU 占满queue.offer(event, 10, MILLISECONDS)让入队操作最多阻塞 10 毫秒超时后不再往里写。这样做会让监控侧丢失突发流量但换来的是交易主链路的确定性。后续的 Storm 消费端要做的是容忍少量缺口并通过交易所回报数据补全而不是要求拦截器保证 100% 不丢。2.3 解析指令时保留三个字段组解析指令不是简单 toString而是把二进制报文、业务键和风险计算字段拆开。我一般会保留三个字段组定位字段、行为字段和原始报文。业务键包括资金账号、证券代码、席位或营业部编码这是后续做按账户分流的依据。行为字段包括委托价格、委托数量、买卖方向、撤单标志、交易所上报时间这是 Esper 规则计算要用的。原始报文一定要留否则规则版本升级后无法复盘。public ParsedCommand parse(byte[] raw) { // 这是简化后的解析示例真实金证报文通常是 SJTX 或私有协议 Header header extractHeader(raw); ParsedCommand cmd new ParsedCommand(); cmd.setRequestId(header.getRequestId()); cmd.setAccountNo(header.getAccountNo()); cmd.setExchangeId(header.getExchangeId()); cmd.setSecurityCode(header.getSecurityCode()); cmd.setPrice(header.getPrice()); cmd.setQuantity(header.getQuantity()); cmd.setSide(header.getSide()); cmd.setRawBytes(raw.length 4096 ? Arrays.copyOf(raw, 4096) : raw); return cmd; }原始报文的长度要设置上限。真实环境中一条异常报文可能被错误拼接得很大如果不截断消息队列和 Storm 传输都会出现不可控内存消耗。截断后配合报文 MD5 或原始文件位置就可以在复核时还原完整上下文。2.4 复制通道的分区键与顺序问题监控规则里大量存在“同一账户在 60 秒内撤单 5 次”这类聚合计算。要保证这类计算准确同一个资金账号的指令必须进入同一个分区并按序消费。用 Kafka 时topic 的分区键常常被误设为证券代码或席位号这会导致同一账户的撤单和报单分散到不同分区Esper 看到的事件乱序误报率大幅上升。正确做法是分区键一律取资金账号。// Kafka 生产者按资金账号做分区键保证同一账户指令顺序消费 private final KafkaProducerString, byte[] producer; public void dispatch(ParsedCommand cmd) { String key cmd.getAccountNo(); ProducerRecordString, byte[] record new ProducerRecord( order-command-flow, key, cmd.getRawBytes()); producer.send(record, (metadata, exception) - { if (exception ! null) { // 这里要记录指标和异常请求号不能静默 metrics.counter(dispatch_failed).inc(); metrics.set(dispatch_failed_request_id, cmd.getRequestId()); } }); }如果金证系统本身提供了有序消息通道就不需要再绕道 Kafka但一旦监控侧需要做历史回放持久化队列就会成为刚需。我的经验是先确认金证中间件是否支持按业务键路由如果支持直接使用如果不支持自己构建 Kafka 通道时要在生产端就做足分区策略不要等拓扑里再调整。3. Esper 规则引擎从事件流到告警的三种典型 EPL 写法3.1 为什么在 Bolt 里维护 Esper 而不是用独立服务Esper 是一个 Java 库不是独立中间件。把 Esper 实例嵌入 Storm 的 Bolt 里最大的好处是状态跟随 Bolt。某个资金账号的指令经过 fieldsGrouping 路由到同一 Bolt 后Esper 的滑动窗口就在这个 Bolt 的 JVM 内完成不需要跨网络调用。另一种做法是单独部署一组 Espr 服务通过 TCP 接收事件这虽然把规则引擎和流计算解耦了但每次事件多一次网络往返延迟增加几毫秒而且状态清理和负载均衡都很麻烦。对于证券异常交易监控这种毫秒级场景我建议把 Esper 放进 Bolt。先看一个最小初始化例子// Esper 引擎初始化每个 Bolt 实例持有一个 provider EPServiceProvider provider EPServiceProviderManager.getDefaultProvider(); EPAdministrator admin provider.getEPAdministrator(); EPRuntime runtime provider.getEPRuntime(); admin.getConfiguration().addEventType(CommandEvent, CommandEvent.class.getName()); EPStatement frequentCancel admin.createEPL( select accountNo, count(*) as cancelCnt from CommandEvent(sidecancel).win:time(60 sec) group by accountNo having count(*) 5 ); frequentCancel.addListener((newEvents, oldEvents) - { // newEvents[0].get(accountNo)可在这里发送告警 });addEventType把 Java 类映射为 Esper 的事件类型类的 getter 就是事件属性。win:time(60 sec)表示一个 60 秒的滑动窗口窗口内满足having条件时触发。这里有一个新手容易犯的错Esper 的count(*)是窗口内当前累计值oldEvents只有在窗口滑动出旧事件时才会触发如果你监听oldEvents做撤销逻辑要明白它代表的是“过期事件”不是规则触发。3.2 频繁撤单、大额报单后撤单、自买自卖的 EPL 写法第一类规则是频繁撤单这是异常交易监控的标配。它的 EPL 写法如下select accountNo, count(*) as cancelCnt from CommandEvent.win:time(60 sec) where commandType CANCEL group by accountNo having count(*) 5这里的时间窗口和触发阈值上篇文章里已经在拦截器里讲过了。将频率阈值设置为 5 次/分钟比较适合容忍测试阶段实际上线要结合参与人身份区分机构账户和散户账户的阈值应该独立配置否则机构高频策略会被打爆。第二类规则是大额报单后快速撤单这类行为在很多市场里被定义为异常报撤单select o.accountNo, o.securityCode, o.quantity, c.cancelTime from pattern [ every oCommandEvent(commandTypeORDER and quantity 1000000) - (cCommandEvent(commandTypeCANCEL and accountNoo.accountNo and securityCodeo.securityCode) where timer:within(10 sec)) ]pattern是 Esper 里的事件组合语法every表示每一次匹配-表示事件顺序timer:within(10 sec)限定后一个事件必须在前一个事件后的 10 秒内出现。这种规则的典型问题是大额阈值写在 EPL 里规则改一次要重新编译整条 statement。第三类规则是自买自卖它不能只靠指令流判定因为真正的自买自卖需要成交回报来确认对手方。指令流里能做的是捕捉“同一账户在短时间内对同一证券先买后卖且价格相同”的嫌疑信号select a.accountNo, a.securityCode from pattern [ every aCommandEvent(sideBUY) - bCommandEvent(sideSELL and accountNoa.accountNo and securityCodea.securityCode and pricea.price) where timer:within(30 sec) ]这条规则产生的是“待确认线索”不是最终违规判定。真实的监控系统会把这类告警转成事件通过订单编号到成交库里去关联撮合结果。如果你把成交回报和指令流都接进 Esper还可以用 join 语法直接完成复杂度会上升很多但误判率更低。3.3 阈值参数化与规则热加载生产环境里规则阈值会频繁调整。把阈值用String.format拼进 EPL再用规则配置表驱动是最直接的做法。public void reloadRules(MapString, Object config) { // 先停掉旧 statement避免同一个规则窗口被重复计算 EPStatement old statements.remove(bigOrderCancel); if (old ! null) old.destroy(); double threshold (double) config.get(bigOrderThreshold); String epl String.format( select accountNo, count(*) as hitCnt from CommandEvent where quantity %f and commandTypeORDER group by accountNo having count(*) 1, threshold); EPStatement stmt admin.createEPL(epl); statements.put(bigOrderCancel, stmt); }注意String.format拼入的threshold必须是经过类型校验的数值。EPL 不像 SQL 那样有成熟的预编译参数绑定所以配置中心传来的值要先做白名单校验防止有人把%f位置替换成恶意 EPL。另外Esper 的 statement 是有生命周期的创建新 statement 前必须destroy()旧 statement否则同名规则会重复计算同一告警会触发两遍。4. Storm 拓扑并行度、分区键与背压的落地参数4.1 指令流接入KafkaSpout 的提交时机金证指令复制到 Kafka 之后Storm 侧的入口一般是 KafkaSpout。它的提交时机决定了故障恢复时会有多少重复数据。监控系统里重复告警比漏告警容易处理所以我采用AT_LEAST_ONCE语义。KafkaSpoutConfigString, byte[] kafkaConfig KafkaSpoutConfig .builder(host:9092, order-command-flow) .setProp(ConsumerConfig.GROUP_ID_CONFIG, abnormal-trading-monitor) .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST) .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE) .build();EARLIEST表示从最早偏移量开始读适合回放场景如果只在盘中监控从头消费这个配置也没问题。真实上线时我更建议用LATEST加独立回放链路因为EARLIEST在系统重启后会从头消费大量历史数据把 Kafka 和 Storm 都拖慢。回放链路可以复用同一个拓扑只是消费起点不同。4.2 用 fieldsGrouping 保证同一账号的指令落到同一个 Esper 引擎Storm 的 grouping 策略决定了事件如何分发到 Bolt。规则计算要求同一账号的指令顺序进入同一个 Esper 引擎否则窗口聚合会分散到多个实例各算各的阈值就失效了。TopologyBuilder builder new TopologyBuilder(); builder.setSpout(commandSpout, new KafkaSpout(kafkaConfig), 8); builder.setBolt(parseBolt, new CommandParseBolt(), 16) .shuffleGrouping(commandSpout); builder.setBolt(esperBolt, new EsperRuleBolt(ruleConfig), 16) .fieldsGrouping(parseBolt, new Fields(accountNo)); builder.setBolt(alertBolt, new AlertDispatchBolt(), 4) .shuffleGrouping(esperBolt);parseBolt用shuffleGrouping是为了让解析负载均匀分散。解析本身不依赖账户上下文所以不需要分区。esperBolt用fieldsGrouping按accountNo把同账户事件发往固定实例这样每个 Bolt 里的 Esper 只负责一部分账户窗口状态不会跨实例漂移。alertBolt没有聚合需求用shuffleGrouping减小线程热点。如果规则是按“席位证券代码”分组而不是按账户分组那 fieldsGrouping 的字段要改成exchangeId securityCode。这里不能直接 copy 我的配置而是先列一个表每类规则按什么键分区、会造成哪些账户/席位跨节点、误报影响是什么然后选分区键。4.3 三个关键参数max.spout.pending、acker 并发与传输 buffer背压是 Storm 应用最常见的稳定性问题。Esper 的滑动窗口在事件量大时会占用不少 RTI实时索引内存如果 Bolt 处理不过来链路就会积压。调优时先动下面三个参数参数建议值范围说明topology.max.spout.pending1000-5000限制 Spout 发出的未确认消息数形成自然背压topology.acker.executorsspout 并发数的一半过高会浪费线程过低会导致 Spout 频繁超时重发topology.transfer.buffer.size1024增大提高吞吐减小降低延迟不能两全当esperBolt的实例处理不过来时它会通过 ack 机制延迟确认进而让 Spout 的未确认消息数达到max.spout.pending最终抑制 KafkaSpout 拉取速度。这个背压链路是 Storm 自带的不需要额外开发。很多团队在遇到延迟时第一反应是加并发但并发加了以后如果分区数不变多出来的 Bolt 实例也不会收到数据。正确顺序是先看 Kafka topic 分区数是否大于 Storm 并发数再调max.spout.pending观察延迟最后才考虑要不要调整并行度。4.4 重启后状态恢复让窗口比进程活得长Esper 的窗口状态默认在 JVM 内存里Storm worker 一旦重启状态就没了。对于 60 秒的窗口重启后前 60 秒的规则会失效。常用做法是在 Bolt 的prepare方法里加载最近一段时间的指令做预热。Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { initEsperEngine(); // 回放最近 5 分钟的指令让窗口先填满再对外告警 replayRecentCommands(Duration.ofMinutes(5)); }这个预热只是让窗口有上下文不能保证精确恢复。因为回放的指令是直接注入 Esper runtime 的不会重复告警注意如果你在prepare里触发监听器它会把预热数据当成新事件发送告警。所以预热入口要和正常入口区分给事件打上isReplaytrue标记监听器侧跳过告警发送。5. 落库、回放校验与运维自检5.1 用“请求号规则版本号”做告警去重告警结果写库或者输出到审计日志时不能只存时间、账户和违规类型。相同指令可能同时命中多条规则同一指令流回放两次还会再次生成告警。我会在告警表里以请求号和规则版本号作为唯一键。CREATE TABLE abnormal_alerts ( request_id VARCHAR(64) NOT NULL, rule_version VARCHAR(32) NOT NULL, account_no VARCHAR(32) NOT NULL, security_code VARCHAR(16) NOT NULL, hit_reason VARCHAR(512) NOT NULL, trigger_time TIMESTAMP(3) NOT NULL, raw_msg_path VARCHAR(256), PRIMARY KEY (request_id, rule_version) );写入使用INSERT ... ON DUPLICATE KEY UPDATE或 Kafka 幂等生产者。这样即使拓扑重启后发生重复处理也不会在库里产生两条相同告警。原始报文路径字段用来关联金证侧的日志文件方便审计追溯。5.2 把历史交易日数据倒灌回 Esper 做回归规则引擎最怕的不是写不出规则而是改了一条规则后不知道影响了多少存量结果。我的常规做法是把某个历史交易日的指令流按原始时间戳重新注入 Kafka然后跑同一套 Storm 拓扑。# 取某天已落盘的指令流按账号分区重放到回放topic for f in /data/replay/20240103/*.avro; do java -jar replay-producer.jar --topic order-command-flow-replay --file $f done # 对比基线库与新回放库的告警差异 java -jar expect-diff.jar --baseline alert_db.20240103 --actual alert_db.replay.20240103回放和正常交易共用一个拓扑时需要把 topic 名和消费组区分开避免回放数据混入生产链路。回放结果对比后重点关注“新增告警”和“消失告警”。新增告警可能是因为规则阈值放宽消失告警可能是因为规则条件收紧两条都需要人工判断。Esper 的窗口状态是冷启动的回放数据的前一个窗口周期内没有历史事件所以对比时要过滤掉前 60 秒或窗口时长的输出。5.3 延延迟与漏报率的观测方法我在生产环境里最常用的两个指标是esper.latency_ms和rule.hit_count。前者是事件进入 Esper 到监听器触发的时间差后者是每条规则的命中次数。延迟突然升高时先看 worker 的 GC 日志Esper 的 RTI 索引会对顺序访问做缓存Full GC 后缓存失效会导致瞬时 CPU 走高。规则命中率长期为 0 不一定是好事先查规则条件里的字段名是否和解析出的类属性一致比如quantity在类里是Integer而金证报文里可能是带负数的净买卖量规则永远不满足。监控系统的最终价值是辅助审计不是替代人工判断。把规则触发的上下文、请求号、回放结果三者绑定在一起才能在事后复核时说得清楚。这就是这套基于 StormEsper 架构的监控系统在金证环境下应该有的收尾能力。本文还有配套的精品资源点击获取