PHP网页游戏任务队列批量处理性能优化求助

问题背景 #

运行一款基于PHP的网页游戏，每位玩家最多可创建100个攻击任务，任务详情与end_time存储在数据库中。网站任何玩家的页面加载都会触发函数，检查并处理所有end_time小于等于当前时间的任务，机制类似《Travian》。但当同时存在20个及以上到期任务时，系统卡顿严重，每次页面加载仅能处理1个任务，极大影响游戏体验，当前采用foreach循环处理查询到的任务队列数组。

相关代码 #

页面加载调用的处理入口函数 #

public function processQueue($type = 3, $playerId = 0)
{
    global $gameConfig;
    $this->load_model('Mutex', 'mutex');
    $this->mutex->releaseOnTimeout();
    if ($this->mutex->lock()) { // 防止多玩家同时处理同一行数据导致数值重复
        $this->processTaskQueue($type, $playerId);
    }
    $this->mutex->release();
}

任务队列处理函数 #

public function processTaskQueue($type, $playerId)
{
    $p_type = QS_ACCOUNT_DELETE . ',' . QS_MERCHANT_GO . ',' . QS_MERCHANT_BACK . ',' . QS_WAR_REINFORCE . ',' . QS_WAR_ATTACK . ',' . QS_WAR_ATTACK_PLUNDER . ',' . QS_WAR_ATTACK_SPY . ',' . QS_CREATEVILLAGE . ',' . QS_TATAR_RAISE . ',' . QS_SITE_RESET . ',' . QS_CROP_DELETE . ',' . QS_ARTEFACTS_RAISE;
    if ($type == 1) {
        $expr = "((q.player_id=" . $playerId . " AND q.proc_type!=" . QS_TROOP_TRAINING . ") OR q.proc_type IN (" . $p_type . "))";
    } elseif ($type == 2) {
        $expr = "(q.player_id=" . $playerId . " OR q.proc_type IN (" . $p_type . "))";
    } else {
        $expr = "q.proc_type!=" . QS_TROOP_TRAINING;
    }
    $result = db::get_all("SELECT  q.id, q.player_id, q.village_id, q.to_player_id, q.to_village_id, q.proc_type, q.building_id, q.proc_params, q.threads, q.execution_time, TIMESTAMPDIFF(SECOND, NOW(),q.end_date) remainingTimeInSeconds FROM p_queue q WHERE TIMESTAMPDIFF(SECOND, NOW(),(q.end_date - INTERVAL (q.execution_time*(q.threads-1)) SECOND)) <= 0 AND $expr ORDER BY TIMESTAMPDIFF(SECOND, NOW(),(q.end_date - INTERVAL (q.execution_time*(q.threads-1)) SECOND)) ASC");
    foreach ($result as $resultRow) {
        $remain = $resultRow['remainingTimeInSeconds'];
        if ($remain < 0) {
            $remain = 0;
        }
        $resultRow['threads_completed_num'] = $resultRow['execution_time'] <= 0 ? $resultRow['threads'] : floor(($resultRow['threads'] * $resultRow['execution_time'] - $remain) / $resultRow['execution_time']);
        if ($this->processTask($resultRow)) {
            unset($result);
            $this->processQueue($type, $playerId);
            break;
        }
    }
    unset($result);
}

锁机制实现代码 #

define("__QS_LOCK_FS_", MODELS_DIR . "lock");

class Mutex_Model extends Model
{
    public function lock()
    {
        if (0 < db::count("UPDATE g_settings gs SET gs.qlocked=1, qlocked_date=NOW() WHERE gs.qlocked=0") && ($fp = fopen(__QS_LOCK_FS_, "r")) != FALSE) {
            if (flock($fp, LOCK_EX)) {
                fclose($fp);
                return TRUE;
            }
            fclose($fp);
        }
        return FALSE;
    }

    public function release()
    {
        $this->_releaseInternal();
        db::query("UPDATE g_settings gs SET gs.qlocked=0");
    }

    public function releaseOnTimeout()
    {
        if (0 < db::count("UPDATE g_settings gs SET gs.qlocked=0 WHERE gs.qlocked=1 AND TIME_TO_SEC(TIMEDIFF(NOW(), gs.qlocked_date)) > 120")) {
            $this->_releaseInternal();
        }
    }

    public function _releaseInternal()
    {
        if (($fp = fopen(__QS_LOCK_FS_, "r")) != FALSE) {
            flock($fp, LOCK_UN);
            fclose($fp);
        }
    }
}

已尝试的无效方案 #

• 删除break让循环执行完毕：玩家收到500错误，获取锁的页面加载极慢
• 用Cron Job循环调用processQueue函数59秒：每秒仅能处理10-20个任务，效率极低
• 移除锁机制：出现数据重复问题（比如玩家发送50名士兵，返回时变成100名）

服务器配置 #

• CPU：12核（仅占用2核）
• 内存：131,744,852
• Apache配置：
  • Start Servers：10
  • Minimum Spare Servers：10
  • Maximum Spare Servers：20
  • Server Limit：8042
  • Max Request Workers：8042
  • Max Connections Per Child：10000
  • Keep-Alive：开启
• PHP-FPM池配置：
  • Max Requests：95000
  • Max Children：10000
  • Process Idle Timeout：30

解决方案建议 #

1. 限制单次任务处理数量，避免请求超时 #

当前代码处理1个任务就递归调用并终止循环，改成单次固定处理N个任务（比如10个），平衡处理效率与请求耗时：

public function processTaskQueue($type, $playerId)
{
    $p_type = QS_ACCOUNT_DELETE . ',' . QS_MERCHANT_GO . ',' . QS_MERCHANT_BACK . ',' . QS_WAR_REINFORCE . ',' . QS_WAR_ATTACK . ',' . QS_WAR_ATTACK_PLUNDER . ',' . QS_WAR_ATTACK_SPY . ',' . QS_CREATEVILLAGE . ',' . QS_TATAR_RAISE . ',' . QS_SITE_RESET . ',' . QS_CROP_DELETE . ',' . QS_ARTEFACTS_RAISE;
    if ($type == 1) {
        $expr = "((q.player_id=" . $playerId . " AND q.proc_type!=" . QS_TROOP_TRAINING . ") OR q.proc_type IN (" . $p_type . "))";
    } elseif ($type == 2) {
        $expr = "(q.player_id=" . $playerId . " OR q.proc_type IN (" . $p_type . "))";
    } else {
        $expr = "q.proc_type!=" . QS_TROOP_TRAINING;
    }
    // 限制单次查询10个任务，避免处理过多导致超时
    $result = db::get_all("SELECT  q.id, q.player_id, q.village_id, q.to_player_id, q.to_village_id, q.proc_type, q.building_id, q.proc_params, q.threads, q.execution_time, TIMESTAMPDIFF(SECOND, NOW(),q.end_date) remainingTimeInSeconds FROM p_queue q WHERE TIMESTAMPDIFF(SECOND, NOW(),(q.end_date - INTERVAL (q.execution_time*(q.threads-1)) SECOND)) <= 0 AND $expr ORDER BY TIMESTAMPDIFF(SECOND, NOW(),(q.end_date - INTERVAL (q.execution_time*(q.threads-1)) SECOND)) ASC LIMIT 10");
    $processedCount = 0;
    $maxPerRun = 10;
    foreach ($result as $resultRow) {
        if ($processedCount >= $maxPerRun) break;
        
        $remain = $resultRow['remainingTimeInSeconds'];
        if ($remain < 0) $remain = 0;
        $resultRow['threads_completed_num'] = $resultRow['execution_time'] <= 0 ? $resultRow['threads'] : floor(($resultRow['threads'] * $resultRow['execution_time'] - $remain) / $resultRow['execution_time']);
        if ($this->processTask($resultRow)) {
            $processedCount++;
        }
    }
    unset($result);
}

2. 优化锁机制，替换全局锁为行级标记 #

全局锁会阻塞所有请求，改成在查询时标记任务为"处理中"，避免重复处理：

// 先标记到期任务为处理中（带超时，防止死锁）
db::query("UPDATE p_queue q SET status='processing', process_start=NOW() 
           WHERE TIMESTAMPDIFF(SECOND, NOW(),(q.end_date - INTERVAL (q.execution_time*(q.threads-1)) SECOND)) <= 0 
           AND $expr AND status='pending' 
           ORDER BY TIMESTAMPDIFF(SECOND, NOW(),(q.end_date - INTERVAL (q.execution_time*(q.threads-1)) SECOND)) ASC 
           LIMIT 10");
// 查询已标记的任务进行处理
$result = db::get_all("SELECT q.id, q.player_id, q.village_id, q.to_player_id, q.to_village_id, q.proc_type, q.building_id, q.proc_params, q.threads, q.execution_time, TIMESTAMPDIFF(SECOND, NOW(),q.end_date) remainingTimeInSeconds 
                       FROM p_queue q 
                       WHERE status='processing' AND process_start > NOW() - INTERVAL 5 MINUTE");

同时需要定期清理超时未完成的"处理中"任务，恢复为"pending"状态。

3. 异步任务队列（长期最优方案） #

将任务处理从页面加载流程中剥离，用异步队列实现：

• 页面加载仅负责检查到期任务，将其推入Redis队列
• 用Supervisor启动多个PHP进程，持续从队列中取出任务处理
• 此方案完全不阻塞用户请求，还能利用多核CPU提升处理效率，适合大规模任务场景

4. 数据库查询优化 #

• 给p_queue表的end_date、proc_type、player_id、status字段建立联合索引，加速查询
• 新增process_time字段，提前计算任务的实际可处理时间，避免每次查询都执行复杂的TIMESTAMPDIFF计算

内容的提问来源于stack exchange，提问作者Zeyd Harb