doujie1908
2015-01-03 18:28
浏览 81
已采纳

Activemq和Php Stomp:同步生产者样本

I'm trying have this principle working:

  • a producer that sends one message (1) and waits for ack which contains some result (json result of an operation, actually)
  • a consumer that checks all pending messages every 5 seconds, and handle all of them in one row, and acknowlegdes all of them in one row, then wait again 5 seconds (infinite loop).

Here are the 30 lines of my stompproducer.php:

<?php

function msg($txt)
{
    echo date('H:i:s > ').$txt."
";
}

$queue  = '/aaaa';
$msg    = 'bar';
if (count($argv)<3) {
    echo $argv[0]." [msg] [nb to send]
";
    exit(1);
}
$msg     = (string)$argv[1];
$to_send = intval($argv[2]);

try {
    $stomp = new Stomp('tcp://localhost:61613');
    while (--$to_send) {
        msg("Sending...");
        $result = $stomp->send(
            $queue,
            $msg." ". date("Y-m-d H:i:s"),
            array('receipt' => 'message-123')
        );
        echo 'result='.var_export($result,true)."
";
        msg("Done.");
    }
} catch(StompException $e) {
    die('Connection failed: ' . $e->getMessage());
}

Here are the 30 lines of my stompconsumer.php:

<?php

$queue  = '/aaaa';
$_waitTimer=5000000;
$_timeLastAsk = microtime(true);

function msg($txt)
{
    echo date('H:i:s > ').$txt."
";
}

try {
    $stomp = new Stomp('tcp://localhost:61613');
    $stomp->subscribe($queue, array('activemq.prefetchSize' => 40));
    $stomp->setReadTimeout(0, 10000);
    while (true) {
        $frames_read=array();
        while ($stomp->hasFrame()) {
            $frame = $stomp->readFrame();
            if ($frame != null) {
                array_push($frames_read, $frame);
            }
            if (count($frames_read)==40) {
                break;
            }
        }
        msg("Nombre de frames lues : ".count($frames_read));
        msg("Pause...");
        $e=$_waitTimer-(microtime(true)-$_timeLastAsk);
        if ($e>0) {
            usleep($e);
        }
        if (count($frames_read)>0) {
            msg("Ack now...");
            foreach ($frames_read as $frame) {
                $stomp->ack($frame);
            }
        }
        $_timeLastAsk = microtime(true);
    }
} catch(StompException $e) {
    die('Connection failed: ' . $e->getMessage());
}

I can't manage to do synchronous producer, ie producer that waits for consumer ack. If you run the samples I've done here, you'll see that producer instantaneously sends all messages, then quits, with all "true" like "ok" results when calling $stomp->send(). I still haven't found good examples, neither good documentation with a simple blocking sample.

What shall I do to make my producer blocking until the consumer sends its ack?

NB: I've read all documentation here and the stomp php questions on stackoverflow here and here.

  • 写回答
  • 好问题 提建议
  • 关注问题
  • 收藏
  • 邀请回答

3条回答 默认 最新

  • dtgvl48608 2016-03-10 13:09
    已采纳

    First thing to pop t my mind: Take a look at this stomp plugin:

    http://activemq.apache.org/message-redelivery-and-dlq-handling.html

    Another workaround I can thing of is: On producer side: 1. Change your producer to send persistent messages

    On your consumer side: Use a timer. 1. Read message/frames until empty or max cap reached. 2. Create a CURL request and empty packed list of messages 3. Sleep your server for 5 secs

    You definitely need to test this further, but should work. Once the process wakes up, you should be able to read all messages queued.

    Things to consider: - persistent messages will need an expiration time - You'll need ACK on your consumer side to make sure to update status of messages already attended. Use ACK=client so you can ACK all messages acknowledged - It's easier if you don't have to wait for your CURL to respond. - Out of the box, it's not supported to send ACK from the consumer (server side).

    Best of luck

    已采纳该答案
    评论
    解决 无用
    打赏 举报
  • duanchen1937 2015-01-03 20:17

    From the question it sounds like you are looking for a request / response type messaging pattern. This is something you must implement yourself as the STOMP ack you reference is only acking the message to the message broker on behalf of the consumer, the producer has no knowledge of this. Request response involves setting a reply-to address on the outbound message and then waiting to receive a response on that address before sending the next message. There are a great many articles out there that document this sort of thing such as this one.

    Or if you only need to know if the broker has received the message from the client and persisted it then you can use STOMP's built in receipt mechanism to have the broker send you a receipt indicating that it has processed your sent message. This however does not guarantee that a consumer has processed the message yet.

    评论
    解决 无用
    打赏 举报
  • douren9077 2016-03-10 13:16

    I just remembered, you can try reactphp/stomp library. It's an event driven library that might help you. specially take a look ad the core functionality addPeriodicTimer

    https://github.com/reactphp/stomp

    Cheers

    评论
    解决 无用
    打赏 举报

相关推荐 更多相似问题