【基于 Swoole+Hyperf 的微服务实战】第十一周 · 周四：异常检测---简单阈值告警与规则引擎

今天我们进入第十一周周四，主题是异常检测：简单阈值告警与规则引擎。在之前的统计基础上，我们已经能够实时查看 UV/PV，但被动观察不够，我们需要系统主动发现异常并通知我们。今天将在日志消费者中加入实时异常检测能力，基于可配置的规则（如错误日志数量、高频访问 IP），当条件触发时，通过钉钉机器人发送告警，实现从“看”到“响应”的升级。

今日目标

1. 理解日志异常检测的常见场景：错误率飙升、流量突增、特定关键词出现。
2. 设计一个可配置的规则引擎，支持从 Nacos 动态加载规则，无需重启。
3. 在log-consumer中实现异常检测消费者LogAlertConsumer，根据规则实时判断。
4. 当规则匹配时，通过钉钉 Webhook发送告警消息，包含异常详情。
5. 模拟异常流量，验证告警的及时性与准确性。

一、环境准备（约 20 分钟）

继续在log-consumer项目中工作，确保 Kafka、Redis、Nacos 等服务已运行。

docker-composeexecswoolebashcd/var/www/log-consumer

安装钉钉通知所需组件（之前用过 Guzzle，可直接使用hyperf/guzzle）：

composerrequire hyperf/guzzle hyperf/config-nacos

（Nacos 客户端若未安装需安装，用于动态规则）

二、知识核心：实时异常检测与规则引擎设计（约 1 小时）

1. 异常检测的几种方法

• 固定阈值：如 1 分钟内错误日志 > 100 条即告警，简单有效。
• 动态基线：根据历史数据计算均值，偏离一定程度告警（如环比增长 200%），更智能但复杂。
• 规则组合：例如“错误数 > 50 且 来自某个 IP 的请求 > 1000”。

我们今天实现固定阈值 + 可配置规则，规则存储在 Nacos，支持动态修改。规则示例：

{"rules":[{"name":"high_error_rate","description":"错误日志超过阈值","window":60,"field":"level","operator":"eq","value":"error","threshold":10},{"name":"single_ip_flood","description":"单IP高频访问","window":60,"field":"ip","operator":"count","threshold":1000}]}

2. 实时检测架构

• LogAlertConsumer消费raw_logs，对于每条日志，根据规则增加 Redis 计数器（使用INCR+EXPIRE创建滑动窗口计数）。
• 每次增加后，检查是否超过threshold，如果超过且未告警过（用 Redis 设置防重复），则触发告警。
• 告警通过 Guzzle 调用钉钉机器人 Webhook。

3. 防重复告警

为了避免同一异常在短期内重复发送，可以设置告警冷却时间：例如同一个规则在 5 分钟内只发送一次。使用 Redis 键alert_lock:{rule_name}设置 TTL 实现。

三、实战：实现异常检测与钉钉告警（约 2.5 小时）

步骤 1：配置 Nacos 规则

在 Nacos 控制台（dev 命名空间）创建配置log-alert-rules，内容为上面 JSON 示例。配置监听器将在服务启动时加载，并在变更时实时更新。

步骤 2：编写规则加载服务

新建app/Service/AlertRuleService.php：

<?phpnamespaceApp\Service;useHyperf\Contract\ConfigInterface;useHyperf\Di\Annotation\Inject;classAlertRuleService{#[Inject]privateConfigInterface$config;publicfunctiongetRules():array{return$this->config->get('alert_rules.rules',[]);}}

在config/autoload/alert_rules.php中定义默认规则（可被 Nacos 覆盖），或直接在config_center.php中配置 Nacos 监听。通过hyperf/config-nacos的动态配置，config('alert_rules')会实时更新。

步骤 3：创建异常检测消费者

新建app/Kafka/Consumer/LogAlertConsumer.php：

<?phpnamespaceApp\Kafka\Consumer;useApp\Service\AlertRuleService;useHyperf\Kafka\AbstractConsumer;useHyperf\Kafka\Annotation\Consumer;uselonglang\phpkafka\Consumer\ConsumeMessage;useHyperf\Kafka\Result;useHyperf\Redis\Redis;useHyperf\Di\Annotation\Inject;useHyperf\Guzzle\ClientFactory;#[Consumer(topic:'raw_logs',groupId:'log-alerts',name:'LogAlertConsumer',nums:1)]classLogAlertConsumerextendsAbstractConsumer{#[Inject]privateRedis$redis;#[Inject]privateAlertRuleService$ruleService;#[Inject]privateClientFactory$guzzleFactory;privatestring$dingtalkWebhook='https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN';publicfunctionconsume(ConsumeMessage$message):string{$data=json_decode($message->getValue(),true);if(!$data)returnResult::ACK;$rules=$this->ruleService->getRules();foreach($rulesas$rule){$this->evaluateRule($rule,$data);}returnResult::ACK;}privatefunctionevaluateRule(array$rule,array$log):void{$window=$rule['window']??60;$field=$rule['field']??'';$op=$rule['operator']??'';$threshold=$rule['threshold']??0;// 根据操作类型统计$count=0;if($op==='count'){// 统计某个字段的出现次数（如IP）$value=$log[$field]??'';if(empty($value))return;$key="alert:count:{$rule['name']}:{$value}:window";$count=$this->redis->incr($key);$this->redis->expire($key,$window);}elseif($op==='eq'){// 检查字段值是否等于指定值（如 level == error）$fieldValue=$log[$field]??'';if($fieldValue!==($rule['value']??''))return;$key="alert:count:{$rule['name']}:total:window";$count=$this->redis->incr($key);$this->redis->expire($key,$window);}// 超过阈值且未在冷却期内if($count>=$threshold){$this->triggerAlert($rule,$log,$count);}}privatefunctiontriggerAlert(array$rule,array$log,int$count):void{$lockKey='alert_lock:'.$rule['name'];if($this->redis->set($lockKey,1,['NX','EX'=>300])){// 5分钟冷却$message="【异常告警】规则:{$rule['description']}\n"."当前值:{$count}\n"."阈值:{$rule['threshold']}\n"."触发日志: ".json_encode($log,JSON_UNESCAPED_UNICODE);$this->sendDingtalk($message);}}privatefunctionsendDingtalk(string$text):void{$client=$this->guzzleFactory->create();try{$client->post($this->dingtalkWebhook,['json'=>['msgtype'=>'text','text'=>['content'=>$text],],]);}catch(\Throwable$e){// 记录失败日志}}}

注意：需要将钉钉机器人 Webhook 地址换成真实的（可在钉钉群添加机器人获得）。为了安全，建议将 token 放在.env中。

步骤 4：启动消费者并模拟异常

重启log-consumer，并向 Kafka 发送大量错误日志：

# 发送100条错误日志foriin{1..100};docurl-XPOST http://localhost:9600/api/logs-H"Content-Type: application/json"\-d'[{"service":"order","message":"payment failed","level":"error","ip":"10.0.0.1","timestamp":"'$(date-Iseconds)'"}]'done

1 分钟内错误数超过 10（规则中 threshold），应触发钉钉告警。检查钉钉群，应收到告警消息。

步骤 5：验证冷却机制

连续发送两批错误，第二批应在冷却期内被抑制，不会重复发送告警。5 分钟后再次发送，应再次告警。

步骤 6：动态修改规则

在 Nacos 中将high_error_rate的threshold改为 5，发布。稍等片刻，消费者无需重启，新的阈值生效。发送 6 条错误日志，立即告警。完美展示动态配置能力。

四、成果测试与验证（约 1 小时）

测试清单

监控与日志

在消费者日志中打印告警触发信息，方便排查。也可以通过 Prometheus 指标记录告警次数。

五、今日作业与学习产出

1. 提交代码：LogAlertConsumer、AlertRuleService、Nacos 配置文件。
2. 完善规则引擎：
   • 增加环比检测规则（需存储历史数据，可用 Redis 时间序列）。
   • 支持正则表达式匹配日志消息内容。
3. 学习笔记：
   • 对比实时异常检测与离线分析（如 ELK Watcher）的适用场景。
   • 画出从日志产生到告警的完整流程图，标出每个组件的处理延时。
4. 挑战任务：
   • 使用Redis Stream替代 Kafka 进行轻量级日志传输，并实现消费者组。
   • 对接Prometheus AlertManager统一告警出口，而非单独钉钉。

通过今天的学习，你的日志系统拥有了“免疫系统”，能主动发现并报告异常，向着智能运维（AIOps）迈出了重要一步。明天我们将对采集端进行性能优化，并构建一个完整的看板前端，让整个平台更加完整。