PHP 開発における同時実行性の高いタスクのスケジュールの問題を解決する方法

// 生产者，将任务添加到消息队列中
$redis = new Redis();
$redis->pconnect('localhost', 6379);
$task = array(
    'task_id' => 1,
    'task_data' => 'This is a task',
);
$redis->lpush('task_queue', json_encode($task));

// 消费者，从消息队列中获取任务并执行
while (true) {
    $redis = new Redis();
    $redis->pconnect('localhost', 6379);
    $taskStr = $redis->rpop('task_queue');
    if ($taskStr) {
        $task = json_decode($taskStr, true);
        // 执行任务
        $result = executeTask($task);
    }
    usleep(1000); // 休眠一毫秒，避免空转浪费CPU资源
}

function executeTask($task) {
    // 执行具体任务逻辑
    // ...
}

$workerNum = 10; // 定义进程池大小
$workers = array(); // 进程数组

// 创建子进程
for ($i = 0; $i < $workerNum; $i++) {
    $pid = pcntl_fork();
    if ($pid < 0) {
        exit("Fork failed");
    } elseif ($pid === 0) {
        // 子进程执行任务
        while (true) {
            // 从消息队列中获取任务并执行
            $redis = new Redis();
            $redis->pconnect('localhost', 6379);
            $taskStr = $redis->rpop('task_queue');
            if ($taskStr) {
                $task = json_decode($taskStr, true);
                // 执行任务
                $result = executeTask($task);
            }
            usleep(1000); // 休眠一毫秒，避免空转浪费CPU资源
        }
        exit(0);
    } else {
        $workers[] = $pid;
    }
}

// 主进程等待所有子进程退出
foreach ($workers as $pid) {
    pcntl_waitpid($pid, $status);
}

function executeTask($task) {
    // 执行具体任务逻辑
    // ...
}

// 生产者，将任务添加到Beanstalkd队列中
$pheanstalk = new Pheanstalk('127.0.0.1');
$taskData = array(
    'task_id' => 1,
    'task_data' => 'This is a task',
);
$pheanstalk->useTube('task_queue')->put(json_encode($taskData));

// 消费者，从Beanstalkd队列中获取任务并执行
$pheanstalk = new Pheanstalk('127.0.0.1');
while (true) {
    $job = $pheanstalk->reserve();
    if ($job) {
        $task = json_decode($job->getData(), true);
        // 执行任务
        $result = executeTask($task);
        $pheanstalk->delete($job); // 任务执行完成后删除任务
    }
    usleep(1000); // 休眠一毫秒，避免空转浪费CPU资源
}

function executeTask($task) {
    // 执行具体任务逻辑
    // ...
}