简介一套基于Java与Apache Storm构建的日志监控告警系统方案面向大数据实时计算、日志分析和运维监控方向的开发者解决日志数据实时消费、规则匹配与多渠道告警联动的典型问题。系统以Kafka作为日志接入层通过Storm拓扑完成从数据读取、规则加载、异常匹配到邮件/短信通知、数据库落地的完整链路。资源包共100个文件大小仅1.17MB以Java源码24个java与编译产物27个class为主体配合4个xml项目配置、3个iml模块文件另有39张png图片和2份md文档便于查看项目结构、运行界面及设计说明。核心实现涵盖Kafka数据源接入、定时规则加载Bolt、日志处理与匹配Bolt、邮件短信通知Bolt、数据库保存Bolt以及规则匹配、通知发送、JDBC等工具类可从中了解完整Storm拓扑的组装方式与告警处理流程。已有159人学习下载。对想动手实践Storm实时计算、理解Kafka对接及告警通知实现细节的读者可基于源码快速搭建自己的日志监控告警原型也可结合图片与文档梳理异常检测逻辑。1. 日志告警不是ELK专利Kafka脏日志到邮件短信的实时链路日志量每天几十万条想第一时间发现报错而不是等业务方找上门的时候大家优先想到的往往是ELK那一整套。但如果你已经在用Kafka团队技术栈又正好是纯JavaApache Storm这套轻量拓扑反而是更顺手的方案。这份基于Java与Storm的日志监控告警系统正好补齐了“Kafka Topic→规则匹配→邮件短信通知→告警落库”的完整链路。工程里包含TopologyMain与TopkeyTopologMain两个拓扑入口、StormTickBolt定时加载监控规则、ProcessDataBolt做规则匹配、NotifyMessageBolt与SaveToDBBolt分别负责通知分发和数据落库还有CommonUtils与JdbcUtils两类工具类兜底。它适合已有Kafka数据源、想用纯Java实时处理日志告警的团队也适合做Storm课程设计的同学直接照着复现。2. 系统架构与核心组件从Kafka Topic到通知下发的数据流先把整条数据流立起来Kafka Spout从指定Topic拉取日志tuple经fieldsGrouping按内容字段分发给ProcessDataBolt做规则匹配StormTickBolt是一个独立定时节点按固定频率从数据库加载监控规则和应用信息再以广播流把规则快照发给所有ProcessDataBolt实例命中的tuple分发给NotifyMessageBolt发送邮件和短信同时SaveToDBBolt把告警记录写入数据库没命中的日志直接ack掉不落库。整条拓扑只有5个节点中间没有窗口聚合节点按tuple逐条处理端到端延迟在秒级适合日志量中等、但对报警及时性有要求的场景。这套设计没有引入独立的规则引擎或消息中间件规则匹配就放在Bolt进程内完成通知直接走邮件和短信网关。这样做的取舍是架构简单、好排查代价是规则复杂度做不深正则和组合条件都靠CommonUtils里那几行逻辑扛。下面逐个拆节点看它是怎么工作的。2.1 Kafka Spout与TopologyMain日志数据进入拓扑的通道Spout是Storm拓扑的数据源头这个工程用的是storm-kafka客户端里的KafkaSpout而不是在Bolt里自己写KafkaConsumer轮询。为什么因为Storm里Spout的nextTuple()由框架反复调用如果用原生Consumer你得自己维护线程、拉取和位移提交而KafkaSpout把fetch、offset提交、失败重放全部封装好了。项目里TopologyMain就是负责装配这些组件的入口类。ZkHosts zkHosts new ZkHosts(zk1:2181,zk2:2181,zk3:2181); SpoutConfig spoutConfig new SpoutConfig( zkHosts, app_log_topic, // 消费的日志 Topic /log_monitor/kafka_offset, // offset 在 ZooKeeper 中的保存根路径 log_monitor_group); // consumer group id spoutConfig.scheme new SchemeAsMultiScheme(new StringScheme()); spoutConfig.forceFromStart false; // false 表示接续上次提交的 offset spoutConfig.zkRoot /log_monitor/kafka_offset; KafkaSpout kafkaSpout new KafkaSpout(spoutConfig); TopologyBuilder builder new TopologyBuilder(); builder.setSpout(kafka_spout, kafkaSpout, 2); // 并行度设为 2参数逐个看ZkHosts只传ZooKeeper地址就够了Kafka的broker元数据注册在ZK的/brokers/ids路径下KafkaSpout启动时会自动拉取集群broker列表后面Kafka扩节点时拓扑不用改配置第二个参数Topic名要和Kafka里建的Topic完全一致第三个参数是offset在ZK里的存储路径多个拓扑消费同一个Topic时这个路径不能一样第四个参数groupId相当于消费组标识两个拓扑用同一个groupId消费同一个Topic会把彼此offset挤掉。forceFromStart为false代表正常接续上次消费进度只有做数据重放测试时才改成true。Spout装配完成以后下游Bolt要关注tuple的可靠性。日志监控场景里普通日志丢一两次问题不大但告警丢了会被投诉所以ProcessDataBolt处理完必须调用collector.ack(input)中途异常就调fail(input)KafkaSpout会按重试机制重新拉取这段消息。这一点在第5章避坑里专门展开。2.2 StormTickBolttick频率与规则加载机制监控规则如果写死在代码里每次加关键字都要重新打包提交拓扑排障场景下几十分钟延迟完全不能忍。这个系统把规则放在数据库表里由StormTickBolt定时加载。定时触发靠的就是Storm的tick tuple给某个Bolt配置tick频率后框架会周期性往它的execute()里塞一条特殊tuplesourceComponent固定为__systemsourceStreamId固定为__tick。public void execute(Tuple input) { if (isTickTuple(input)) { // 到点刷新数据库里的规则、应用、用户信息 ListMonitorRule rules loadRulesFromDB(); // 把规则快照通过独立流广播给下游所有 process_bolt 实例 collector.emit(rule_stream, input, new Values(rules)); return; } // 非 tick 数据tick_bolt 本身不处理业务数据直接 ack collector.ack(input); } private boolean isTickTuple(Tuple input) { return __system.equals(input.getSourceComponent()) __tick.equals(input.getSourceStreamId()); }这段逻辑的重点是loadRulesFromDB()的实现方式。不能先clear再load那会留一个空缓存窗口ProcessDataBolt查不到规则导致告警漏报。我一般会新建一个List或HashMap查完库整体替换旧引用ProcessDataBolt拿到的永远是一份完整的规则快照。tick频率在提交拓扑时通过Config配置典型值是60秒一次也就是数据库改规则后最多延迟1分钟生效。如果要求秒级响应可以调到10秒但每次刷新都是一次全量查库规则表几万行以内无所谓再大就要考虑增量加载。2.3 告警通知链路邮件与短信分发ProcessDataBolt匹配到异常日志后把日志内容、命中规则ID、用户信息封装成告警tuple发给NotifyMessageBolt。这个Bolt职责很集中根据告警tuple里的通知方式调用邮件或短信工具把消息推出去。工程里MailInfo是邮件实体封装收件人、主题、正文、SMTP服务器、账号密码这些字段MessageSenderUtil封装了JavaMail的Transport发送逻辑ShortMessageUtil负责调短信网关的HTTP接口。这里有一个值得说的设计点邮件和短信发送都是阻塞IO。如果某个短信网关接口抖动响应三五秒NotifyMessageBolt整个线程就被卡住后续所有告警tuple排队等待上游Bolt也会因为背压导致消费缓慢。我实际踩到过这个问题后来的习惯是给这个Bolt塞一个固定大小的线程池发送请求丢给线程池执行Bolt本身只负责收tuple和提交发送任务再配合失败重试队列短信网关再慢也只是发送线程池排队不影响主链路的tuple消费。2.4 数据落库与工具类SaveToDBBolt和JdbcUtils的职责边界告警记录必须留底否则事后排查问题只能靠运气。SaveToDBBolt接收ProcessDataBolt的告警数据通过JdbcUtils拿到数据库连接执行INSERT把告警内容、匹配规则ID、通知时间和状态写进告警表。这里最容易翻车的点是数据库连接的获取方式。最忌讳的做法是每个tuple都新建一个Connection日志量一大数据库连接数直接被打满表现为数据库端大量报Too many connections。常规做法是在SaveToDBBolt的prepare()里初始化一个连接池或者用JdbcUtils里维护的池化连接Bolt生命周期内反复复用。CommonUtils在工程里是公共门面规则匹配判断、告警对象构造、通知调用入口都在它里面避免ProcessDataBolt和NotifyMessageBolt各写一套重复逻辑。它同时也是这几个Bolt之间的隐式约定层字段名改了只需要动一个类。数据流到这里就完整了KafkaSpout进通知落库出每一层的职责都不重叠。3. 核心代码拆解拓扑装配、Bolt实现与工具类实战架构层面的分工已经清楚接下来直接看代码实现。整个工程虽然类不多但分层是清楚的两个拓扑入口类负责装配Bolt类各自管一段处理逻辑工具类提供公共能力实体类做数据传输。我会按入口、Bolt、工具三个层次逐个拆每一步给关键代码和参数说明方便你直接对着改。3.1 TopologyMain与TopkeyTopologMain两个拓扑入口的差异TopologyMain是主告警拓扑入口装配的是“消费日志→规则匹配→通知→落库”这条链路。TopkeyTopologMain从命名看是TopKey统计拓扑的入口配合TopkeyCountBolt做日志关键词的频次统计它不直接发告警而是输出统计结果。这两个入口在工程里是独立的main方法提交时用storm jar分别指定主类即可它们跑在各自的Worker进程里互不影响。TopologyMain的装配顺序很关键看典型代码public static void main(String[] args) throws Exception { // 1. 构造 Kafka SpoutspoutConfig 完整配置见前面章节的示例 SpoutConfig spoutConfig buildSpoutConfig(); KafkaSpout kafkaSpout new KafkaSpout(spoutConfig); // 2. 构建拓扑按数据流方向逐级 set TopologyBuilder builder new TopologyBuilder(); builder.setSpout(kafka_spout, kafkaSpout, 2); // 独立定时节点只消费系统 tick加载规则后按广播流下发 builder.setBolt(tick_bolt, new StormTickBolt(), 1); // 规则匹配 Bolt日志按 str 字段分桶规则快照全量广播 builder.setBolt(process_bolt, new ProcessDataBolt(), 4) .fieldsGrouping(kafka_spout, new Fields(str)) .allGrouping(tick_bolt, rule_stream); // 通知 BoltshuffleGrouping 随机分发所有 process 输出均分 builder.setBolt(notify_bolt, new NotifyMessageBolt(), 2) .shuffleGrouping(process_bolt); // 落库 Bolt单独一组避免写库阻塞通知 builder.setBolt(save_bolt, new SaveToDBBolt(), 2) .shuffleGrouping(process_bolt); // 3. 提交到 Storm 集群 Config conf new Config(); conf.setNumWorkers(4); conf.setMaxSpoutPending(200); StormSubmitter.submitTopology(log_monitor_topo, conf, builder.createTopology()); }几个关键点说明。setSpout和setBolt的第三个参数是并行度Kafka Spout并行度建议和Topic分区数保持一致分区是物理分片多设并行度也没用process_bolt的4和notify/save的2是处理能力分层规则匹配是CPU密集可以多开通知和落库是IO密集开多了反而打爆数据库和短信网关。fieldsGrouping按str字段分桶这里str是Kafka Spout通过StringScheme输出字段的固定名字它保证同一来源日志顺序不乱tick_bolt输出规则快照流process_bolt用allGrouping接收保证每个process实例都能拿到完整规则不依赖任何共享内存。shuffleGrouping适合无状态节点均匀分发。setMaxSpoutPending(200)控制Spout最多在途tuple数量是一把双刃剑设小了吞吐上不去设大了出问题时重放的数据量也大日志监控场景200是一个不会出大错的起步值。TopkeyTopologMain的装配结构类似只是后半段换成TopkeyCountBolt前段也可以复用同一个KafkaTopic但groupId要换一个否则会和主拓扑抢offset。3.2 StormTickBolt与TopkeyCountBolt定时刷新与窗口统计StormTickBolt在prepare()阶段要做的事包括初始化数据库连接、加载第一份规则缓存并且确认tick频率已配置。prepare只执行一次连接资源在这里建好整个Bolt生命周期复用。TopkeyCountBolt则是另一个方向的代码核心是累积计数和TopN输出。public class TopkeyCountBolt extends BaseRichBolt { private OutputCollector collector; // 关键词 - 计数。窗口内统计达到窗口大小整体输出一次 private MapString, Long countMap new HashMap(); public void execute(Tuple input) { String log input.getStringByField(str); String keyword extractTopKey(log); // 从日志里摘出需要统计的字段或关键词 if (keyword ! null) { countMap.merge(keyword, 1L, Long::sum); } collector.ack(input); } private String extractTopKey(String log) { // 这里的解析逻辑要按日志格式定制 return null; } }这个Bolt的计数器用Map的merge语法简洁且线程安全边界清晰。它只处理在途tuple不往下游发数据所以execute末尾直接ack。真正把TopN结果打出去的动作常见做法不是每来一条日志就算一次而是由另一个定时线程或tick tuple触发每30秒或者每1万个tuple输出一次排序结果到日志或存储这样既避免高频IO也不会丢中间态。如果要改成精确的滑动窗口统计直接用Storm内置的WindowedBolt更合适但TopkeyCountBolt这种累加器胜在实现直观、依赖少。3.3 CommonUtils与JdbcUtils规则匹配与数据库操作的实现CommonUtils是规则匹配和通知入口的公共类ProcessDataBolt匹配日志时不需要自己写for循环直接调它。规则匹配的常见实现是遍历规则集合逐个用contains判断日志内容是否包含关键字段含正则的规则再走一层Pattern匹配。public static boolean matchRule(String log, MonitorRule rule) { if (rule.getKeyword() ! null log.contains(rule.getKeyword())) { return true; } if (rule.getRegexPattern() ! null rule.getRegexPattern().matcher(log).find()) { return true; } return false; }这段逻辑虽然短但有两个参数细节要注意规则里的keyword用contains而不是equals因为日志是一整串文本只要包含关键词就算命中正则分支的Pattern对象应该在规则加载阶段就编译好缓存起来如果每条日志进来再编译一次正则CPU开销会成倍上涨。工程里MonitorRule这类封装对象就是StormTickBolt加载到内存缓存的规则快照。JdbcUtils承担所有数据库访问包括加载规则列表和写入告警记录。连接通过外部传入SQL用PreparedStatement而不是拼接一是防注入二是数据库可以复用执行计划。public static ListMonitorRule loadAlertRules(Connection conn) throws SQLException { ListMonitorRule list new ArrayList(); String sql SELECT rule_id, keyword, regex_pattern, notify_type FROM alert_rule WHERE status 1; try (PreparedStatement ps conn.prepareStatement(sql); ResultSet rs ps.executeQuery()) { while (rs.next()) { MonitorRule rule new MonitorRule(); rule.setRuleId(rs.getInt(rule_id)); rule.setKeyword(rs.getString(keyword)); rule.setRegexPattern(rs.getString(regex_pattern)); rule.setNotifyType(rs.getInt(notify_type)); list.add(rule); } } return list; }这里用try-with-resources管理ResultSet和PreparedStatement方法结束自动关闭避免连接泄漏。注意传入的Connection是外部管理的这里只负责关闭Statement不关Connection否则复用就断掉了。这个边界很多人会搞混连着把Connection也close掉下次调用又得新建。实际项目里如果规则表和告警表在同一个库JdbcUtils里还应该维护一个简单的连接池把Connection的创建和归还统一管起来。4. 部署与运行环境准备、配置项与提交命令代码只是第一步拓扑真正在集群里跑起来才算数。这一章从零梳理部署流程先列环境依赖和版本匹配再给数据库建表和关键配置项最后是提交、停止、查看拓扑的常用命令。日志监控这种场景集群通常不大三步就能走完。4.1 环境准备JDK、ZooKeeper、Kafka与Storm的版本匹配这套系统依赖四个组件版本匹配是部署期最大的坑。从工程里使用KafkaSpout和storm-kafka的API风格来看它对应的是Storm 1.x生态建议组合是JDK 1.8、ZooKeeper 3.4.x/3.5.x、Kafka 0.10.x到0.11.x、Storm 1.2.x。JDK版本不要随意升到11以上Storm 1.x的某些字节码库在JDK 11下有兼容问题。Kafka版本尤其要注意storm-kafka的KafkaSpout只实现了旧版Consumer API对应Kafka 0.8到0.11这一代如果Kafka升到2.x就得换storm-kafka-client的KafkaSpoutAPI完全不一样工程改动会比较大。组件部署顺序也固定先起ZooKeeper再起Kafka等Kafka的broker注册到ZK后最后配Storm。保证ZooKeeper集群的地址在Kafka和Storm两边的配置里都能访问。单机测试时可以全部装在一台机器上生产环境至少要ZK三节点、Kafka两节点起Storm的Nimbus和Supervisor分开部署。4.2 配置与初始化storm.yaml、规则表结构与参数清单Storm集群的配置在conf/storm.yaml里拓扑提交前需要确认几个基础项storm.zookeeper.servers: - zk1 - zk2 - zk3 nimbus.seeds: [nimbus1] storm.local.dir: /data/storm supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703supervisor的slot端口决定每台机器能跑多少个Worker4个端口就是4个Worker槽位。之前见过一个翻车案例拓扑提交成功但任务一直不调度原因就是Worker槽位被别的拓扑占满这里留的4个端口足够支撑一个监控拓扑。数据库侧需要两张核心表建表语句给到CREATE TABLE alert_rule ( rule_id INT PRIMARY KEY AUTO_INCREMENT, app_name VARCHAR(64) NOT NULL, keyword VARCHAR(128), regex_pattern VARCHAR(255), notify_type TINYINT DEFAULT 1, status TINYINT DEFAULT 1, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE alert_record ( alert_id BIGINT PRIMARY KEY AUTO_INCREMENT, log_content TEXT, rule_id INT, notify_type TINYINT, notify_status TINYINT, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP );alert_rule是StormTickBolt定时加载的数据来源status字段用来启停规则不需要删记录alert_record是告警落库表notify_status记录邮件短信发送成功还是失败。两张表的字符集统一用utf8mb4日志里经常有特殊字符用utf8可以但遇到emoji会报错utf8mb4省心。几个关键参数先列个清单部署时对着调参数建议值说明spout并行度等于Kafka分区数多设无效少设会漏消费分区setMaxSpoutPending200~500在途tuple上限控制吞吐与重放量tick刷新频率60秒规则生效延迟改小需评估查库压力process_bolt并行度CPU核数×2规则匹配是CPU密集操作notify/save并行度2IO密集节点不宜过高4.3 打包提交与运维命令storm jar、storm kill与日志查看工程用Maven管理依赖时打包要带上Storm相关依赖但提交时Storm框架的jar已经存在集群里所以这些依赖要设成provided否则提交会报类冲突。标准操作是mvn clean package后用storm jar命令把拓扑提交到集群# 打包跳过测试 mvn clean package -DskipTests # 提交主监控拓扑 storm jar target/log-monitor-1.0.jar \ com.yourpackage.TopologyMain prod # 提交 TopKey 统计拓扑主类不同 storm jar target/log-monitor-1.0.jar \ com.yourpackage.TopkeyTopologMain prod # 查看已提交的拓扑 storm list # 停止拓扑-w 表示等待多少秒后再杀进程 storm kill log_monitor_topo -w 10storm list返回结果里看STATUS字段ACTIVE表示正常运行KILLED表示停止中REBALANCING表示正在调整并行度。拓扑报错时第一时间查看Worker日志位置在Storm目录的logs/workers-artifacts/拓扑名/ /worker.log。如果这个命令在生产环境需要后台常驻用nohup把它挂在系统后台同时配合系统性能监控工具观察Worker进程的CPU和内存占用后面避坑部分会提到。5. 避坑指南五个高频故障的现象、原因与解决方法这一章写我在类似架构上真实踩过和帮人排查过的坑全部按现象→原因→解决的结构写。每一条都可以直接对号入座。5.1 现象StormTickBolt定时刷新不生效改规则后告警一直不变化现象数据库里改了keyword等到预期时间后日志命中规则的行为没有变化甚至重启拓扑才生效。原因大多数情况是tick频率没配置上。tick机制不是Bolt类里写了isTickTuple判断就生效的还要在拓扑的Config里设置对应的tick频率而且这个参数是绑定到具体Bolt的漏配就直接没有tick数据进来。另一个常见原因是在全局Config和Bolt的getComponentConfiguration里同时出现后者的值覆盖了前者导致你以为配了60秒实际跑的还是默认值。解决核对拓扑代码里是否对StormTickBolt单独设置了Config.setTickTupleFreqSecs确认这个方法把Bolt的componentId和频率都传了不要只写在全局Config里。改完配置后执行storm rebalance让拓扑重新调度静态代码里改tick频率需要重启或rebalance才生效。5.2 现象Kafka Spout消费速率上不去日志积压越来越多现象Kafka的消费组监控显示lag持续上涨但Storm UI里Spout的CPU占用并不高没有任何异常报错。原因并行度与Topic分区数不匹配。Kafka一个分区同一时刻只能被一个Spout实例消费如果Spout并行度大于分区数多出的实例空转如果小于分区数一部分分区没人消费。另一个典型原因是setMaxSpoutPending设置过小Spout发出的tuple没被Bolt及时ack在途数量触顶后Spout主动不拉数据。解决先查Topic的分区数用kafka-topics.sh --describe再把Spout并行度设成等于分区数。setMaxSpoutPending的值根据Bolt处理延迟来调process_bolt平均每条处理10毫秒200在途够用延迟高就加大到500再观察。还有一个容易被忽略的点同一个consumer group下如果多个拓扑并行消费同一个Topicoffset互相覆盖会出现消费位置来回跳表现为消费速率极不稳定。5.3 现象告警重复发送同一条日志连续收到多次邮件现象一条包含错误关键字的日志邮箱里收到两三条内容完全一样的告警。原因tuple处理超时没有ackKafkaSpout重发。Storm里tuple默认超时时间是30秒ProcessDataBolt如果因为数据库慢或通知接口慢超过这个时间还没调用ackStorm会认为处理失败Spout按重试机制重新发送这条日志下游会再走一遍规则匹配和通知。这不是业务逻辑重复是可靠性重放机制在工作。解决先看重复记录在告警表里的create_time如果间隔恰好接近tuple超时时间就是重放导致的。对策有三个方向把Config.setMessageTimeoutSecs适当调大比如60到120秒给Bolt更多处理时间检查通知链路的IO耗时尽量把发送动作丢到异步线程池避免处理时间超过超时阈值如果业务允许在NotifyMessageBolt里按日志内容加一个时间窗口去重缓存同一内容的告警在10秒内只发一次这招在通知链路抖动时最管用。5.4 现象JdbcUtils查询一段时间后报连接失效拓扑间接性失败现象拓扑运行几个小时后SaveToDBBolt开始抛SQLException错误信息指向连接已关闭或通信链路异常重启后恢复过几小时又复现。原因数据库的连接有生命周期空闲超过wait_timeout时间服务端会主动断开。Bolt里的Connection如果在prepare阶段建好后就一直复用中间没有任何SQL执行的话等下一次突然来数据时这条连接已经被数据库回收了再executeUpdate就会报错。解决给JdbcUtils的连接管理加一个简单的心跳每次从连接池取连接时先validate无效就重建。更稳的做法是不在Bolt里长期持有单一连接而是用Druid或HikariCP这样的连接池由连接池负责空闲检测和重连getConnection和close成对调用每次用完归还给池而不是真关闭。这一改动涉及代码但不大对长时间运行的拓扑是刚需。5.5 现象提交时报NoSuchMethodError或类冲突拓扑起不来现象storm jar提交后Nimbus日志里抛出NoSuchMethodError、NoClassDefFoundError或类冲突异常拓扑反复提交都失败。原因storm-kafka与Storm主版本不匹配或者打包时把Storm依赖以非provided方式打进了fat jar。旧版storm-kafka里的类对Kafka客户端版本敏感Kafka 2.x客户端里某些类和方法已经被移除NoSuchMethodError就是这么来的。解决对照第4章的版本组合确认Kafka客户端版本和storm-kafka选型一致。如果用的Storm 1.2.xKafka客户端不要超过0.11打包时在pom里把storm-core和storm-kafka的scope设为provided提交时用集群里的jar。排查类冲突时用storm classpath看看Nimbus实际的classpath顺序再根据报错类名反查是哪个jar引入的。6. 进阶用一条测试消息验证全链路再做TopKey拓扑扩展拓扑部署到集群后先不要急着写更多规则先用一个最简单的动作验证它真能跑通整条链路。测试前先准备一条符合你场景的样例日志最好包含alert_rule表里已配置的关键字。然后往Kafka的app_log_topic发送一条消息# 生产一条测试日志内容包含规则里的关键字 exception echo {log: [2025-01-01 12:00:01] ERROR order-service: NullPointerException at OrderServiceImpl} | \ kafka-console-producer.sh --broker-list kafka1:9092 --topic app_log_topic发送成功后按顺序检查三处第一看数据库的alert_record表确认新增了一条告警记录notify_status字段如果是1说明通知链路也返回了成功第二查一下邮件收件箱或短信网关日志确认告警真的触达了接收人第三到Storm UI看Bolt的指标主要看process_bolt的executed和failed数值failed为0且executed持续递增说明tuple一直在正常推进。如果这三个检查全部通过这条监控链路就完成了闭环验证后面加规则、换模板都不需要再动拓扑。验证通过之后再谈扩展。TopkeyTopologMain就是一个很好的扩展方向它把日志统计从告警链路中单独拆出来专门做关键词的频次TopN统计不参与通知和落库。部署时让它消费同一个KafkaTopic但用独立的groupId避免offset互相干扰。统计结果挂在另一个Bolt输出直接落到单独的表里定时任务再扫这张表生成日报。这样主监控拓扑管实时告警统计拓扑管离线分析两个拓扑互不拖累Worker资源也能按各自负载单独调整并行度。把验证和扩展做完这套资源里的TopologyMain和TopkeyTopologMain源码基本可以直接拿过去改一下包名、Topic名和数据库连接串按第4章的步骤提交就能跑起来。最后说个我自己的习惯从那以后我每次提交改动都会先跑一遍这条“生产一条测试日志→查alert_record→看Storm UI的failed指标”的流程确认上下游闭环再合代码。规则表每次变更也强制检查一次status和notify_type字段避免配了一条不生效的规则误导排障。这套校验流程用不了五分钟但它能拦住大部分低级问题希望帮到你。本文还有配套的精品资源点击获取
简介一套基于Java与Apache Storm构建的日志监控告警系统方案面向大数据实时计算、日志分析和运维监控方向的开发者解决日志数据实时消费、规则匹配与多渠道告警联动的典型问题。系统以Kafka作为日志接入层通过Storm拓扑完成从数据读取、规则加载、异常匹配到邮件/短信通知、数据库落地的完整链路。资源包共100个文件大小仅1.17MB以Java源码24个java与编译产物27个class为主体配合4个xml项目配置、3个iml模块文件另有39张png图片和2份md文档便于查看项目结构、运行界面及设计说明。核心实现涵盖Kafka数据源接入、定时规则加载Bolt、日志处理与匹配Bolt、邮件短信通知Bolt、数据库保存Bolt以及规则匹配、通知发送、JDBC等工具类可从中了解完整Storm拓扑的组装方式与告警处理流程。已有159人学习下载。对想动手实践Storm实时计算、理解Kafka对接及告警通知实现细节的读者可基于源码快速搭建自己的日志监控告警原型也可结合图片与文档梳理异常检测逻辑。1. 日志告警不是ELK专利Kafka脏日志到邮件短信的实时链路日志量每天几十万条想第一时间发现报错而不是等业务方找上门的时候大家优先想到的往往是ELK那一整套。但如果你已经在用Kafka团队技术栈又正好是纯JavaApache Storm这套轻量拓扑反而是更顺手的方案。这份基于Java与Storm的日志监控告警系统正好补齐了“Kafka Topic→规则匹配→邮件短信通知→告警落库”的完整链路。工程里包含TopologyMain与TopkeyTopologMain两个拓扑入口、StormTickBolt定时加载监控规则、ProcessDataBolt做规则匹配、NotifyMessageBolt与SaveToDBBolt分别负责通知分发和数据落库还有CommonUtils与JdbcUtils两类工具类兜底。它适合已有Kafka数据源、想用纯Java实时处理日志告警的团队也适合做Storm课程设计的同学直接照着复现。2. 系统架构与核心组件从Kafka Topic到通知下发的数据流先把整条数据流立起来Kafka Spout从指定Topic拉取日志tuple经fieldsGrouping按内容字段分发给ProcessDataBolt做规则匹配StormTickBolt是一个独立定时节点按固定频率从数据库加载监控规则和应用信息再以广播流把规则快照发给所有ProcessDataBolt实例命中的tuple分发给NotifyMessageBolt发送邮件和短信同时SaveToDBBolt把告警记录写入数据库没命中的日志直接ack掉不落库。整条拓扑只有5个节点中间没有窗口聚合节点按tuple逐条处理端到端延迟在秒级适合日志量中等、但对报警及时性有要求的场景。这套设计没有引入独立的规则引擎或消息中间件规则匹配就放在Bolt进程内完成通知直接走邮件和短信网关。这样做的取舍是架构简单、好排查代价是规则复杂度做不深正则和组合条件都靠CommonUtils里那几行逻辑扛。下面逐个拆节点看它是怎么工作的。2.1 Kafka Spout与TopologyMain日志数据进入拓扑的通道Spout是Storm拓扑的数据源头这个工程用的是storm-kafka客户端里的KafkaSpout而不是在Bolt里自己写KafkaConsumer轮询。为什么因为Storm里Spout的nextTuple()由框架反复调用如果用原生Consumer你得自己维护线程、拉取和位移提交而KafkaSpout把fetch、offset提交、失败重放全部封装好了。项目里TopologyMain就是负责装配这些组件的入口类。ZkHosts zkHosts new ZkHosts(zk1:2181,zk2:2181,zk3:2181); SpoutConfig spoutConfig new SpoutConfig( zkHosts, app_log_topic, // 消费的日志 Topic /log_monitor/kafka_offset, // offset 在 ZooKeeper 中的保存根路径 log_monitor_group); // consumer group id spoutConfig.scheme new SchemeAsMultiScheme(new StringScheme()); spoutConfig.forceFromStart false; // false 表示接续上次提交的 offset spoutConfig.zkRoot /log_monitor/kafka_offset; KafkaSpout kafkaSpout new KafkaSpout(spoutConfig); TopologyBuilder builder new TopologyBuilder(); builder.setSpout(kafka_spout, kafkaSpout, 2); // 并行度设为 2参数逐个看ZkHosts只传ZooKeeper地址就够了Kafka的broker元数据注册在ZK的/brokers/ids路径下KafkaSpout启动时会自动拉取集群broker列表后面Kafka扩节点时拓扑不用改配置第二个参数Topic名要和Kafka里建的Topic完全一致第三个参数是offset在ZK里的存储路径多个拓扑消费同一个Topic时这个路径不能一样第四个参数groupId相当于消费组标识两个拓扑用同一个groupId消费同一个Topic会把彼此offset挤掉。forceFromStart为false代表正常接续上次消费进度只有做数据重放测试时才改成true。Spout装配完成以后下游Bolt要关注tuple的可靠性。日志监控场景里普通日志丢一两次问题不大但告警丢了会被投诉所以ProcessDataBolt处理完必须调用collector.ack(input)中途异常就调fail(input)KafkaSpout会按重试机制重新拉取这段消息。这一点在第5章避坑里专门展开。2.2 StormTickBolttick频率与规则加载机制监控规则如果写死在代码里每次加关键字都要重新打包提交拓扑排障场景下几十分钟延迟完全不能忍。这个系统把规则放在数据库表里由StormTickBolt定时加载。定时触发靠的就是Storm的tick tuple给某个Bolt配置tick频率后框架会周期性往它的execute()里塞一条特殊tuplesourceComponent固定为__systemsourceStreamId固定为__tick。public void execute(Tuple input) { if (isTickTuple(input)) { // 到点刷新数据库里的规则、应用、用户信息 ListMonitorRule rules loadRulesFromDB(); // 把规则快照通过独立流广播给下游所有 process_bolt 实例 collector.emit(rule_stream, input, new Values(rules)); return; } // 非 tick 数据tick_bolt 本身不处理业务数据直接 ack collector.ack(input); } private boolean isTickTuple(Tuple input) { return __system.equals(input.getSourceComponent()) __tick.equals(input.getSourceStreamId()); }这段逻辑的重点是loadRulesFromDB()的实现方式。不能先clear再load那会留一个空缓存窗口ProcessDataBolt查不到规则导致告警漏报。我一般会新建一个List或HashMap查完库整体替换旧引用ProcessDataBolt拿到的永远是一份完整的规则快照。tick频率在提交拓扑时通过Config配置典型值是60秒一次也就是数据库改规则后最多延迟1分钟生效。如果要求秒级响应可以调到10秒但每次刷新都是一次全量查库规则表几万行以内无所谓再大就要考虑增量加载。2.3 告警通知链路邮件与短信分发ProcessDataBolt匹配到异常日志后把日志内容、命中规则ID、用户信息封装成告警tuple发给NotifyMessageBolt。这个Bolt职责很集中根据告警tuple里的通知方式调用邮件或短信工具把消息推出去。工程里MailInfo是邮件实体封装收件人、主题、正文、SMTP服务器、账号密码这些字段MessageSenderUtil封装了JavaMail的Transport发送逻辑ShortMessageUtil负责调短信网关的HTTP接口。这里有一个值得说的设计点邮件和短信发送都是阻塞IO。如果某个短信网关接口抖动响应三五秒NotifyMessageBolt整个线程就被卡住后续所有告警tuple排队等待上游Bolt也会因为背压导致消费缓慢。我实际踩到过这个问题后来的习惯是给这个Bolt塞一个固定大小的线程池发送请求丢给线程池执行Bolt本身只负责收tuple和提交发送任务再配合失败重试队列短信网关再慢也只是发送线程池排队不影响主链路的tuple消费。2.4 数据落库与工具类SaveToDBBolt和JdbcUtils的职责边界告警记录必须留底否则事后排查问题只能靠运气。SaveToDBBolt接收ProcessDataBolt的告警数据通过JdbcUtils拿到数据库连接执行INSERT把告警内容、匹配规则ID、通知时间和状态写进告警表。这里最容易翻车的点是数据库连接的获取方式。最忌讳的做法是每个tuple都新建一个Connection日志量一大数据库连接数直接被打满表现为数据库端大量报Too many connections。常规做法是在SaveToDBBolt的prepare()里初始化一个连接池或者用JdbcUtils里维护的池化连接Bolt生命周期内反复复用。CommonUtils在工程里是公共门面规则匹配判断、告警对象构造、通知调用入口都在它里面避免ProcessDataBolt和NotifyMessageBolt各写一套重复逻辑。它同时也是这几个Bolt之间的隐式约定层字段名改了只需要动一个类。数据流到这里就完整了KafkaSpout进通知落库出每一层的职责都不重叠。3. 核心代码拆解拓扑装配、Bolt实现与工具类实战架构层面的分工已经清楚接下来直接看代码实现。整个工程虽然类不多但分层是清楚的两个拓扑入口类负责装配Bolt类各自管一段处理逻辑工具类提供公共能力实体类做数据传输。我会按入口、Bolt、工具三个层次逐个拆每一步给关键代码和参数说明方便你直接对着改。3.1 TopologyMain与TopkeyTopologMain两个拓扑入口的差异TopologyMain是主告警拓扑入口装配的是“消费日志→规则匹配→通知→落库”这条链路。TopkeyTopologMain从命名看是TopKey统计拓扑的入口配合TopkeyCountBolt做日志关键词的频次统计它不直接发告警而是输出统计结果。这两个入口在工程里是独立的main方法提交时用storm jar分别指定主类即可它们跑在各自的Worker进程里互不影响。TopologyMain的装配顺序很关键看典型代码public static void main(String[] args) throws Exception { // 1. 构造 Kafka SpoutspoutConfig 完整配置见前面章节的示例 SpoutConfig spoutConfig buildSpoutConfig(); KafkaSpout kafkaSpout new KafkaSpout(spoutConfig); // 2. 构建拓扑按数据流方向逐级 set TopologyBuilder builder new TopologyBuilder(); builder.setSpout(kafka_spout, kafkaSpout, 2); // 独立定时节点只消费系统 tick加载规则后按广播流下发 builder.setBolt(tick_bolt, new StormTickBolt(), 1); // 规则匹配 Bolt日志按 str 字段分桶规则快照全量广播 builder.setBolt(process_bolt, new ProcessDataBolt(), 4) .fieldsGrouping(kafka_spout, new Fields(str)) .allGrouping(tick_bolt, rule_stream); // 通知 BoltshuffleGrouping 随机分发所有 process 输出均分 builder.setBolt(notify_bolt, new NotifyMessageBolt(), 2) .shuffleGrouping(process_bolt); // 落库 Bolt单独一组避免写库阻塞通知 builder.setBolt(save_bolt, new SaveToDBBolt(), 2) .shuffleGrouping(process_bolt); // 3. 提交到 Storm 集群 Config conf new Config(); conf.setNumWorkers(4); conf.setMaxSpoutPending(200); StormSubmitter.submitTopology(log_monitor_topo, conf, builder.createTopology()); }几个关键点说明。setSpout和setBolt的第三个参数是并行度Kafka Spout并行度建议和Topic分区数保持一致分区是物理分片多设并行度也没用process_bolt的4和notify/save的2是处理能力分层规则匹配是CPU密集可以多开通知和落库是IO密集开多了反而打爆数据库和短信网关。fieldsGrouping按str字段分桶这里str是Kafka Spout通过StringScheme输出字段的固定名字它保证同一来源日志顺序不乱tick_bolt输出规则快照流process_bolt用allGrouping接收保证每个process实例都能拿到完整规则不依赖任何共享内存。shuffleGrouping适合无状态节点均匀分发。setMaxSpoutPending(200)控制Spout最多在途tuple数量是一把双刃剑设小了吞吐上不去设大了出问题时重放的数据量也大日志监控场景200是一个不会出大错的起步值。TopkeyTopologMain的装配结构类似只是后半段换成TopkeyCountBolt前段也可以复用同一个KafkaTopic但groupId要换一个否则会和主拓扑抢offset。3.2 StormTickBolt与TopkeyCountBolt定时刷新与窗口统计StormTickBolt在prepare()阶段要做的事包括初始化数据库连接、加载第一份规则缓存并且确认tick频率已配置。prepare只执行一次连接资源在这里建好整个Bolt生命周期复用。TopkeyCountBolt则是另一个方向的代码核心是累积计数和TopN输出。public class TopkeyCountBolt extends BaseRichBolt { private OutputCollector collector; // 关键词 - 计数。窗口内统计达到窗口大小整体输出一次 private MapString, Long countMap new HashMap(); public void execute(Tuple input) { String log input.getStringByField(str); String keyword extractTopKey(log); // 从日志里摘出需要统计的字段或关键词 if (keyword ! null) { countMap.merge(keyword, 1L, Long::sum); } collector.ack(input); } private String extractTopKey(String log) { // 这里的解析逻辑要按日志格式定制 return null; } }这个Bolt的计数器用Map的merge语法简洁且线程安全边界清晰。它只处理在途tuple不往下游发数据所以execute末尾直接ack。真正把TopN结果打出去的动作常见做法不是每来一条日志就算一次而是由另一个定时线程或tick tuple触发每30秒或者每1万个tuple输出一次排序结果到日志或存储这样既避免高频IO也不会丢中间态。如果要改成精确的滑动窗口统计直接用Storm内置的WindowedBolt更合适但TopkeyCountBolt这种累加器胜在实现直观、依赖少。3.3 CommonUtils与JdbcUtils规则匹配与数据库操作的实现CommonUtils是规则匹配和通知入口的公共类ProcessDataBolt匹配日志时不需要自己写for循环直接调它。规则匹配的常见实现是遍历规则集合逐个用contains判断日志内容是否包含关键字段含正则的规则再走一层Pattern匹配。public static boolean matchRule(String log, MonitorRule rule) { if (rule.getKeyword() ! null log.contains(rule.getKeyword())) { return true; } if (rule.getRegexPattern() ! null rule.getRegexPattern().matcher(log).find()) { return true; } return false; }这段逻辑虽然短但有两个参数细节要注意规则里的keyword用contains而不是equals因为日志是一整串文本只要包含关键词就算命中正则分支的Pattern对象应该在规则加载阶段就编译好缓存起来如果每条日志进来再编译一次正则CPU开销会成倍上涨。工程里MonitorRule这类封装对象就是StormTickBolt加载到内存缓存的规则快照。JdbcUtils承担所有数据库访问包括加载规则列表和写入告警记录。连接通过外部传入SQL用PreparedStatement而不是拼接一是防注入二是数据库可以复用执行计划。public static ListMonitorRule loadAlertRules(Connection conn) throws SQLException { ListMonitorRule list new ArrayList(); String sql SELECT rule_id, keyword, regex_pattern, notify_type FROM alert_rule WHERE status 1; try (PreparedStatement ps conn.prepareStatement(sql); ResultSet rs ps.executeQuery()) { while (rs.next()) { MonitorRule rule new MonitorRule(); rule.setRuleId(rs.getInt(rule_id)); rule.setKeyword(rs.getString(keyword)); rule.setRegexPattern(rs.getString(regex_pattern)); rule.setNotifyType(rs.getInt(notify_type)); list.add(rule); } } return list; }这里用try-with-resources管理ResultSet和PreparedStatement方法结束自动关闭避免连接泄漏。注意传入的Connection是外部管理的这里只负责关闭Statement不关Connection否则复用就断掉了。这个边界很多人会搞混连着把Connection也close掉下次调用又得新建。实际项目里如果规则表和告警表在同一个库JdbcUtils里还应该维护一个简单的连接池把Connection的创建和归还统一管起来。4. 部署与运行环境准备、配置项与提交命令代码只是第一步拓扑真正在集群里跑起来才算数。这一章从零梳理部署流程先列环境依赖和版本匹配再给数据库建表和关键配置项最后是提交、停止、查看拓扑的常用命令。日志监控这种场景集群通常不大三步就能走完。4.1 环境准备JDK、ZooKeeper、Kafka与Storm的版本匹配这套系统依赖四个组件版本匹配是部署期最大的坑。从工程里使用KafkaSpout和storm-kafka的API风格来看它对应的是Storm 1.x生态建议组合是JDK 1.8、ZooKeeper 3.4.x/3.5.x、Kafka 0.10.x到0.11.x、Storm 1.2.x。JDK版本不要随意升到11以上Storm 1.x的某些字节码库在JDK 11下有兼容问题。Kafka版本尤其要注意storm-kafka的KafkaSpout只实现了旧版Consumer API对应Kafka 0.8到0.11这一代如果Kafka升到2.x就得换storm-kafka-client的KafkaSpoutAPI完全不一样工程改动会比较大。组件部署顺序也固定先起ZooKeeper再起Kafka等Kafka的broker注册到ZK后最后配Storm。保证ZooKeeper集群的地址在Kafka和Storm两边的配置里都能访问。单机测试时可以全部装在一台机器上生产环境至少要ZK三节点、Kafka两节点起Storm的Nimbus和Supervisor分开部署。4.2 配置与初始化storm.yaml、规则表结构与参数清单Storm集群的配置在conf/storm.yaml里拓扑提交前需要确认几个基础项storm.zookeeper.servers: - zk1 - zk2 - zk3 nimbus.seeds: [nimbus1] storm.local.dir: /data/storm supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703supervisor的slot端口决定每台机器能跑多少个Worker4个端口就是4个Worker槽位。之前见过一个翻车案例拓扑提交成功但任务一直不调度原因就是Worker槽位被别的拓扑占满这里留的4个端口足够支撑一个监控拓扑。数据库侧需要两张核心表建表语句给到CREATE TABLE alert_rule ( rule_id INT PRIMARY KEY AUTO_INCREMENT, app_name VARCHAR(64) NOT NULL, keyword VARCHAR(128), regex_pattern VARCHAR(255), notify_type TINYINT DEFAULT 1, status TINYINT DEFAULT 1, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE alert_record ( alert_id BIGINT PRIMARY KEY AUTO_INCREMENT, log_content TEXT, rule_id INT, notify_type TINYINT, notify_status TINYINT, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP );alert_rule是StormTickBolt定时加载的数据来源status字段用来启停规则不需要删记录alert_record是告警落库表notify_status记录邮件短信发送成功还是失败。两张表的字符集统一用utf8mb4日志里经常有特殊字符用utf8可以但遇到emoji会报错utf8mb4省心。几个关键参数先列个清单部署时对着调参数建议值说明spout并行度等于Kafka分区数多设无效少设会漏消费分区setMaxSpoutPending200~500在途tuple上限控制吞吐与重放量tick刷新频率60秒规则生效延迟改小需评估查库压力process_bolt并行度CPU核数×2规则匹配是CPU密集操作notify/save并行度2IO密集节点不宜过高4.3 打包提交与运维命令storm jar、storm kill与日志查看工程用Maven管理依赖时打包要带上Storm相关依赖但提交时Storm框架的jar已经存在集群里所以这些依赖要设成provided否则提交会报类冲突。标准操作是mvn clean package后用storm jar命令把拓扑提交到集群# 打包跳过测试 mvn clean package -DskipTests # 提交主监控拓扑 storm jar target/log-monitor-1.0.jar \ com.yourpackage.TopologyMain prod # 提交 TopKey 统计拓扑主类不同 storm jar target/log-monitor-1.0.jar \ com.yourpackage.TopkeyTopologMain prod # 查看已提交的拓扑 storm list # 停止拓扑-w 表示等待多少秒后再杀进程 storm kill log_monitor_topo -w 10storm list返回结果里看STATUS字段ACTIVE表示正常运行KILLED表示停止中REBALANCING表示正在调整并行度。拓扑报错时第一时间查看Worker日志位置在Storm目录的logs/workers-artifacts/拓扑名/ /worker.log。如果这个命令在生产环境需要后台常驻用nohup把它挂在系统后台同时配合系统性能监控工具观察Worker进程的CPU和内存占用后面避坑部分会提到。5. 避坑指南五个高频故障的现象、原因与解决方法这一章写我在类似架构上真实踩过和帮人排查过的坑全部按现象→原因→解决的结构写。每一条都可以直接对号入座。5.1 现象StormTickBolt定时刷新不生效改规则后告警一直不变化现象数据库里改了keyword等到预期时间后日志命中规则的行为没有变化甚至重启拓扑才生效。原因大多数情况是tick频率没配置上。tick机制不是Bolt类里写了isTickTuple判断就生效的还要在拓扑的Config里设置对应的tick频率而且这个参数是绑定到具体Bolt的漏配就直接没有tick数据进来。另一个常见原因是在全局Config和Bolt的getComponentConfiguration里同时出现后者的值覆盖了前者导致你以为配了60秒实际跑的还是默认值。解决核对拓扑代码里是否对StormTickBolt单独设置了Config.setTickTupleFreqSecs确认这个方法把Bolt的componentId和频率都传了不要只写在全局Config里。改完配置后执行storm rebalance让拓扑重新调度静态代码里改tick频率需要重启或rebalance才生效。5.2 现象Kafka Spout消费速率上不去日志积压越来越多现象Kafka的消费组监控显示lag持续上涨但Storm UI里Spout的CPU占用并不高没有任何异常报错。原因并行度与Topic分区数不匹配。Kafka一个分区同一时刻只能被一个Spout实例消费如果Spout并行度大于分区数多出的实例空转如果小于分区数一部分分区没人消费。另一个典型原因是setMaxSpoutPending设置过小Spout发出的tuple没被Bolt及时ack在途数量触顶后Spout主动不拉数据。解决先查Topic的分区数用kafka-topics.sh --describe再把Spout并行度设成等于分区数。setMaxSpoutPending的值根据Bolt处理延迟来调process_bolt平均每条处理10毫秒200在途够用延迟高就加大到500再观察。还有一个容易被忽略的点同一个consumer group下如果多个拓扑并行消费同一个Topicoffset互相覆盖会出现消费位置来回跳表现为消费速率极不稳定。5.3 现象告警重复发送同一条日志连续收到多次邮件现象一条包含错误关键字的日志邮箱里收到两三条内容完全一样的告警。原因tuple处理超时没有ackKafkaSpout重发。Storm里tuple默认超时时间是30秒ProcessDataBolt如果因为数据库慢或通知接口慢超过这个时间还没调用ackStorm会认为处理失败Spout按重试机制重新发送这条日志下游会再走一遍规则匹配和通知。这不是业务逻辑重复是可靠性重放机制在工作。解决先看重复记录在告警表里的create_time如果间隔恰好接近tuple超时时间就是重放导致的。对策有三个方向把Config.setMessageTimeoutSecs适当调大比如60到120秒给Bolt更多处理时间检查通知链路的IO耗时尽量把发送动作丢到异步线程池避免处理时间超过超时阈值如果业务允许在NotifyMessageBolt里按日志内容加一个时间窗口去重缓存同一内容的告警在10秒内只发一次这招在通知链路抖动时最管用。5.4 现象JdbcUtils查询一段时间后报连接失效拓扑间接性失败现象拓扑运行几个小时后SaveToDBBolt开始抛SQLException错误信息指向连接已关闭或通信链路异常重启后恢复过几小时又复现。原因数据库的连接有生命周期空闲超过wait_timeout时间服务端会主动断开。Bolt里的Connection如果在prepare阶段建好后就一直复用中间没有任何SQL执行的话等下一次突然来数据时这条连接已经被数据库回收了再executeUpdate就会报错。解决给JdbcUtils的连接管理加一个简单的心跳每次从连接池取连接时先validate无效就重建。更稳的做法是不在Bolt里长期持有单一连接而是用Druid或HikariCP这样的连接池由连接池负责空闲检测和重连getConnection和close成对调用每次用完归还给池而不是真关闭。这一改动涉及代码但不大对长时间运行的拓扑是刚需。5.5 现象提交时报NoSuchMethodError或类冲突拓扑起不来现象storm jar提交后Nimbus日志里抛出NoSuchMethodError、NoClassDefFoundError或类冲突异常拓扑反复提交都失败。原因storm-kafka与Storm主版本不匹配或者打包时把Storm依赖以非provided方式打进了fat jar。旧版storm-kafka里的类对Kafka客户端版本敏感Kafka 2.x客户端里某些类和方法已经被移除NoSuchMethodError就是这么来的。解决对照第4章的版本组合确认Kafka客户端版本和storm-kafka选型一致。如果用的Storm 1.2.xKafka客户端不要超过0.11打包时在pom里把storm-core和storm-kafka的scope设为provided提交时用集群里的jar。排查类冲突时用storm classpath看看Nimbus实际的classpath顺序再根据报错类名反查是哪个jar引入的。6. 进阶用一条测试消息验证全链路再做TopKey拓扑扩展拓扑部署到集群后先不要急着写更多规则先用一个最简单的动作验证它真能跑通整条链路。测试前先准备一条符合你场景的样例日志最好包含alert_rule表里已配置的关键字。然后往Kafka的app_log_topic发送一条消息# 生产一条测试日志内容包含规则里的关键字 exception echo {log: [2025-01-01 12:00:01] ERROR order-service: NullPointerException at OrderServiceImpl} | \ kafka-console-producer.sh --broker-list kafka1:9092 --topic app_log_topic发送成功后按顺序检查三处第一看数据库的alert_record表确认新增了一条告警记录notify_status字段如果是1说明通知链路也返回了成功第二查一下邮件收件箱或短信网关日志确认告警真的触达了接收人第三到Storm UI看Bolt的指标主要看process_bolt的executed和failed数值failed为0且executed持续递增说明tuple一直在正常推进。如果这三个检查全部通过这条监控链路就完成了闭环验证后面加规则、换模板都不需要再动拓扑。验证通过之后再谈扩展。TopkeyTopologMain就是一个很好的扩展方向它把日志统计从告警链路中单独拆出来专门做关键词的频次TopN统计不参与通知和落库。部署时让它消费同一个KafkaTopic但用独立的groupId避免offset互相干扰。统计结果挂在另一个Bolt输出直接落到单独的表里定时任务再扫这张表生成日报。这样主监控拓扑管实时告警统计拓扑管离线分析两个拓扑互不拖累Worker资源也能按各自负载单独调整并行度。把验证和扩展做完这套资源里的TopologyMain和TopkeyTopologMain源码基本可以直接拿过去改一下包名、Topic名和数据库连接串按第4章的步骤提交就能跑起来。最后说个我自己的习惯从那以后我每次提交改动都会先跑一遍这条“生产一条测试日志→查alert_record→看Storm UI的failed指标”的流程确认上下游闭环再合代码。规则表每次变更也强制检查一次status和notify_type字段避免配了一条不生效的规则误导排障。这套校验流程用不了五分钟但它能拦住大部分低级问题希望帮到你。本文还有配套的精品资源点击获取
