如何在ActiveMQ中为PHP Stomp客户端组织broker消息重新传递


How to organize broker message redelivery in ActiveMQ for PHP Stomp client

我正在尝试使用ActiveMQ 5.8.0和PHP Stomp扩展实现代理重新交付。但这并没有如预期的那样奏效。

回滚消息的PHP订阅者

define('QUEUE_NAME', '/queue/Task.Test');
$stomp = new Stomp();
$stomp->subscribe(QUEUE_NAME, array(
    'activemq.prefetchSize' => 10,
));
while (true) {
    $tid = uniqid('tid');
    $stomp->begin($tid);
    try {
        $frame = $stomp->readFrame();
        if ($frame) {
            print_r($frame->headers);
            $stomp->ack($frame->headers['message-id'], array('transaction' => $tid));
            echo "Get message {$frame->headers['message-id']}, rollback it", PHP_EOL;
            $stomp->abort($tid);
        } else {
           $stomp->commit($tid);
        }
    } catch (StompException $e) {
        $stomp->abort($tid);
    }
}

ActiveMQ配置(已启用调度程序):

    <plugins>
        <redeliveryPlugin fallbackToDeadLetter="true" sendToDlqIfMaxRetriesExceeded="true">
            <redeliveryPolicyMap>
                <redeliveryPolicyMap>
                    <redeliveryPolicyEntries>
                        <!-- a destination specific policy -->
                        <redeliveryPolicy queue=">" maximumRedeliveries="2" redeliveryDelay="10000" />
                    </redeliveryPolicyEntries>
                    <!-- the fallback policy for all other destinations -->
                    <defaultEntry>
                        <redeliveryPolicy maximumRedeliveries="2" initialRedeliveryDelay="5000" redeliveryDelay="10000" />
                    </defaultEntry>
                </redeliveryPolicyMap>
            </redeliveryPolicyMap>
        </redeliveryPlugin>
    </plugins>

通过该配置,订阅者接收所有台面并回滚它们,因此它们返回到队列,但我希望它们在延迟一段时间后重新交付。ACK有一个特殊的状态:"毒ACK",但我不知道如何指定。

如何在经纪人方面实现重新交付?

实际上,重新交付需要在STOMP协议版本>=1.1中定义的NACK命令。PHP Stomp客户端仅支持1.0版本,因此不可能为该扩展启用重新交付。然而,使用nack()方法和头处理来扩展Stomp类是很容易的。希望这些信息对某些人有帮助。