spring下queue与持久订阅topic实现

使用JMS Topic的发布/订阅式消息发送在几方面类似于点到点的消息发送。以下是两种消息发送方式共同的特点：

•消息发送可以是面向对象的，允许将整个的对象作为消息发送。
•消息发送可以是事务性的。
•消息发送可以是同步或异步的。
•消息发送可以与基础的第三方产品集成。
•消息可以发送给在消息发送时不在运行的消息消费者（即QueueReceiver或TopicSubscriber）。
•消息一旦被递送到队列或主题，发送消息的函数调用就会立即返回。
•可以显式地或自动地确认收到消息。


发布/订阅式消息发送与点到点消息发送之间也有几点不同：

•发布/订阅式消息发送是一对多的，而点到点消息发送是一对一的。
•发布的消息只递送给Topic当前的预订者。客户只能接收到他向一个Topic预订了的那些消息。而在点到点消息发送中，永久的消息将一直在Queue中，直到它超时或者某个接收者来取走该消息。
•发布/订阅式消息发送中的永久消息是由“耐久的预订”提供的。JMS提供者存储由于预订者出于某种原因不可用而无法递送给预订者的消息。在下次预订者连接上的时候，这些存储的消息将会被递送给他。这确保了客户预订一个Topic之后，所有发布的消息都会递送给他，哪怕该预订者不是总在运行。如果预订不是耐久的，在预订者掉线时发布的任何消息都不会递送到预订者。

本文章所描述的QUEUE与TOPIC实现如下：
附件中的jmsproducer压缩包中JAVA工程为生成QUEUE与TOPIC的例子，jmsconsumer压缩包中JAVA工程为消费QUEUE与TOPIC例子。
另外每个工程下LIB目录下的activemq-common.jar为打包的公共类。
在jmsproducer工程中的activemq-producer.xml文件中为持久订阅者提供clientID
配置如下：

<bean id="durableTopicJmsConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61617?wireFormat.maxInactivityDuration=0"/>	
		<property name="clientIDPrefix" value="topicClient"/>
	</bean>

clientIDPrefix指定了consumer的clientID的前缀，符合此前缀规则的consumer将成为该TOPIC的持久订阅者。

对应的jmsconsumer工程的TOPIC container配置：

<bean id="topicListenerContainerA"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="durableTopicJmsConnectionFactory" />
		<property name="destination" ref="TOPIC" />
		<property name="messageListener" ref="topicListenerA" />
		<property name="sessionAcknowledgeModeName" value="AUTO_ACKNOWLEDGE"/>
		<property name="receiveTimeout" value="20000"/>
		<property name="subscriptionDurable" value="true"/>
		<property name="clientId" value="topicClientA"/>
		<property name="durableSubscriptionName" value="topicClientA" />
	</bean>

若将subscriptionDurable设置为ture，则必须提供clientId。

评论

2 楼 sharron5 2010-12-27  

很明显是端口被占用了

1 楼 vtrtbb 2010-12-10  

楼主，你这个我怎么调试都不行 啊，报这个错误：


ERROR [broker.BrokerService]-491 Failed to start ActiveMQ JMS Message Broker. Reason: java.io.IOException: Failed to bind to server socket: tcp://localhost:61616 due to: java.net.BindException: Address already in use: JVM_Bind
java.io.IOException: Failed to bind to server socket: tcp://localhost:61616 due to: java.net.BindException: Address already in use: JVM_Bind