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) { // 执行具体任务逻辑 // ... }