Spring中使用JMS模板/消息订阅服务器的JMS主题订阅服务器

Question

我有一个使用ActiveMQ的JMS生产者/订阅者的简单Spring应用程序，配置如下：

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://localhost:61616" />
    <property name="userName" value="user" />
    <property name="password" value="password" />
</bean>
<bean id="messageDestination" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg value="messageQueue1" />
</bean>

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="sessionAcknowledgeModeName" value="CLIENT_ACKNOWLEDGE">
    </property>
</bean>

<bean id="springJmsProducer" class="SpringJmsProducer">
    <property name="destination" ref="messageDestination" />
    <property name="jmsTemplate" ref="jmsTemplate" />
</bean>

<bean id="springJmsConsumer" class="SpringJmsConsumer">
    <property name="destination" ref="messageDestination" />
    <property name="jmsTemplate" ref="jmsTemplate" />
</bean>

public class SpringJmsProducer {
private JmsTemplate jmsTemplate;
private Destination destination;

public JmsTemplate getJmsTemplate() {
    return jmsTemplate;
}

public void setJmsTemplate(JmsTemplate jmsTemplate) {
    this.jmsTemplate = jmsTemplate;
}

public Destination getDestination() {
    return destination;
}

public void setDestination(Destination destination) {
    this.destination = destination;
}

public void sendMessage(final String msg) {
    jmsTemplate.send(destination, new MessageCreator() {
        public Message createMessage(Session session) throws JMSException {
            return session.createTextMessage(msg);
        }});        
 }
}

public class SpringJmsConsumer {
private JmsTemplate jmsTemplate;
private Destination destination;

public JmsTemplate getJmsTemplate() {
    return jmsTemplate;
}

public void setJmsTemplate(JmsTemplate jmsTemplate) {
    this.jmsTemplate = jmsTemplate;
}

public Destination getDestination() {
    return destination;
}

public void setDestination(Destination destination) {
    this.destination = destination;
}

public String receiveMessage() throws JMSException {
    TextMessage textMessage =(TextMessage) jmsTemplate.receive(destination);        
    return textMessage.getText();
 }
}

我试过所有可能的解决办法，但没有一个奏效。我们非常感谢任何帮助

Answer 1

如果您希望消费者在开始之前收到发送到主题的消息，您有2个选择：

1.使用Activemq追溯使用者

背景一个追溯使用者只是一个常规的JMS主题使用者，它指示在订阅开始时，每一次尝试都应该用来回到过去，并发送使用者可能错过的任何旧消息（或上一次在该主题上发送的消息）。

topic = new ActiveMQTopic("TEST.Topic?consumer.retroactive=true");

2.使用持久订户：

请注意，持久订阅者在第二次运行开始之前接收发送到主题的消息

http://activemq.apache.org/manage-production-subscribers.html

这可以异步使用DefaultMessageListenerContainer

<bean id="jmsContainer" destroy-method="shutdown"
    class="org.springframework.jms.listener.DefaultMessageListenerContainer" >
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="destination" ref="messageDestination" />
    <property name="messageListener" ref="messageListenerAdapter" />
    <property name="sessionAcknowledgeModeName" value="CLIENT_ACKNOWLEDGE" />
    <property name="subscriptionDurable" value="true" />
    <property name="clientId" value="UniqueClientId" />
</bean>

<bean id="messageListenerAdapter"
    class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
    <constructor-arg ref="springJmsConsumer" />
</bean>
<bean id="springJmsConsumer" class="SpringJmsConsumer">
</bean>

并更新您的使用者：

public class SpringJmsConsumer implements javax.jms.MessageListener {

    public void onMessage(javax.jms.Message message) {
        // treat message;
        message.acknowledge();
    }
}

更新以使用

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;

public class SpringJmsConsumer {

    private Connection conn;
    private TopicSubscriber topicSubscriber;

    public SpringJmsConsumer(ConnectionFactory connectionFactory, Topic destination ) {
        conn = connectionFactory.createConnection("user", "password");
        Session session = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        topicSubscriber = session.createDurableSubscriber(destination, "UniqueClientId");
        conn.start();
    }

    public String receiveMessage() throws JMSException {
        TextMessage textMessage = (TextMessage) topicSubscriber.receive();
        return textMessage.getText();
    }
}

并更新springJmsConsumer

<bean id="springJmsConsumer" class="SpringJmsConsumer">
    <constructor-arg ref="connectionFactory" />
    <constructor-arg ref="messageDestination" />
</bean>

请注意，此代码不管理连接故障。