PHP项目同步中断如何断点续传恢复任务

wen PHP项目 2026-06-26 255

本文目录导读：

PHP项目同步中断如何断点续传恢复任务

文件上传断点续传
任务断点续传（如大文件处理/数据导入）
核心优化策略
最佳实践总结

在PHP项目中实现断点续传恢复任务，主要涉及文件上传断点续传和任务（如大文件处理/导入）断点续传两种情况,以下是针对这两种场景的详细实现方案：

文件上传断点续传

前端处理

使用File.slice()分片上传,记录已上传分片：

// 前端分片上传示例
const CHUNK_SIZE = 1024 * 1024; // 1MB
function uploadFile(file) {
    let start = 0;
    const fileId = generateFileId(file); // 唯一标识
    // 检查已上传进度
    fetch(`/check-progress?file_id=${fileId}`)
        .then(res => res.json())
        .then(data => {
            start = data.uploadedSize || 0;
            uploadChunks(file, start, fileId);
        });
}
function uploadChunks(file, start, fileId) {
    const chunk = file.slice(start, start + CHUNK_SIZE);
    const formData = new FormData();
    formData.append('file', chunk);
    formData.append('file_id', fileId);
    formData.append('chunk_index', Math.floor(start / CHUNK_SIZE));
    fetch('/upload-chunk', { method: 'POST', body: formData })
        .then(res => res.json())
        .then(data => {
            if (data.complete) {
                mergeChunks(fileId);
            } else {
                uploadChunks(file, start + CHUNK_SIZE, fileId);
            }
        });
}
function mergeChunks(fileId) {
    fetch('/merge-chunks', {
        method: 'POST',
        body: JSON.stringify({ file_id: fileId })
    });
}

后端PHP实现

检查进度接口

// /check-progress
public function checkProgress() {
    $fileId = $_GET['file_id'];
    $uploadDir = '/tmp/chunks/' . $fileId;
    $uploadedSize = 0;
    if (is_dir($uploadDir)) {
        $files = scandir($uploadDir);
        foreach ($files as $file) {
            if ($file !== '.' && $file !== '..') {
                $uploadedSize += filesize($uploadDir . '/' . $file);
            }
        }
    }
    return json_encode([
        'uploadedSize' => $uploadedSize,
        'fileId' => $fileId
    ]);
}

上传分片接口

// /upload-chunk
public function uploadChunk() {
    $fileId = $_POST['file_id'];
    $chunkIndex = $_POST['chunk_index'];
    $uploadDir = '/tmp/chunks/' . $fileId;
    if (!is_dir($uploadDir)) {
        mkdir($uploadDir, 0777, true);
    }
    $chunkFile = $uploadDir . '/' . $chunkIndex;
    move_uploaded_file($_FILES['file']['tmp_name'], $chunkFile);
    return json_encode([
        'success' => true,
        'complete' => $this->isUploadComplete($fileId)
    ]);
}
private function isUploadComplete($fileId) {
    // 根据业务逻辑判断是否所有分片已上传
    // 例如检查分片数量和总大小
}

合并分片接口

// /merge-chunks
public function mergeChunks() {
    $data = json_decode(file_get_contents('php://input'), true);
    $fileId = $data['file_id'];
    $chunkDir = '/tmp/chunks/' . $fileId;
    $finalPath = '/uploads/' . $fileId . '_final.zip';
    $chunks = scandir($chunkDir);
    sort($chunks); // 按顺序合并
    $finalFile = fopen($finalPath, 'wb');
    foreach ($chunks as $chunk) {
        if ($chunk === '.' || $chunk === '..') continue;
        $chunkContent = file_get_contents($chunkDir . '/' . $chunk);
        fwrite($finalFile, $chunkContent);
    }
    fclose($finalFile);
    // 清理临时文件
    array_map('unlink', glob($chunkDir . '/*'));
    rmdir($chunkDir);
    return json_encode(['success' => true]);
}

任务断点续传（如大文件处理/数据导入）

任务状态记录方案

使用数据库或文件记录任务进度：

CREATE TABLE task_progress (
    task_id VARCHAR(64) PRIMARY KEY,
    file_id VARCHAR(64),
    total_rows INT DEFAULT 0,
    processed_rows INT DEFAULT 0,
    current_offset BIGINT DEFAULT 0,
    status ENUM('pending','processing','paused','completed','failed') DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

任务执行引擎

class ResumeableTask {
    private $taskId;
    private $filePath;
    private $chunkSize = 1000; // 每次处理1000行
    public function __construct($taskId, $filePath) {
        $this->taskId = $taskId;
        $this->filePath = $filePath;
    }
    public function process() {
        // 获取当前进度
        $progress = $this->getProgress();
        // 打开文件并跳过已处理部分
        $handle = fopen($this->filePath, 'r');
        fseek($handle, $progress['current_offset']);
        $rowCount = 0;
        while (($line = fgets($handle)) !== false) {
            // 处理每一行
            $this->processLine($line);
            $rowCount++;
            $progress['processed_rows']++;
            // 每处理chunkSize行更新一次进度
            if ($rowCount % $this->chunkSize === 0) {
                $progress['current_offset'] = ftell($handle);
                $this->updateProgress($progress);
                // 检查是否暂停
                if ($this->shouldPause()) {
                    break;
                }
            }
        }
        fclose($handle);
        // 任务完成
        $this->markCompleted($progress);
    }
    // 断点恢复
    public function resume() {
        $progress = $this->getProgress();
        if ($progress['status'] === 'paused' || $progress['status'] === 'failed') {
            $this->process();
            return true;
        }
        return false;
    }
    private function getProgress() {
        // 从数据库或文件读取进度
        return DB::table('task_progress')
            ->where('task_id', $this->taskId)
            ->first();
    }
    private function updateProgress($progress) {
        DB::table('task_progress')
            ->where('task_id', $this->taskId)
            ->update($progress);
    }
    private function shouldPause() {
        // 检查外部信号（如文件锁、Redis标志等）
        return file_exists("/tmp/pause_{$this->taskId}.flag");
    }
}

暂停/恢复API

// 暂停任务
public function pauseTask($taskId) {
    // 创建暂停标志
    file_put_contents("/tmp/pause_{$taskId}.flag", 'pause');
    return json_encode(['success' => true]);
}
// 恢复任务
public function resumeTask($taskId) {
    // 删除暂停标志
    @unlink("/tmp/pause_{$taskId}.flag");
    // 启动恢复处理（可通过后台进程或队列）
    $task = new ResumeableTask($taskId, '/path/to/file.csv');
    $task->resume();
    return json_encode(['success' => true]);
}
// 前端轮询进度
public function getTaskProgress($taskId) {
    $progress = DB::table('task_progress')
        ->where('task_id', $taskId)
        ->first();
    return json_encode([
        'total' => $progress->total_rows,
        'processed' => $progress->processed_rows,
        'status' => $progress->status
    ]);
}

核心优化策略

原子性操作

使用数据库事务或文件锁确保进度更新安全：

DB::transaction(function() use ($progress) {
    DB::table('task_progress')
        ->where('task_id', $taskId)
        ->update($progress);
});

超时与重试机制

// 设置超时
set_time_limit(0);
// 重试逻辑
$retryTimes = 3;
while ($retryTimes > 0) {
    try {
        // 处理逻辑
        break;
    } catch (Exception $e) {
        $retryTimes--;
        if ($retryTimes === 0) throw $e;
        sleep(1); // 等待后重试
    }
}

数据校验

在合并分片或恢复任务前,执行完整性校验：

public function verifyChunks($fileId) {
    // 检查所有分片是否存在
    $expectedChunks = ceil($this->totalSize / CHUNK_SIZE);
    // 使用MD5校验每个分片
}

并发控制

对于多线程/多进程场景,使用Redis锁：

$lockKey = "task_lock:{$taskId}";
if ($redis->setnx($lockKey, 1)) {
    $redis->expire($lockKey, 60);
    // 执行任务
    $redis->del($lockKey);
}

最佳实践总结

场景	关键点	实现方式
文件上传断点续传	分片、进度追踪、合并	前端File.slice() + 后端按序合并
任务处理断点续传	状态持久化、精确偏移量	数据库记录 + fseek() 跳转
大规模数据处理	批处理、周期性提交	每处理N条记录更新一次进度
可靠性保证	原子操作、校验和、重试	事务、MD5哈希、指数退避重试

通过以上方案，可以在PHP项目中实现稳定可靠的断点续传恢复机制，支持从各种中断（网络故障、服务器重启、用户暂停等）中无缝恢复任务。

本文地址： https://www.bmjup.com/post/8858.html