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

今日目标

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

在这里插入图片描述

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

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

docker-compose exec swoole bash
cd /var/www/log-consumer

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

composer require 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：

<?php
namespace App\Service;

use Hyperf\Contract\ConfigInterface;
use Hyperf\Di\Annotation\Inject;

class AlertRuleService
{
    #[Inject]
    private ConfigInterface $config;

    public function getRules(): 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：

<?php
namespace App\Kafka\Consumer;

use App\Service\AlertRuleService;
use Hyperf\Kafka\AbstractConsumer;
use Hyperf\Kafka\Annotation\Consumer;
use longlang\phpkafka\Consumer\ConsumeMessage;
use Hyperf\Kafka\Result;
use Hyperf\Redis\Redis;
use Hyperf\Di\Annotation\Inject;
use Hyperf\Guzzle\ClientFactory;

#[Consumer(topic: 'raw_logs', groupId: 'log-alerts', name: 'LogAlertConsumer', nums: 1)]
class LogAlertConsumer extends AbstractConsumer
{
    #[Inject]
    private Redis $redis;

    #[Inject]
    private AlertRuleService $ruleService;

    #[Inject]
    private ClientFactory $guzzleFactory;

    private string $dingtalkWebhook = 'https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN';

    public function consume(ConsumeMessage $message): string
    {
        $data = json_decode($message->getValue(), true);
        if (!$data) return Result::ACK;

        $rules = $this->ruleService->getRules();
        foreach ($rules as $rule) {
            $this->evaluateRule($rule, $data);
        }
        return Result::ACK;
    }

    private function evaluateRule(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);
        }
    }

    private function triggerAlert(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);
        }
    }

    private function sendDingtalk(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条错误日志
for i in {1..100}; do
  curl -X POST 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 小时）

测试清单
检验项方法通过标准
阈值告警触发发送超过阈值的错误日志钉钉收到告警
冷却期抑制再次发送同样异常，5分钟内无重复告警
不同规则独立工作触发 IP 洪水规则钉钉收到不同告警
动态规则生效修改 Nacos 阈值，发送少量日志新阈值下触发告警
规则配置异常容错错误 JSON 规则，消费者不崩溃跳过非法规则，正常消费
无异常时无告警发送正常日志无告警消息
监控与日志

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

五、今日作业与学习产出

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

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

# swoole # 微服务 # 后端 # 架构 # php
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

六项评测超越GPT-6 Astra！大晓ACE-Brain-0.6以数据Scaling推动具身空间智能跃迁

avatar DAMO开发者矩阵
cover

可量化能力验证：从机器人秋招到模拟交易训练的技术逻辑

avatar DAMO开发者矩阵

企业微信 API 接口能解决哪些重复工作？常见自动化场景

企微 API 最务实的价值不是搭炫酷的机器人，而是把团队每天在做的那几件重复活自动化。这篇不讲功能清单，讲五个真实团队在做的自动化场景，每个场景给一句"怎么做"。

avatar DAMO开发者矩阵