【基于 SwooleHyperf 的微服务实战】第十一周·周二构建异步日志采集与分析系统今天我们进入第十一周周二继续构建异步日志采集与分析系统。昨天我们实现了日志接收服务将日志投递到 Kafka。今天要打造系统的消费与存储能力编写 Kafka 消费者拉取原始日志解析富化后批量写入 Elasticsearch让日志可被搜索、分析和可视化为后续的实时统计打下数据基础。今日目标创建log-consumer服务配置 Kafka 消费者与 Elasticsearch 客户端。实现消费者拉取raw_logs消息解析User-Agent、IP归属地等字段丰富日志内容。使用批量写入的方式将日志索引到 Elasticsearch提升吞吐效率。验证日志能通过 Kibana 搜索并观察消费者组的并行处理能力。配置消费者 Offset 自动提交保证消息不丢失、不重复。一、环境准备约 30 分钟1. 启动依赖确保 Kafka、Elasticsearch、Kibana 容器运行Elasticsearch 已在第六周部署若未部署请参考docker-compose.yml添加。确认 Kibana 可访问http://localhost:5601。2. 创建消费者项目docker-composeexecswoolebashcd/var/wwwcomposercreate-project hyperf/hyperf-skeleton log-consumercdlog-consumercomposerrequire hyperf/kafka elasticsearch/elasticsearch hyperf/redis hyperf/metric端口设为9601项目专注于消费可以不需要 HTTP 服务或保留简单健康检查。3. 配置 Kafka 消费者编辑config/autoload/kafka.php增加消费者配置consumers[[topicraw_logs,group_idlog-processors,client_idlog-consumer-1,auto_commitfalse,// 手动提交 offset保证处理完再提交enable_auto_commitfalse,max_poll_records500,// 一次拉取最大条数consume_timeout3000,],],4. 配置 Elasticsearch创建config/autoload/elasticsearch.php?phpreturn[hosts[env(ES_HOST,elasticsearch:9200)],retries2,handler\Hyperf\Guzzle\RingPHP\CoroutineHandler::class,// 协程化 HTTP 客户端];二、知识核心Kafka 消费者组、日志富化与 ES 批量写入约 1 小时1. Kafka 消费者组与分区同一个消费者组内的多个实例或协程会分摊 Topic 的分区实现并行消费。组 ID 必须相同。增加消费者实例时Kafka 会触发 Rebalance重新分配分区。分区数应 消费者实例数才能充分利用并发。Offset 管理我们采用手动提交在处理完一批消息后统一提交确保消息被成功写入 ES 后才标记已消费。2. 日志富化Enrichment原始日志只包含基础信息我们可以添加User-Agent 解析提取浏览器、操作系统、设备类型。可以使用whichbrowser/parser或ua-parser/uap-php库。IP 归属地利用离线库如 GeoIP2或调用内部服务添加国家、城市、ISP。时间戳标准化统一为 ISO 8601 格式方便 ES 的时间字段索引。业务字段补齐根据日志中的service或trace_id关联额外元数据。3. Elasticsearch 批量索引逐条插入 ES 效率极低。我们收集一批消息如 500 条使用BulkAPI 一次性写入可大幅降低网络往返。ES 客户端提供了bulk()方法我们只需构造[body [...actions...]]。三、实战消费者实现与 ES 写入约 2.5 小时步骤 1安装日志富化库composerrequire uap-php/uap-php geoip2/geoip2准备 GeoIP2 数据库如 GeoLite2-City.mmdb放入项目storage/目录可从 MaxMind 免费下载。步骤 2创建消费者类新建app/Kafka/Consumer/LogConsumer.php?phpnamespaceApp\Kafka\Consumer;useHyperf\Kafka\AbstractConsumer;useHyperf\Kafka\Annotation\Consumer;uselonglang\phpkafka\Consumer\ConsumeMessage;useHyperf\Kafka\Result;useHyperf\Di\Annotation\Inject;useElasticsearch\ClientBuilder;#[Consumer(topic:raw_logs,groupId:log-processors,name:LogConsumer,nums:4)]classLogConsumerextendsAbstractConsumer{#[Inject]private\Hyperf\Contract\ConfigInterface$config;private$esClient;private$uaParser;private$geoReader;publicfunction__construct(){$this-esClientClientBuilder::create()-setHosts($this-config-get(elasticsearch.hosts))-build();$this-uaParser\UAParser\Parser::create();// 加载 GeoIP 数据库$this-geoReadernew\GeoIp2\Database\Reader(BASE_PATH./storage/GeoLite2-City.mmdb);}publicfunctionconsume(ConsumeMessage$message):string{$datajson_decode($message-getValue(),true);if(!$data)returnResult::ACK;// 富化if(!empty($data[user_agent])){$ua$this-uaParser-parse($data[user_agent]);$data[browser]$ua-ua-family. .$ua-ua-toVersion();$data[os]$ua-os-family. .$ua-os-toVersion();}if(!empty($data[ip])){try{$record$this-geoReader-city($data[ip]);$data[geo][country$record-country-name,city$record-city-name,lat$record-location-latitude,lon$record-location-longitude,];}catch(\Exception$e){}}// 存入待批量写入的缓冲区使用协程安全的 Channel 或 Redis List// 简单起见每条消息直接记录到内存数组由定时器批量写入本章后面实现// 但消费者是多协程并发需要共享安全队列。我们使用 Swoole\Coroutine\Channel// 在进程启动时创建全局 Channel消费者 push另一个协程 pop 批量写入 ES。// 由于时间关系今天先实现逐条写入性能差明天改造批量。$params[indexapp-logs-.date(Y.m.d),body$data,];$this-esClient-index($params);returnResult::ACK;}}注意逐条写入 ES 性能较差仅用于今天的功能验证。明天我们将用协程 Channel 定时器实现批量写入。步骤 3配置消费者进程与并发在config/autoload/processes.php中添加\Hyperf\Kafka\Process\ConsumerProcess::class。消费者nums: 4表示启动 4 个协程并发消费。步骤 4启动消费者并测试启动服务php bin/hyperf.php start发送日志通过昨天的采集服务或直接向 Kafka 发送测试数据curl-XPOST http://localhost:9600/api/logs-HContent-Type: application/json\-d[{service:gateway,message:request,ip:8.8.8.8,user_agent:Mozilla/5.0 ...,timestamp:2025-01-01T00:00:00}]观察消费者日志确认无异常。然后在 Kibana 中创建索引模式app-logs-*在 Discover 中搜索应能看到富化后的日志包含browser,os,geo等字段。步骤 5验证 Offset 提交消费者抛出异常时消息不会被确认会重新消费。我们可以手动制造异常如暂时停止 ES消息会堆积但不会丢失待 ES 恢复后继续写入。auto_commit设为 false由ConsumerProcess自动在成功返回后提交。四、成果测试与验证约 1 小时测试清单检验项方法通过标准Kafka 消费者连接服务启动日志显示Kafka consumer started日志富化Kibana 查看文档包含browser,os,geo字段多分区并行设置 6 分区nums4日志分布均匀4 个协程各自消费不同分区ES 索引自动创建每天新索引app-logs-YYYY.MM.DD索引自动出现消息不丢失停止消费者积累消息重启后继续消费Kibana 文档数等于发送数错误重试写入 ES 时故意返回错误如端口错消息重新消费不会被跳过性能观察尽管逐条写入较慢但当前功能已跑通。记录消费速率约 200 条/秒为明天批量优化提供基准。五、今日作业与学习产出提交代码消费者服务及相关配置。完善富化添加referer解析提取来源网站。尝试关联trace_id从缓存获取额外上下文。学习笔记画出 Kafka 消费者组与分区分配示意图。解释为什么批量写入 ES 能显著提高吞吐。挑战任务实现死信处理无法解析的日志写入 Kafka 死信 Topic单独监控。使用Elasticsearch Ingest Pipeline替代部分 PHP 富化如 GeoIP减轻消费者负担。今天你成功将原始日志变成了结构化、可搜索的数据资产。明天我们将实现批量写入优化和实时统计让日志分析系统更具实用价值。