From 9fe55f81bf438e4c013bd409610f436c20d5bbd3 Mon Sep 17 00:00:00 2001 From: augushong Date: Tue, 28 Jul 2026 00:39:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(nginx-log):=20NginxLogReader=20=E5=A2=9E?= =?UTF-8?q?=E9=87=8F=E8=AF=BB=E5=8F=96=20service=EF=BC=88Base/App=20?= =?UTF-8?q?=E5=8F=8C=E5=B1=82=EF=BC=8C=E5=90=AB=E8=BD=AE=E8=BD=AC=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/common/service/NginxLogReader.php | 18 ++ .../common/service/NginxLogReaderBase.php | 226 ++++++++++++++++++ 2 files changed, 244 insertions(+) create mode 100644 app/common/service/NginxLogReader.php create mode 100644 extend/base/common/service/NginxLogReaderBase.php diff --git a/app/common/service/NginxLogReader.php b/app/common/service/NginxLogReader.php new file mode 100644 index 0000000..440a1b5 --- /dev/null +++ b/app/common/service/NginxLogReader.php @@ -0,0 +1,18 @@ + 每次迭代返回一行（不含换行符） + * + * @throws \RuntimeException 文件无法打开或 stat/fseek 失败时抛出（交由上层 try-catch） + */ + public function read(string $filePath, int $maxLines = 100000): \Generator + { + $fp = @fopen($filePath, 'rb'); + if ($fp === false) { + throw new \RuntimeException("NginxLogReader: 无法打开文件 {$filePath}"); + } + + try { + // 步骤 1：clearstatcache + fstat 取当前 inode/size + clearstatcache(true, $filePath); + $stat = fstat($fp); + if ($stat === false) { + throw new \RuntimeException("NginxLogReader: fstat 失败 {$filePath}"); + } + $currentInode = (int) $stat['ino']; + $currentSize = (int) $stat['size']; + + // 步骤 2：读 position（无则默认 offset=0） + $position = $this->getLastPosition($filePath); + $inode = $currentInode; + $offset = 0; + if ($position !== null) { + $inode = (int) $position['inode']; + $offset = (int) $position['offset']; + + // 步骤 3：轮转检测双条件 + // inode 不一致 → rename+rebuild 模式（Linux inode 有效时主判定） + // filesize < offset → copytruncate 模式 / Windows 上 inode 恒为 0 时的主判定 + if ($inode !== $currentInode || $currentSize < $offset) { + $offset = 0; + $inode = $currentInode; + } + } + + // 步骤 4：fseek 到 offset + if ($offset > 0 && fseek($fp, $offset) !== 0) { + throw new \RuntimeException("NginxLogReader: fseek 失败 offset={$offset} file={$filePath}"); + } + + $yielded = 0; + $lastLineHash = null; + $buffer = ''; + // $bufferOffset 始终等于 buffer[0] 对应的文件字节偏移 + $bufferOffset = $offset; + + // 步骤 5-7：循环读取 + yield + while (true) { + $chunk = fread($fp, self::BUFFER_SIZE); + if ($chunk === false) { + // 读错误：交上层处理，不保存进度（下次重读） + throw new \RuntimeException("NginxLogReader: fread 失败 file={$filePath}"); + } + if ($chunk === '') { + // EOF + break; + } + $buffer .= $chunk; + + // 内层循环：处理 buffer 中所有完整行（含末尾 \n 的） + $pos = 0; + while (($nlPos = strpos($buffer, "\n", $pos)) !== false) { + $line = substr($buffer, $pos, $nlPos - $pos); + $pos = $nlPos + 1; + + yield $line; + $yielded++; + $lastLineHash = md5($line); + + // 步骤 7：maxLines 达到即保存并退出 + if ($yielded >= $maxLines) { + $newOffset = $bufferOffset + $pos; + $this->savePosition($filePath, $inode, $newOffset, $lastLineHash); + return; + } + } + + // 步骤 6：buffer 中已消费 $pos 字节，半行（buffer[$pos..]）保留 + $bufferOffset += $pos; + $buffer = substr($buffer, $pos); + + // 防御：buffer 无 \n 累积过长（异常畸形行），强制按一行处理避免 OOM + if (strlen($buffer) >= self::MAX_LINE_BYTES) { + yield $buffer; + $yielded++; + $lastLineHash = md5($buffer); + $bufferOffset += strlen($buffer); + $buffer = ''; + + if ($yielded >= $maxLines) { + $this->savePosition($filePath, $inode, $bufferOffset, $lastLineHash); + return; + } + } + } + + // 步骤 8：EOF 处理 + if ($buffer === '') { + // 文件恰好在最后一个 \n 处结束：offset = 文件末尾（ftell） + $finalOffset = ftell($fp); + $this->savePosition($filePath, $inode, (int) $finalOffset, $lastLineHash); + } else { + // 尾部半行（无 \n）：可能是日志正在写入，不 yield，offset 回退到半行起点，下次重读 + $this->savePosition($filePath, $inode, $bufferOffset, $lastLineHash); + } + } finally { + if (is_resource($fp)) { + fclose($fp); + } + } + } + + /** + * 查询文件的上次读取位置. + * + * @param string $filePath 日志文件绝对路径 + * + * @return array|null 命中时返回 [inode, offset, last_line_hash, parse_fail_count, parse_fail_samples]，无记录返回 null + */ + public function getLastPosition(string $filePath): ?array + { + $row = NginxLogPosition::where('file_path', $filePath)->find(); + if ($row === null) { + return null; + } + + return [ + 'inode' => (int) $row->getData('inode'), + 'offset' => (int) $row->getData('offset'), + 'last_line_hash' => $row->getData('last_line_hash'), + 'parse_fail_count' => (int) $row->getData('parse_fail_count'), + 'parse_fail_samples' => $row->getData('parse_fail_samples'), + ]; + } + + /** + * 保存读取位置（upsert）. + * + * Reader 内部在 maxLines 达到或 EOF 时调用，传 lastLineHash（最后一行 md5）。 + * failCount 默认 0：每次新读取批次开始时由 Reader 重置；上层若需记录解析失败， + * 在 parse 后再次调用本方法覆盖 failCount/failSamples 即可（offset 保持一致）。 + * + * @param string $filePath 日志文件绝对路径（主键） + * @param int $inode 当前文件 inode + * @param int $offset 新的字节偏移 + * @param string|null $lastLineHash 最后一行 md5（用于去重/检测） + * @param int $failCount 解析失败累计次数（默认 0） + * @param string|null $failSamples 解析失败样本（默认 null） + */ + public function savePosition(string $filePath, int $inode, int $offset, ?string $lastLineHash, int $failCount = 0, ?string $failSamples = null): void + { + $now = time(); + + $existing = NginxLogPosition::where('file_path', $filePath)->find(); + if ($existing !== null) { + $existing->save([ + 'inode' => $inode, + 'offset' => $offset, + 'last_read_time' => $now, + 'last_line_hash' => $lastLineHash, + 'parse_fail_count' => $failCount, + 'parse_fail_samples' => $failSamples, + ]); + + return; + } + + NginxLogPosition::create([ + 'file_path' => $filePath, + 'inode' => $inode, + 'offset' => $offset, + 'last_read_time' => $now, + 'last_line_hash' => $lastLineHash, + 'parse_fail_count' => $failCount, + 'parse_fail_samples' => $failSamples, + 'create_time' => $now, + 'update_time' => $now, + ]); + } +}