PHPAPI任务进度怎么上报

wen PHP项目 2026-07-05 23

本文目录导读：

PHPAPI任务进度怎么上报

文章标题：PHP API 任务进度上报：从轮询到WebSocket的完整实现指南
问题背景与常见场景
基础方案：数据库+轮询
进阶方案：Redis Pub/Sub + 长轮询
实时方案：WebSocket + 异步任务队列
核心代码示例与安全考量
问答环节

PHP API 任务进度上报：从轮询到WebSocket的完整实现指南

目录导读

问题背景与常见场景：为什么需要任务进度上报？
基础方案：数据库+轮询：简单可靠的实现方式
进阶方案：Redis Pub/Sub + 长轮询：低延迟的折中方案
实时方案：WebSocket + 异步任务队列：真正的实时进度推送
核心代码示例与安全考量：防重复提交、权限验证与状态机设计
问答环节：解决开发者最常遇到的5个问题

问题背景与常见场景

在实际开发中，许多耗时的任务（如数据导出、批量导入、API数据同步）无法在单一HTTP请求中完成，用户需要了解当前处理到第几步、剩余时间、成功率等信息。
典型需求包括：

用户上传10万条Excel数据，后台逐条处理时需显示“已处理8,423/100,000条”。
第三方平台调用你的API批量生成报告，要求返回唯一task_id并通过另一个接口查询进度。

任务进度上报成为连接异步处理与用户感知的关键桥梁。

基础方案：数据库+轮询

实现原理：

任务启动时，在task_progress表中插入一条记录，状态设为pending。
后台worker（如supervisor管理的PHP cli脚本）处理后逐条更新进度。
用户前端通过setInterval每隔2秒调用GET /api/progress?task_id=xxx接口。

核心代码（简略）：

// 启动任务
$taskId = createTask(); // 写入数据库
dispatchJob($taskId);   // 丢进队列
// 更新进度（由Worker执行）
function updateProgress($taskId, $step, $total, $status='processing') {
    DB::table('task_progress')->where('id', $taskId)->update([
        'step' => $step,
        'total' => $total,
        'status' => $status,
        'updated_at' => now()
    ]);
}
// 查询接口
public function progress($taskId) {
    $task = DB::table('task_progress')->find($taskId);
    if ($task->status === 'completed') {
        return ['progress' => 100, 'download_url' => $task->result_url];
    }
    return ['progress' => round($task->step / $task->total * 100, 1)];
}

优缺点：

✅ 实现简单，兼容所有环境。
❌ 轮询浪费资源；实时性差（2秒延迟可接受，但若任务耗时1分钟，用户仍会感觉卡顿）。

进阶方案：Redis Pub/Sub + 长轮询

改进方向：

去掉数据库写入，改用Redis List存储进度片段。
前端改用长轮询（Long Polling）：客户端请求后，服务端等待数据更新再返回，而非固定时间轮询。

关键实现：

// 后端Worker推送进度
$redis->publish('task_progress:'.$taskId, json_encode([
    'step' => $step,
    'total' => $total,
    'percentage' => round($step/$total*100, 2)
]));
// 前端长轮询接口
public function longPollingProgress($taskId) {
    $startTime = time();
    $maxWait = 30; // 最长等待30秒
    while (time() - $startTime < $maxWait) {
        $latest = $redis->get('task_status:'.$taskId);
        if ($latest) {
            return response()->json($latest);
        }
        usleep(500000); // 等待0.5秒重试
    }
    return response()->json(['status' => 'timeout']);
}

注意：此方案仍需客户端发起多次连接，但比传统轮询节约70%请求量。

实时方案：WebSocket + 异步任务队列

对于需要毫秒级反馈的场景（如实时监控图标、大文件分段上传进度），推荐使用WebSocket。
架构建议：

用Laravel+Pusher或Swoole+WebSocket。
任务进度通过WebSocket推送，同时用Redis Pub/Sub解耦worker和推送服务器。

示例（基于Swoole）：

// worker端：推送进度到Redis
$redis->publish('task:progress:'.$taskId, $progressMessage);
// WebSocket服务端：订阅Redis并转发到连接的客户端
$server->on('open', function($req) {
    $taskId = $req->get['task_id'] ?? null;
    if (!$taskId) return;
    // 客户端连接后，订阅对应频道
    $server->task(['action' => 'subscribe', 'channel' => 'task:progress:'.$taskId]);
});
$server->on('task', function($server, $task) {
    $redis = new Redis();
    $redis->subscribe([$task['channel']], function($redis, $channel, $msg) use ($server) {
        $server->push($server->connections, $msg);
    });
});

优点：

✅ 真正的实时推送，不浪费任何资源。
✅ 支持百万级并发（需合理设计）。
缺点：
❌ 需要额外搭建WebSocket服务，学习成本高。

核心代码示例与安全考量

状态机设计（防止脏数据）：

// 状态流转规则
const STATES = [
    'pending'   => ['running'],
    'running'   => ['paused', 'failed', 'completed'],
    'paused'    => ['running'],
    'failed'    => [],
    'completed' => []
];
function transitionState($taskId, $newState) {
    $current = DB::table('tasks')->where('id', $taskId)->value('status');
    if (!in_array($newState, STATES[$current])) {
        throw new \Exception('非法状态变更');
    }
    DB::table('tasks')->where('id', $taskId)->update(['status' => $newState]);
}

安全防护：

限制查询频率：在查询接口加上RateLimiter，每个IP每分钟最多查询50次。
权限验证：只有任务创建者才能查看该任务的进度。
过期清理：task完成后1小时自动清理Redis/Db中的进度数据。

问答环节

Q1：如果我必须使用轮询，如何减少数据库压力？
A：优先使用Redis缓存进度，每10秒持久化一次到数据库，前端轮询时，直接读取Redis而非数据库。

Q2：WebSocket方案中，客户端断开重连后如何恢复进度？
A：在WebSocket连接对象中存储task_id和last_known_percentage，重连后，服务端先发送最后一条已知进度的后续数据。

Q3：是否可以用Server-Sent Events（SSE）替代WebSocket？
A：可以，SSE更简单（只需HTTP协议），但只支持单向推送（服务端到客户端），适合进度通知场景，推荐用于Node.js或PHP+nginx的chunked传输。

Q4：我的任务是数据库批处理，进度如何精确到“每5%更新一次”？
A：在worker循环中增加逻辑判断：只有当(当前行数 % 总行数) >= 0.05时才更新进度，可显著减少IO次数。

Q5：如何处理任务中途意外中断？
A：使用消息队列（如RabbitMQ）的ack机制，worker崩溃后，未确认的任务自动重试，同时数据库记录worker_id，超时后标记为failed。

温馨提示：选择方案时需平衡开发成本与用户体验，内部系统用数据库轮询足以；对外付费API建议用WebSocket推送websocket.advanced-api.com（示例域名）这类服务，若需更详细的代码示例,欢迎在评论区留下你的具体场景。