1

当我使用 Stomp 从 AMQ 读取一条消息时，我得到 3 或 4 条消息出队，不知道为什么。

填充 AMQ 的代码：

public function populate($queue, $c = 10) {

    for($i = 0; $i < $c; $i++) {
        $this->stomp->send($queue,' Random populated:'.rand(0, PHP_INT_MAX));
    }

}

填充后的 ActiveMQ 队列 填充后的 ActiveMQ 消息

读取 AMQ 的代码：

public function read($queue = null) {

    if(is_null($queue)) {
        if(!$this->isSubscribed()) {
            return false;
        }
    } else {
        $this->subscribe($queue);
    }

    return $this->stomp->readFrame();

}

踩下 readFrame() 代码：

public function readFrame ()
{
    if (!$this->hasFrameToRead()) {
        return false;
    }

    $rb = 1024;
    $data = '';
    $end = false;

    do {
        $read = fread($this->_socket, $rb);
        if ($read === false) {
            $this->_reconnect();
            return $this->readFrame();
        }
        $data .= $read;
        if (strpos($data, "\x00") !== false) {
            $end = true;
            $data = rtrim($data, "\n");
        }
        $len = strlen($data);
    } while ($len < 2 || $end == false);

    list ($header, $body) = explode("\n\n", $data, 2);
    $header = explode("\n", $header);
    $headers = array();
    $command = null;
    foreach ($header as $v) {
        if (isset($command)) {
            list ($name, $value) = explode(':', $v, 2);
            $headers[$name] = $value;
        } else {
            $command = $v;
        }
    }
    $frame = new StompFrame($command, $headers, trim($body));
    if (isset($frame->headers['transformation']) && $frame->headers['transformation'] == 'jms-map-json') {
        require_once 'Stomp/Message/Map.php';
        return new StompMessageMap($frame);
    } else {
        return $frame;
    }
    return $frame;
}

我 100% 确定代码只执行一次，但结果是： 读取一条消息后的 ActiveMQ 队列 读取一条消息后的 ActiveMQ 消息

Var_dumped 消息：

object(StompFrame)[4]
public 'command' => string 'MESSAGE' (length=7)
public 'headers' => 
array (size=5)
  'message-id' => string 'ID:**********_-49723-1350635513276-2:1:-1:1:1' (length=45)
  'destination' => string '/queue/test' (length=11)
  'timestamp' => string '1350635842968' (length=13)
  'expires' => string '0' (length=1)
  'priority' => string '4' (length=1)
public 'body' => string 'Random populated:1859256320' (length=27)

有谁知道这种行为的原因是什么？

注意事项：

  • 没有消息确认，因此甚至一条消息都不应该出队：|
  • ACK 处于客户端模式
  • 预取大小设置为 1
4

1 回答 1

3

我没有看到您连接代码，但我假设您正在使用 Auto Ack 模式进行连接。在具有 Auto Ack 模式的 STOMP 中，消息一旦到达线路就会被确认，并且由于我还假设您没有更改预取大小，因此代理将向您发送一批消息，以便您从套接字和更多可以发送更多将出列。如果您想对消息消费进行更细粒度的控制，您应该使用另一种 ack 模式，例如客户端 ack 并在每条消息到达时对其进行确认。您还可以为订阅设置预取窗口，以减少批量发送到客户端的消息数量。

有关 AMQ STOMP 配置选项，请参阅此页面。您可能还想再次查看 STOMP 规范。

于 2012-10-19T10:15:21.427 回答