ActiveMQ-cpp Broker URI with PrefetchPolicy has no effect

Question

I am using activemq-cpp 3.7.0 with VS 2010 to build a client, the server is ActiveMQ 5.8. I have created a message consumer using code similar to the following, based on the CMS configurations mentioned here. ConnClass is a ExceptionListener and a MessageListener. I only want to consume a single message before calling cms::Session::commit().

void ConnClass::setup()
{

    // Create a ConnectionFactory
    std::tr1::shared_ptr<ConnectionFactory> connectionFactory(
        ConnectionFactory::createCMSConnectionFactory(
            "tcp://localhost:61616?cms.PrefetchPolicy.queuePrefetch=1");

    // Create a Connection
    m_connection = std::tr1::shared_ptr<cms::Connection>(
        connectionFactory->createConnection());

    m_connection->start();
    m_connection->setExceptionListener(this);

    // Create a Session
    m_session = std::tr1::shared_ptr<cms::Session>(
        m_connection->createSession(Session::SESSION_TRANSACTED));


    // Create the destination (Queue)
    m_destination = std::tr1::shared_ptr<cms::Destination>(
        m_session->createQueue("myqueue?consumer.prefetchSize=1"));

    // Create a MessageConsumer from the Session to the Queue
    m_consumer = std::tr1::shared_ptr<cms::MessageConsumer>(
        m_session->createConsumer( m_destination.get() ));

    m_consumer->setMessageListener( this );

}

void ConnClass::onMessage( const Message* message )
{
    // read message code ...
    // schedule a processing event for
    // another thread that calls m_session->commit() when done
}

The problem is I am receiving multiple messages instead of one message before calling m_session->commit() -- I know this because the commit() call is triggered by user input. How can I ensure onMessage() is only called once before each call to commit()?

Answer 1

It doesn't work that way. When using async consumers the messages are delivered as fast as the onMessage method completes. If you want to consume one and only one message then use a sync receive call.

For an async consumer the prefetch allows the broker to buffer up work on the client instead of firing one at a time so you can generally get better proformance, in your case as the async onMessage call completes an ack is sent back to the broker an the next message is sent to the client.

Answer 2

Yes, I find this too. However, when I use the Destination URI option ( "consumer.prefetchSize=15" , http://activemq.apache.org/cms/configuring.html#Configuring-DestinationURIParameters ) for the asynchronous consumer, It works well.

BTW, I just use the latest ActiveMQ-CPP v3.9.4 by Tim , and ActiveMQ v5.12.1 on CentOS 7.

Thanks!