每次迭代返回一行（不含换行符） * * @throws \RuntimeException 文件无法打开或 stat/fseek/fread 失败时抛出（交由上层 try-catch） */ public function read(string $filePath, int $maxLines = 100000): \Generator { // per-run 状态复位（同一个 reader 实例可串行处理多个文件） $this->reachedEof = false; $this->droppedLines = 0; $this->currentLineHash = null; $fp = @fopen($filePath, 'rb'); if ($fp === false) { throw new \RuntimeException("XhprofLogReaderService: 无法打开文件 {$filePath}"); } try { clearstatcache(true, $filePath); $stat = fstat($fp); if ($stat === false) { throw new \RuntimeException("XhprofLogReaderService: fstat 失败 {$filePath}"); } $currentInode = (int) $stat['ino']; $currentSize = (int) $stat['size']; $this->currentInode = $currentInode; // 读 position（无则默认 offset=0） $offset = 0; $position = $this->getLastPosition($filePath); if ($position !== null) { $offset = (int) $position['offset']; // 轮转检测：inode 变化（同日文件被重建）OR size < offset（防御性归零） if ((int) $position['inode'] !== $currentInode || $currentSize < $offset) { $offset = 0; } } $this->currentOffset = $offset; if ($offset > 0 && fseek($fp, $offset) !== 0) { throw new \RuntimeException("XhprofLogReaderService: fseek 失败 offset={$offset} file={$filePath}"); } $yielded = 0; $buffer = ''; // $bufferOffset 始终等于 buffer[0] 对应的文件字节偏移 $bufferOffset = $offset; while (true) { $chunk = fread($fp, self::BUFFER_SIZE); if ($chunk === false) { throw new \RuntimeException("XhprofLogReaderService: fread 失败 file={$filePath}"); } if ($chunk === '') { // EOF：buffer 为空说明干净收尾；有半行则 currentOffset 已停在最后完整行末尾 $this->reachedEof = ($buffer === ''); return; } $buffer .= $chunk; // 内层循环：处理 buffer 中所有完整行（含末尾 \n 的） $pos = 0; while (($nlPos = strpos($buffer, "\n", $pos)) !== false) { $line = substr($buffer, $pos, $nlPos - $pos); $pos = $nlPos + 1; // 先推进安全偏移再 yield：消费方在事务内 savePosition 拿到的 // 一定是"已 yield 行末尾"，与该事务内入库的数据严格对齐 $this->currentOffset = $bufferOffset + $pos; $this->currentLineHash = md5($line); yield $line; $yielded++; if ($yielded >= $maxLines) { return; } } // buffer 中已消费 $pos 字节，半行（buffer[$pos..]）保留 $bufferOffset += $pos; $buffer = substr($buffer, $pos); // 超长行防御：整行丢弃（禁止强制切行——会把一条 JSON 切成两条坏行） if (strlen($buffer) >= self::MAX_LINE_BYTES) { $this->droppedLines++; $buffer = ''; // 扫描至下一个换行符，将其后内容作为新 buffer 继续 while (true) { $chunk = fread($fp, self::BUFFER_SIZE); if ($chunk === false || $chunk === '') { // 至 EOF 仍未等到换行符：整个尾部按已丢弃处理，偏移推到文件末尾 $bufferOffset = (int) ftell($fp); $this->currentOffset = $bufferOffset; $this->reachedEof = true; return; } $nl = strpos($chunk, "\n"); if ($nl !== false) { $bufferOffset = (int) ftell($fp) - (strlen($chunk) - $nl - 1); $this->currentOffset = $bufferOffset; $buffer = substr($chunk, $nl + 1); break; } } } } } finally { if (is_resource($fp)) { fclose($fp); } } } /** * 当前安全读取位置（read 期间/结束后可调）. * * @return array{inode: int, offset: int, last_line_hash: string|null} */ public function getCurrentPosition(): array { return [ 'inode' => $this->currentInode, 'offset' => $this->currentOffset, 'last_line_hash' => $this->currentLineHash, ]; } /** * 本轮 read 是否干净到达 EOF（尾部无半行）. * * 导入侧据此判断"文件已导完"：reachedEof 且 offset >= filesize 才允许 unlink。 */ public function reachedEof(): bool { return $this->reachedEof; } /** * 本轮 read 丢弃的超长行数（由导入侧并入 fails 统计）. */ public function getDroppedLines(): int { return $this->droppedLines; } /** * 查询文件的上次读取位置（按 node_id 隔离）. * * @return array|null 命中时返回 [inode, offset, last_line_hash, parse_fail_count, parse_fail_samples]，无记录返回 null */ public function getLastPosition(string $filePath): ?array { $nodeId = HostService::getNodeId(); $row = XhprofLogPosition::where('file_path', $filePath) ->where('node_id', $nodeId) ->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，按 node_id 隔离）. * * 由导入任务在"insertBatch 同一事务"内显式调用（本类不自动保存）， * 保证崩溃时入库与进度要么同时生效、要么同时回滚，消重复导入窗口。 * * @param string $filePath jsonl 文件绝对路径 * @param int $inode 当前文件 inode * @param int $offset 新的字节偏移（最后一条完整行末尾） * @param string|null $lastLineHash 最后一条已入库行 md5 * @param int $failCount 本批解析失败累计次数 * @param string|null $failSamples 解析失败样本（json 或 null） */ public function savePosition(string $filePath, int $inode, int $offset, ?string $lastLineHash, int $failCount = 0, ?string $failSamples = null): void { $now = time(); $nodeId = HostService::getNodeId(); $existing = XhprofLogPosition::where('file_path', $filePath) ->where('node_id', $nodeId) ->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; } XhprofLogPosition::create([ 'node_id' => $nodeId, '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, ]); } }