Pheanstalk保留两次或更多的Beanstalkd工作

Question

我有一个3 beanstalkd队列进程在同一IP上运行，但端口不同。我有一个单独的服务器运行主管在平行（每个beanstalkd端口20）中产卵PHP工人来处理队列。我的问题是，似乎两个过程可以同时在同一服务器上保留同一工作ID。

这是我日志中的一些示例输出：

2017-02-23 09:59:56 --> START JOB (port: 11301 | u: 0.45138600 1487861996 | jid:1695074 | pid:30019 | j:leads_to_tags_add | tr:1)
2017-02-23 09:59:57 --> START JOB (port: 11301 | u: 0.55024800 1487861997 | jid:1695074 | pid:30157 | j:leads_to_tags_add | tr:2)
2017-02-23 09:59:58 --> DEL   JOB (port: 11301 | u: 0.54731000 1487861998 | jid:1695074 | pid:30019 | j:leads_to_tags_add)
2017-02-23 09:59:58 --> DEL   JOB (port: 11301 | u: 0.58927900 1487861998 | jid:1695074 | pid:30157 | j:leads_to_tags_add)

似乎两个储备金彼此之后发生，第二个储备金发生在第一个过程完成并删除作业之前。

我添加了每个jobid中的柜台，很明显，到第二次保留时，计数器一次上升（tr）。TTRR设置为3600，因此在第一个过程完成之前无法到期。

这是第二个过程保留后的工作状态的样子：

PheanstalkResponseArrayResponse::__set_state(array(
   'id' => '1695074',
   'tube' => 'action-medium',
   'state' => 'reserved',
   'pri' => '0',
   'age' => '1',
   'delay' => '0',
   'ttr' => '3600',
   'time-left' => '3599',
   'file' => '385',
   'reserves' => '2',
   'timeouts' => '0',
   'releases' => '0',
   'buries' => '0',
   'kicks' => '0',
))

这种行为是非常随机的，有时只有一个过程能够保留，直到工作锁定，有时为2，有时甚至是4个或更多（很少）。当然，这会产生不一致的重复作业。

代码的简短版本：

$this->job = $this->pheanstalk->watch($tube)->reserve($timeout);

set_error_handler(function($errno, $errstr, $errfile, $errline, array $errcontext) {
    // error was suppressed with the @-operator
    if (0 === error_reporting()) {
        return false;
    }
    throw new ErrorException($errstr, 0, $errno, $errfile, $errline);
});
$this->log_message('info', __METHOD__ . ": START JOB (" . $this->_logDetails() . " | tr:{$tries})");

if ($this->_process_job()) {
    $this->log_message('info', __METHOD__ . ": FINISHED JOB (" . $this->_logDetails() . ")");
    $this->_delete_job();
} else {
    $this->log_message('error', __METHOD__ . ": FAILED JOB (" . $this->_logDetails() . ")");
}
restore_error_handler();

和

protected function _delete_job()
{
    $this->pheanstalk->delete($this->job);
    $this->log_message('info', __METHOD__ . ": DELETED JOB (" . $this->_logDetails() . ")");
}

Answer 1

这个问题是我代码的crocess_job位中发生的TCP断开连接。pheanstalk对象将在执行链的某个地方覆盖。

我有一个库包装pheanstalk，在发送put命令之前，它可以检查最空的beanstalkd服务器。然后，它将创建一个带有最佳服务器详细信息的新的Pheanstalk实例，将其保存为当前连接，然后派遣命令。

有时工人可以将招聘人员发送到队列。该库最初将加载到工人中以获取作业，然后再次在process_job中发送到队列。由于codeigniter中的依赖注入，该库将参考同一对象，并且在crocess_job内部将覆盖工人的当前连接并导致TCP断开连接。