带有 PrefetchPolicy 的 ActiveMQ-cpp Broker URI 无效答案

【问题标题】：ActiveMQ-cpp Broker URI with PrefetchPolicy has no effect带有 PrefetchPolicy 的 ActiveMQ-cpp Broker URI 无效
【发布时间】：2017-12-08 17:29:16
【问题描述】：

我正在使用activemq-cpp 3.7.0和VS 2010来构建客户端，服务器是ActiveMQ 5.8。我基于提到的 CMS 配置here，使用类似于以下的代码创建了一个消息使用者。 ConnClass 是 ExceptionListener 和 MessageListener。我只想在调用 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
}

问题是我在调用m_session->commit() 之前收到多条消息而不是一条消息——我知道这一点是因为commit() 调用是由用户输入触发的。如何确保在每次调用 commit() 之前只调用一次 onMessage()？

【问题讨论】：

标签： c++ activemq prefetch activemq-cpp

【解决方案1】：

这样不行。使用异步消费者时，消息的传递速度与 onMessage 方法完成的速度一样快。如果您只想使用一条消息，请使用同步接收调用。

对于异步消费者，预取允许代理缓冲客户端上的工作，而不是一次触发一个，因此您通常可以获得更好的性能，在您的情况下，当异步 onMessage 调用完成时，一个 ack 被发送回代理下一条消息被发送到客户端。

【讨论】：

但是为什么消息的数量不能通过预取“大小”选项来控制呢？还是我错误地理解了预取选项？
您不了解该选项。它只是向经纪人表明要尝试在客户端上堆积多少工作。异步消费者中的异步一词应该是您的线索。
我使用了异步，因此我不需要管理一个额外的线程，我必须通过同步明确地这样做。我试图限制在提交事务之前收到的消息数量。您是说这不能使用异步方法的预取策略中的 size 参数来完成吗？这是否意味着如果我只想要一条消息，我应该强制 onMessage 在方法结束时阻止，直到我提交？顺便说一句，我使用的是 SESSION_TRANSACTED 模式，所以我使用提交而不是确认消息。
我不确定还有多少种方式可以说预取不是限制。它是异步的，因此您可以发明一些方法来阻止它，或者您可以只使用同步消费者，让您的生活更简单。
我明白了，那么预取大小必须是内部库缓冲区设置。谢谢你的耐心。我会尝试同步消费者。

【解决方案2】：

是的，我也发现了这个。但是，当我为异步消费者使用目标 URI 选项（“consumer.prefetchSize=15”，http://activemq.apache.org/cms/configuring.html#Configuring-DestinationURIParameters）时，效果很好。

顺便说一句，我只使用 Tim 最新的 ActiveMQ-CPP v3.9.4 和 CentOS 7 上的 ActiveMQ v5.12.1。

谢谢！

【讨论】：