使用 Pheanstalk 进行 PHP 进程分叉

Question

我正在尝试创建一个在后台运行并派生子进程的 PHP 脚本。 （我知道这可能会导致服务器爆炸；有超出此问题范围的额外保护措施）

简而言之，代码是这样工作的：

$pids = array();
$ok = true;
while ($ok)
{
    $job = $pheanstalk->watch('jobs')->ignore('default')->reserve();
    echo 'Reserved job #' . $job->getId();
    $pids[$job->getId()] = pcntl_fork();
    if(!$pids[$job->getId()]) {
       doStuff();
       $pheanstalk->delete($job);
       exit();
    }
}

问题是，一旦我 fork 进程，我就会得到错误：

Reserved job #0
PHP Fatal error:  Uncaught exception 'Pheanstalk_Exception_ServerException' with message 'Cannot delete job 0: NOT_FOUND'

我的问题是，pheanstalk 是如何返回一个没有 ID 和有效负载的作业的？一旦我 fork 它几乎感觉就像 $pheanstalk 损坏了。如果我删除分叉，一切正常。 （虽然它必须等待每个进程）

Answer 1

在删除 pheanstalk 作业之前加上这个 if 条件：

if ($job) $pheanstalk->delete($job);

那是因为在代码到达此位置之前，您文件的另一个 php 实例很可能已经删除了该作业。 （其他实例仍然可以使用 reserve() 检索此作业，直到该作业从队列中删除。

Answer 2

您遇到此问题的原因是该作业由主进程保留。在您调用pcntl_fork() 之后，实际上有$worker 变量的副本，因此主进程锁定了作业，而第二个作业在尝试删除它时说它不存在（或者在这种情况下它由另一个进程保留）。下面的代码通过创建一个新的 worker 来处理它，然后释放主 worker 上的作业，并尝试在第二个 worker 上接它。

# worker for the main process
$worker = new \Pheanstalk\Pheanstalk($host, $port, $timeout, $persistent);

$pid = -1;

# seek out new jobs
while ($job = $worker->watch('queue_caller')->reserve()) {

    # make sure pcntl is installed & enabled
    if (function_exists('pcntl_fork')) {
        # create a child process
        $pid = pcntl_fork();
    }

    if ($pid === -1) {
        # code will run in single threaded mode
    } elseif ($pid !== 0) {
        # parent process
        # release the job so it can be picked up by the fork
        $worker->release($job);

        # short wait (1/20000th second) to ensure the fork executes first
        # adjust this value to what is appropriate for your environment
        usleep(50);

        # clear out zombie processes after they're completed
        pcntl_waitpid(0, $pidStatus, WNOHANG);

        # go back to looking for jobs
        continue;
    } else {
        # child worker is needed, because it has to own the job to delete it
        /** @var Pheanstalk $worker */
        $worker = new \Pheanstalk\Pheanstalk($host, $port, $timeout, $persistent);

        # only look for jobs for 1 second, in theory it should always find something
        $job = $worker->watch('queue_caller')->reserve(1);

        if (false === $job) {
            # for some reason there is no job
            # terminate the child process with an error code
            exit(1);
        }

    }

    /** Start your code **/

    do_something();

    /** End your code **/

    # delete the job from the queue        
    $worker->delete($job);

    # only terminate if it's the child process 
    if ($pid === 0) {
        # terminate the child process with success code
        exit(0);
    }

}