PHP队列系统

Question

我需要建立一个队列系统，偶然发现了这个话题How to Build a PHP Queue System。

当队列监听器运行时，这就像一个魅力。但是偶尔，队列侦听器可能会崩溃或停止，我需要重新启动它。不是问题。

存在的问题是，与此同时，作业被添加到队列中。但是当再次启动侦听器时，这些作业不会立即执行。队列是空的。如果队列侦听器未运行，我如何注意在队列侦听器启动时作业继续排队并处理？

队列监听器：

<?php
//queue-listener.php

function process($job) {
    sleep(1); //. make it look like we did work.
    return;
}

$queue = array();

//////// setup our named pipe ////////
$pipefile = '/storage/queueserver';

umask(0);
if(!file_exists($pipefile)){
    if(!posix_mkfifo($pipefile,0666)) {
        die('unable to create named pipe');
    }
}

$pipe = fopen($pipefile,'r+');
if(!$pipe) die('unable to open the named pipe');
stream_set_blocking($pipe,false);

//////// process the queue ////////
while(1) {

    while($input = trim(fgets($pipe))) {
        stream_set_blocking($pipe,false);
        $queue[] = $input;
    }

    $job = current($queue);
    $jobkey = key($queue);
    if($job) {
        echo 'processing job ', $job, PHP_EOL;

        process($job);

        next($queue);
        unset($job,$queue[$jobkey]);
    } else {
        echo 'no jobs to do - waiting...', PHP_EOL;
        stream_set_blocking($pipe,true);
    }
}

向队列中添加内容的代码

$pipefile = '/storage/queueserver';
$fhp = fopen($pipefile, 'r+') or die ("can't open file $pipefile");
fwrite($fhp, "GenerateLabel|". date('H:i:s')."\n");

Answer 1

我现在假设每次启动您的脚本时，都会使用 posix_mkfifo 函数再次创建文件。

您能否通过添加到您的脚本来确认：

umask(0);
if(!file_exists($pipefile)){
    if(!posix_mkfifo($pipefile,0666)) {
        die('unable to create named pipe');
    } else {
        echo "Pipe created";
    }
}

然后第一次启动它。它会回声。然后退出它，然后重新启动。如果还是回显，说明问题出在PHP没有看到文件已经存在（可能是文件的权限？）