好的,不用详细介绍我设置的整个系统,
我遇到的问题是,当消费者取消(AMQPChannel->basic_cancel)监听队列时,它会留下一条额外的消息未被该工作者确认。它也不会触发正常回调来处理此消息。
一些细节
-
队列阻塞(使用等待)
while(count($channel->callbacks)) { $channel->wait( ... ) ... }
-
Pre-fetch是1,这是您可以拥有的最小值
-
消费者可以动态地倾听(
AMQPChannel->basic_consume
)
-
消费者可以动态地忘记(
AMQPChannel->basic_cancel
)
我不会以我告诉给定消费者消费或取消给定队列的确切方式进行讨论。但这一切都很完美,无论是开始消费还是取消都很好。但是,当我用一个仍然有消息的队列取消时,他们只是忘记了最后一条消息,我相信取消会删除该队列的回调,因此,除了杀死消费者之外,没有办法恢复该消息,这是不可取的。
我做了一些调试(只是一个例子,不是我实际做的)
Debug::dump($this->getAmqpChannel()->getMethodQueue());
$tag = $this->_tags[$queue]; //I keep track of the consumer tag on a queue by queue basis, $queue == {queuename} below
$this->getAmqpChannel()->basic_cancel( $tag );
Debug::dump($this->getAmqpChannel()->getMethodQueue());
这项研究的成果大致相当
array()
RunCommand: basic_cancel //this works fine consumer forgets queue except ->
array(1){
[0] => array(3){
[0] => string(5) "60,60",
[1] => string(114) "amq.ctag-D9om-gD_cPevzeon52zpig\0\0\0\0\0\0\0\0\0G{queuename}", //{queuename} is the name of the queue, which is based on clients information I cant share (such as their name)
[2] => object(PhpAmqpLib\Message\AMQPMessage)#0 (9) {
["DELIVERY_MODE_NON_PERSISTENT":constant] => int(1),
["DELIVERY_MODE_PERSISTENT":constant] => int(2),
["body":public] => string(1358647) "{ ... "correlation_id":32,"max_correlation_id":38}"
["body_size":public] => int(135864),
["is_truncated":public] => bool(false),
["content_encoding":public] => null,
["propertyDefinitions":protected static] => array(14){ ... }
["delivery_info":public] => array(0){},
["prop_types":protected] => array(14){ ... }
}
}
一旦工作者死亡(或者我杀死了它),消息就会被放回队列,我可以在get messages下的RabbitMq管理工具(插件)中将其取出。就在这里,
Properties
correlation_id: 32:38
delivery_mode: 2
content_encoding: text/plain
content_type: application/json
这个
"correlation_id":32,"max_correlation_id":38
对应于
correlation_id: 32:38
因为我需要跟踪消息部分。所以我知道这是同样的信息。
那么,为什么在我取消之后,我会收到最后一条被困在僵尸世界里的消息,并且无论如何都会把它踢回队列,而不是杀死消费者。
而且这不是一次性的,每次我取消一个仍然有消息的队列时都会发生这种情况。所以它与给定的信息无关。就好像它收到了最后一条预处理的消息,然后因为它被取消了,所以最后一条消息没有回调,它只是陷入了困境。记住,0预取是获取所有消息,1是可以设置的最低值。
任何能帮上忙的都会很棒。
使现代化
我可以打电话来解决问题
$this->getAmqpChannel()->basic_recover(true); //basic_recover($requeue)
要么就在考试之前,要么就在考试之后
basic_cancel
这拒绝了信息,我甚至可以测试
$this->getAmqpChannel()->getMethodQueue()
如上图所示,查看
$queue
我要取消的消息已被隐藏(尚未实现)。
我试图避免使用
recover
但我认为这应该是可以的,因为消费者使用单一通道,并且正在阻塞,最坏的情况下,它只会拒绝一条有效消息1次,这虽然不理想,但应该是可以接受的。
然而,在某些情况下,我从兔子那里得到了一个额外的例外,
PRECONDITION_FAILED - unknown delivery tag {n}
如果有人对这个额外的错误有任何细节,那就太好了。此外,所有队列都需要确认,没有一个是自动的。
更新1
我注意到堆栈中的痕迹
queue_unbind
所以我所做的是,在内部跟踪绑定,这样我可以确保解除绑定只进行一次。在明天进行更多测试之后,我将发布一些代码,但在实现之后的初始测试不再产生错误。
所有这些听起来可能有点“奇怪”,我可以解释我为什么要这么做,以及我在做什么,但这可能超出了问题的范围。我会说我已经在生产中使用这个系统超过2年(我设计的),我们可以做180K搜索一分钟(大约100,如果你考虑系统的所有部分)。我们还用它进行了2800多亿次搜索,这是我创建的。我们现在也是业界领先的公司,要么淘汰了我们的竞争对手,要么他们把他们的东西寄给我们,不再在公司内部这么做。这在很大程度上是因为我们的快速周转以及数据的质量。所以这个系统确实有效,而且运行得很好。
但在最近的审计中,我注意到,每天的消费者只需要处理大约1000万行(大约100分钟的工作),而晚上的消费者需要处理大约1亿行(约20小时的工作)。日常用户可以在工作时间之外做夜间工作(因为这样可以减少白天的响应时间),因此大约有10个小时的时间窗口,夜间工作只能在更小、性能更低的服务器上运行。这给我们的解决方案是,如果没有日常作业(客户提交的作业),他们可以动态地交换到夜间的东西(数据仓库)上。这将保持大部分响应能力,同时在没有提交作业时不会浪费资源。我们可以在搜索上横向扩展到我们想要的范围,但我们确实为我们的主服务器支付了很多费用,并且浪费了大约8个小时的工作时间。
我可能会在一本小书中介绍这一切是如何运作的,但希望这能让我对自己的工作有一些基本的了解。我还受到合同中一些保密和竞业禁止条款的约束,所以我可以真正了解具体细节。
