【基于 Swoole+Hyperf 的微服务实战】第十一周·周一：异步日志采集与实时分析系统

【基于 SwooleHyperf 的微服务实战】第十一周·周一异步日志采集与实时分析系统今天我们进入第十一周周一正式开启第二个综合项目——异步日志采集与实时分析系统。在第一个电商项目中我们重点攻克了微服务间的同步通信与分布式事务而在这个项目中我们将面对海量数据流的采集、传输、存储与分析挑战。今天作为项目奠基日我们将完成系统设计、环境搭建并实现核心的日志接收与 Kafka 投递链路确保日志不丢失、高吞吐。今日目标理解异步日志采集系统的典型架构与 Kafka 的核心作用。使用 Hyperf 构建日志接收服务对外提供 HTTP 接口接收日志。配置 Kafka 生产者将接收到的日志异步批量发送到 Kafka Topic。验证在高并发请求下日志不丢失Kafka 无数据落差。搭建项目骨架为明天的消费与存储奠定基础。一、项目背景与架构设计约 30 分钟1. 业务场景我们需要为整个微服务集群如电商系统构建一个统一的日志采集与分析平台。各服务将日志发送到统一的采集服务而非各自写文件。采集服务将日志异步投递到 Kafka后续由消费者进行实时分析如 UV/PV 统计、异常检测并存储到 Elasticsearch 或 ClickHouse。2. 系统架构图[业务服务] --HTTP/RSocket-- [日志采集服务 (Hyperf)] --Kafka-- [日志消费者 (Hyperf)] --批量写入-- [Elasticsearch / ClickHouse] \ - [Redis] (实时统计)核心组件日志采集服务暴露 HTTP 接口接收 JSON 日志校验后异步投递 Kafka。Kafka高吞吐消息中间件缓冲日志数据。日志消费者从 Kafka 拉取日志解析、富化后批量写入存储。存储与分析Elasticsearch Kibana 用于全文检索ClickHouse 用于聚合分析Redis 用于实时指标。3. 今日聚焦创建log-collector服务实现日志接收端点POST /api/logs。集成hyperf/kafka配置生产者将日志推送到raw_logsTopic。使用wrk进行压力测试同时监控 Kafka 消息数确保零丢失。二、环境准备约 30 分钟1. 启动 Kafka沿用之前的docker-compose.yml确保kafka容器已运行。如果没有从电商项目复制或新增。确保 Kafka 监听 9092 端口。2. 创建项目docker-composeexecswoolebashcd/var/wwwcomposercreate-project hyperf/hyperf-skeleton log-collectorcdlog-collectorcomposerrequire hyperf/kafka hyperf/redis hyperf/metric端口设为 9600避免与其他服务冲突。3. 配置 Kafka编辑config/autoload/kafka.php?phpreturn[default[hostenv(KAFKA_HOST,kafka),port(int)env(KAFKA_PORT,9092),pool[min_connections1,max_connections10,connect_timeout10.0,wait_timeout3.0,max_idle_time60,],],];4. 创建 Kafka Topicdockerexec-itkafka-lab kafka-topics--create--topicraw_logs --bootstrap-server kafka:9092--partitions6--replication-factor1设置 6 个分区便于并行消费。三、知识核心异步日志采集与 Kafka 生产者优化约 1 小时1. 日志采集接口设计为了高吞吐日志接口应极简接收 JSON 数组或单条 JSON支持批量提交。快速校验格式、必填字段后立即返回 200不进行复杂处理。通过协程或队列异步将日志发送到 Kafka不阻塞客户端。对于发送失败可以内置重试或降级写入本地文件后续补偿。2. Kafka 生产者关键参数acks-1all保证最强持久性不会丢数据但吞吐略降1为领导分区确认0不等待确认。根据需求我们选择1平衡或-1。compression.type启用snappy或gzip压缩节省带宽。batch.size和linger.ms批量发送提高吞吐。hyperf/kafka基于longlang/phpkafka支持这些配置。我们将在生产者配置中设置。3. 协程化与连接池所有 Kafka 发送操作必须是协程安全的并复用连接。hyperf/kafka已内置协程支持和连接池通过Pool::get()获取连接。四、实战构建日志接收与 Kafka 投递约 2.5 小时步骤 1创建日志接收控制器创建app/Controller/LogController.php?phpnamespaceApp\Controller;useApp\Kafka\Producer\LogProducer;useHyperf\HttpServer\Annotation\Controller;useHyperf\HttpServer\Annotation\RequestMapping;useHyperf\Di\Annotation\Inject;#[Controller(prefix:/api)]classLogControllerextendsAbstractController{#[Inject]privateLogProducer$logProducer;#[RequestMapping(path:logs,methods:post)]publicfunctioncollect(){$body$this-request-getBody()-getContents();$logsjson_decode($body,true);if(empty($logs)){return$this-response-json([code400,messageInvalid JSON])-withStatus(400);}// 支持单条或批量if(isset($logs[timestamp])){$logs[$logs];}$count0;foreach($logsas$log){// 快速校验if(empty($log[service])||empty($log[message])){continue;}// 补充接收时间戳$log[received_at]date(Y-m-d\TH:i:s.v);// 异步发送到 Kafka$this-logProducer-sendAsync(json_encode($log),$log[service]??unknown);$count;}return[code200,messageAccepted{$count}logs];}}步骤 2创建 Kafka 生产者新建app/Kafka/Producer/LogProducer.php?phpnamespaceApp\Kafka\Producer;useHyperf\Kafka\AbstractProducer;useHyperf\Kafka\Annotation\Producer;#[Producer(topic:raw_logs,name:LogProducer)]classLogProducerextendsAbstractProducer{// sendAsync 方法由基类提供第二个参数为消息 key}步骤 3配置生产者选项压缩、acks在config/autoload/kafka.php中增加生产者配置producers[LogProducer[compressionsnappy,acks1,// 领导确认batch.size16384,linger.ms10,],],hyperf/kafka会将这些配置注入到生产者实例。步骤 4添加批量发送优化可选sendAsync默认是异步非阻塞但高频调用时可能产生大量协程切换。Hyperf 的 Kafka 生产者支持批量发送可以将多个消息放入缓冲区定时或满批量后统一发送。为简单我们今天先用单条异步明天在消费者端优化。步骤 5启动服务并验证 Kafka 投递php bin/hyperf.php start发送测试日志curl-XPOST http://localhost:9600/api/logs\-HContent-Type: application/json\-d[ {service:user,message:用户登录,level:info,timestamp:2025-01-01T00:00:00}, {service:order,message:订单创建,level:info,timestamp:2025-01-01T00:00:01} ]检查 Kafka 消息dockerexec-itkafka-lab kafka-console-consumer --bootstrap-server kafka:9092--topicraw_logs --from-beginning --max-messages2应看到两条 JSON 日志。步骤 6压测验证不丢数据使用wrk发送大量日志# 准备一个包含20条日志的JSON数组文件 body.json# 使用 wrk 的 Lua 脚本动态生成请求体略# 简化用 ab 发送固定 bodyab-n50000-c200-pbatch_20_logs.json-Tapplication/jsonhttp://localhost:9600/api/logs同时在压测前后查看 Kafka Topic 的消息数量dockerexec-itkafka-lab kafka-run-class kafka.tools.GetOffsetShell --broker-list kafka:9092--topicraw_logs--time-1计算压测期间新增的消息数应等于请求数 * 每条请求包含的日志数减去校验忽略的无丢失。观察采集服务的 CPU、内存确保没有因为异步队列爆满导致 OOM。hyperf/kafka内部有缓冲区如果发送失败会重试我们可以在生产代码中捕获异常并记录失败日志。五、成果测试与验证约 1 小时测试清单检验项方法通过标准日志接收接口返回 200curl发送 JSON返回 accepted count 正确单条日志投递 Kafkakafka-console-consumer消费出现日志内容批量日志投递发送数组Kafka 收到多条数量匹配高并发压测无丢失ab 50000 请求后检查 Kafka offset增量 请求内日志总数错误日志处理发送非法 JSON、缺少字段返回 400 或忽略不影响其他日志生产者连接池长时间压测无连接耗尽异常稳定观察点如果 Kafka 暂时不可用生产者会抛出异常日志丢失。我们可以在LogProducer中增加try-catch失败时写入本地文件队列由定时任务重试。sendAsync是否真正异步可以在控制器中记录时间返回极快证明没有阻塞在 Kafka 网络 IO。六、今日作业与学习产出提交代码日志采集服务及 Kafka 配置。增强健壮性增加失败日志本地写入JSON 行文件定时任务重新发送。为采集接口添加限流防止恶意刷日志。学习笔记画出日志采集系统的数据流向图各服务 - 采集服务 - Kafka - 消费者 - 存储。对比 Kafka 与 RabbitMQ 在日志采集场景下的优劣势。挑战任务使用RSocket协议代替 HTTP 接收日志提升性能需要hyperf/rsocket。配置Kafka 多节点集群测试分区 leader 切换时生产者的容错。通过今天的学习你已经为日志分析系统搭建好了数据入口管道。明天我们将编写 Kafka 消费者解析日志并存入 Elasticsearch实现日志的集中存储与搜索。