Spring JMS 接收主题消息

6

我正在制作一个简单的教程。我有一个发布者发送主题消息和订阅者接收它。当我启动应用程序时,Spring配置文件会加载,然后我会得到以下错误:

    2011-10-20 21:50:39,340 DEBUG [org.springframework.jms.support.destination.JndiDestinationResolver] - Located object with JNDI name [RateTopic]
2011-10-20 21:50:39,340 DEBUG [org.springframework.jms.support.destination.JndiDestinationResolver] - Located object with JNDI name [RateTopic]
2011-10-20 21:50:39,340 DEBUG [org.springframework.jms.connection.CachingConnectionFactory] - Closing cached Session: ActiveMQSession {id=ID:Reverb0253-PC-62259-1319161839013-0:1:3,started=true}
2011-10-20 21:50:39,340 DEBUG [org.springframework.jms.connection.CachingConnectionFactory] - Closing cached Session: ActiveMQSession {id=ID:Reverb0253-PC-62259-1319161839013-0:1:2,started=true}
2011-10-20 21:50:44,348 WARN [org.springframework.jms.listener.DefaultMessageListenerContainer] - Setup of JMS message listener invoker failed for destination 'RateTopic' - trying to recover. Cause: Destination [RateTopic] is not of expected type [javax.jms.Queue]
org.springframework.jms.support.destination.DestinationResolutionException: Destination [RateTopic] is not of expected type [javax.jms.Queue]
    at org.springframework.jms.support.destination.JndiDestinationResolver.validateDestination(JndiDestinationResolver.java:147)
    at org.springframework.jms.support.destination.JndiDestinationResolver.resolveDestinationName(JndiDestinationResolver.java:112)
    at org.springframework.jms.support.destination.JmsDestinationAccessor.resolveDestinationName(JmsDestinationAccessor.java:100)
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.createListenerConsumer(AbstractPollingMessageListenerContainer.java:221)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.initResourcesIfNecessary(DefaultMessageListenerContainer.java:1081)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1057)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1050)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:947)
    at java.lang.Thread.run(Thread.java:722)

为什么Spring认为它应该是队列而不是主题

我的JNDI文件如下所示

java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory
java.naming.provider.url = tcp://localhost:61616
java.naming.security.principal=system
java.naming.security.credentials=manager
connectionFactoryNames = TopicCF
topic.RateTopic = RateTopic

Spring配置文件是什么?

<bean id="jndiTemplate" class="org.springframework.jndi.JndiTemplate">
    <property name="environment">
        <props>
            <prop key="java.naming.factory.initial">
                org.apache.activemq.jndi.ActiveMQInitialContextFactory
            </prop>
            <prop key="java.naming.provider.url">tcp://localhost:61616</prop>
            <prop key="java.naming.security.principal">system</prop>
            <prop key="java.naming.security.credentials">manager</prop>
        </props>
    </property>
</bean>

<bean id="jndiTopicConnFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
    <property name="jndiTemplate" ref="jndiTemplate"/>
    <!-- JNDI name of connection factory as defined by provider -->
    <property name="jndiName" value="TopicCF"/>
</bean>

<bean id="topicConnFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    <property name="targetConnectionFactory" ref="jndiTopicConnFactory"/>
    <!-- Number of sessions that will be cached -->
    <property name="sessionCacheSize" value="1"/>
</bean>

<bean id="destinationResolver" class="org.springframework.jms.support.destination.JndiDestinationResolver">
    <property name="jndiTemplate" ref="jndiTemplate"/>
    <property name="cache" value="true"/>
    <!-- do not create a dynamic destination if the destination name is not found in JNDI -->
    <property name="fallbackToDynamicDestination" value="false"/>
</bean>

<bean id="messageListener" class="com.merc.springjmspubsublenderborrower.TBorrower"/>

<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="topicConnFactory"/>
    <property name="destinationResolver" ref="destinationResolver"/>
    <property name="concurrentConsumers" value="3" />
    <property name="destinationName" value="RateTopic"/>
    <property name="messageListener" ref="messageListener" />
    <property name="sessionAcknowledgeModeName" value="AUTO_ACKNOWLEDGE"/>
</bean>

我的订阅者实现了MessageListener接口

@Override
public void onMessage(Message message) {
    try {
        // Get the data from the message
        BytesMessage msg = (BytesMessage) message;
        double newRate = msg.readDouble();
        // If the rate is at least 1 point lower than the current rate, then
        //recommend refinancing
        if ((currentRate - newRate) >= 1.0) {
            System.out.println(
                    "New rate = " + newRate + " - Consider refinancing loan");
        } else {
            System.out.println("New rate = " + newRate + " - Keep existing loan");
        }
        System.out.println("\nWaiting for rate updates...");
    } catch (Exception ex) {
        ex.printStackTrace(System.out);
        System.exit(1);
    }
}

public static void main(String argv[]) {

    ApplicationContext ctx = new ClassPathXmlApplicationContext("app-context.xml");

    try {
        // Run until enter is pressed
        BufferedReader stdin = new BufferedReader(new InputStreamReader(System.in));
        System.out.println("TBorrower application started");
        System.out.println("Press enter to quit application");
        stdin.readLine();
    } catch (IOException ioe) {
        ioe.printStackTrace();
    }
}

主题模式下,不要将concurrentConsumers设置为大于1Spring建议:不要增加主题的并发消费者数量。这会导致对同一条消息的并发消费,这几乎从来都是不可取的。 - Mohsen
主题的整个意义在于有多个消费者,这样多个进程都可以获得相同的消息。 - hdost
晚到派对了,但多个消费者的概念是为了区分消费者。这样每个消费者只有在其感兴趣的消息时才会消费。 - Olgun Kaya
1个回答

16
您正在尝试从主题中消费,但是您没有在DefaultMessageListenerContainer上设置pubSubDomain属性,它默认为“false”,意味着点对点,意味着队列而不是主题。因此出现错误消息告诉您RateTopic不是javax.jms.Queue

谢谢你的建议,它起作用了。我浏览了很多在线教程,但没有一个在监听器上有pubSubDomain,只有在jmsTemplate上有。 - user373201
@user373201,你能把你的订阅者Java代码复制粘贴一下吗?我正在尝试实现我的订阅者,谢谢。 - rayman
2
@rayman:你最好提出一个新问题,并包括你尝试过什么以及哪些部分不起作用。 - Ryan Stewart

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接