PHP与数据库队列的集成

<?php
function addToQueue($type, $params) {
    $dbh = new PDO('Mysql:host=localhost;dbname=dbname', 'username', 'passWord');
    $sql = "INSERT INTO `queue` (`type`, `params`, `status`) VALUES (:type, :params, 0)";
    $stmt = $dbh->prepare($sql);
    $stmt->bindParam(':type', $type, PDO::PARAM_STR);
    $stmt->bindParam(':params', $params, PDO::PARAM_STR);
    $stmt->execute();
}

<?php
function processQueue() {
    $dbh = new PDO('mysql:host=localhost;dbname=dbname', 'username', 'password');
    $dbh->beginTransaction();

    // 查询是否正在处理任务
    $sql = "SELECT * FROM `queue` WHERE `status` = 1 FOR UPDATE";
    $stmt = $dbh->prepare($sql);
    $stmt->execute();
    $currentTask = $stmt->fetch(PDO::FETCH_ASSOC);

    if (!$currentTask) {
        // 如果没有正在处理的任务，从队列中取出第一个任务
        $sql = "SELECT * FROM `queue` WHERE `status` = 0 ORDER BY `id` ASC LIMIT 1 FOR UPDATE";
        $stmt = $dbh->prepare($sql);
        $stmt->execute();
        $currentTask = $stmt->fetch(PDO::FETCH_ASSOC);
        if ($currentTask) {
            // 标记任务为正在处理
            $sql = "UPDATE `queue` SET `status` = 1 WHERE `id` = :id";
            $stmt = $dbh->prepare($sql);
            $stmt->bindParam(':id', $currentTask['id'], PDO::PARAM_INT);
            $stmt->execute();
        }
    }

    if ($currentTask) {
        // 处理当前任务
        try {
            if ($currentTask['type'] == 'example') {
                // 异步处理任务
                // ...
                // 标记任务为已完成
                $sql = "UPDATE `queue` SET `status` = 2 WHERE `id` = :id";
                $stmt = $dbh->prepare($sql);
                $stmt->bindParam(':id', $currentTask['id'], PDO::PARAM_INT);
                $stmt->execute();
            }
        } catch(Exception $e) {
            // 标记任务为失败
            $sql = "UPDATE `queue` SET `status` = 3 WHERE `id` = :id";
            $stmt = $dbh->prepare($sql);
            $stmt->bindParam(':id', $currentTask['id'], PDO::PARAM_INT);
            $stmt->execute();
        }
    }

    $dbh->commit();
}