Spring 与 ActiveMQ 集成：当队列没有更多待处理消息时获取通知

Question

我正在使用 Spring 与 ActiveMQ 的集成。当我的队列为空时（不再有待处理的消息），我需要做一些处理。我已经实现了这样的目标，

package my.com.spring.integration;

public class MyListenerContainer extends DefaultMessageListenerContainer {

        @Override protected void messageReceived(Object invoker, Session session) {
            // I mark lastMessageReceived time here 
            super.messageReceived(invoker, session);
        }

        @Override protected void noMessageReceived(Object invoker, Session session) {
            // I wait for 1 minute from last message messageReceived and after 
            // that I consider that queue has no more messages now and
            // I start my stuff
            super.noMessageReceived(invoker, session);
        }
    }

而我的 spring xml 看起来像这样，

<bean id="my.jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://127.0.0.1:61616?jms.prefetchPolicy.queuePrefetch=250&jms.useAsyncSend=true" />
    <property name="optimizeAcknowledge" value="true" />
</bean>

<bean id="my.jms.cachedConnectionFactory"
    class="org.springframework.jms.connection.CachingConnectionFactory"
    p:targetConnectionFactory-ref="my.jmsConnectionFactory"
    p:sessionCacheSize="10" p:reconnectOnException="true" />

 <bean id="myListenerContainer "
        class="my.com.spring.integration.MyListenerContainer ">
        <property name="connectionFactory"
            ref="my.jms.cachedConnectionFactory" />
        <property name="destination" ref="myQ" />
        <property name="concurrentConsumers" value="1" />
        <property name="maxConcurrentConsumers" value="1" />
    </bean>

    <bean id="myMessageListener"
        class="org.springframework.integration.jms.ChannelPublishingJmsMessageListener" />

    <bean id="myJmsEndpoint"
        class="org.springframework.integration.jms.JmsMessageDrivenEndpoint">
        <constructor-arg ref="myListenerContainer" />
        <constructor-arg ref="myMessageListener" />
        <property name="outputChannel" ref="myConsumerChannel" />
    </bean>

到目前为止一切正常。当队列为空并且我能够做我想要的事情时，我会收到通知。

问题：除了队列为空之外，还有任何其他原因不再接收消息吗？就像在 Spring 代码执行中发生任何异常一样，会发生什么？这是实现我的目的的正确方法吗？如果没有，我该如何改进？

Answer 1

noMessageReceived 仅在没有收到消息时调用，而不是在抛出异常时调用。

这意味着 DMLC 没有收到消息，但请记住，只有当您只有一个 DMLC 时，您的目的才是好的。

因为 AMQ 向每个消费者分派了很多消息（在您的情况下为 jms.prefetchPolicy.queuePrefetch=250），并且分派的消息正在对每个消费者进行处理，如果它在没有确认的情况下断开连接，则消息仍然可以发送给另一个消费者。

看看http://activemq.apache.org/what-is-the-prefetch-limit-for.html

如果您想确保从代理获得有关队列的信息，您可以使用 JMX 来获取已调度消息的大小或数量。

import java.util.HashMap;
import java.util.Map;

import javax.management.MBeanServerConnection;
import javax.management.MBeanServerInvocationHandler;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;

import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.QueueViewMBean;

public class JMXGetDestinationInfos {

    public static void main(String[] args) throws Exception {
        JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi");
        Map<String, String[]> env = new HashMap<>();
        String[] creds = { "admin", "admin" };
        env.put(JMXConnector.CREDENTIALS, creds);
        JMXConnector jmxc = JMXConnectorFactory.connect(url, env);
        MBeanServerConnection conn = jmxc.getMBeanServerConnection();

        ObjectName activeMq = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost");

        BrokerViewMBean mbean = MBeanServerInvocationHandler.newProxyInstance(conn, activeMq, BrokerViewMBean.class,
                true);
        for (ObjectName name : mbean.getQueues()) {
            if (("myQ".equals(name.getKeyProperty("destinationName")))) {
                QueueViewMBean queueMbean = MBeanServerInvocationHandler.newProxyInstance(conn, name,
                        QueueViewMBean.class, true);
                // ="Number of messages in the destination which are yet to be
                // consumed. Potentially dispatched but unacknowledged.")
                System.out.println(queueMbean.getQueueSize());
                // Returns the number of messages that have been delivered
                // (potentially not acknowledged) to consumers.
                System.out.println(queueMbean.getDispatchCount());
            }
        }
    }
}