使用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();
    }
}

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

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);
    }

}