东莞市长安镇做网站/百度指数网

上篇文章讲的是queue点对点模式,这篇讲下如何进行topic订阅模式

上篇用的spring是3的，有的老，会报错，所以我把spring升级为4的，pom.xml如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.cwh.activeMQ</groupId><artifactId>activeMQ</artifactId><version>0.0.1-SNAPSHOT</version><packaging>war</packaging><properties><junit.version>4.10</junit.version><spring.version>4.1.6.RELEASE</spring.version><mysql.version>5.1.22</mysql.version><mybatis.version>3.2.7</mybatis.version><mybatis-spring.version>1.2.2</mybatis-spring.version><aspectjweaver.version>1.8.1</aspectjweaver.version><jackson-core-asl.version>1.9.2</jackson-core-asl.version><jackson-mapper-asl.version>1.9.2</jackson-mapper-asl.version><javax.servlet-api.version>3.1.0</javax.servlet-api.version><log4j.version>1.2.16</log4j.version><commons-dbcp.version>1.4</commons-dbcp.version><fastjson.version>1.1.37</fastjson.version><jstl.version>1.1.0</jstl.version><standard.version>1.1.0</standard.version><commons-configuration.version>1.6</commons-configuration.version><commons-fileupload.version>1.2.1</commons-fileupload.version></properties><!-- <dependencyManagement> --><dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>${junit.version}</version><scope>test</scope></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>${spring.version}</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>${mysql.version}</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-core</artifactId><version>${spring.version}</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>${spring.version}</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-tx</artifactId><version>${spring.version}</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-jdbc</artifactId><version>${spring.version}</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-webmvc</artifactId><version>${spring.version}</version></dependency><dependency><groupId>org.mybatis</groupId><artifactId>mybatis</artifactId><version>${mybatis.version}</version></dependency><dependency><groupId>org.mybatis</groupId><artifactId>mybatis-spring</artifactId><version>${mybatis-spring.version}</version></dependency><dependency><groupId>org.aspectj</groupId><artifactId>aspectjweaver</artifactId><version>${aspectjweaver.version}</version></dependency><dependency><groupId>org.codehaus.jackson</groupId><artifactId>jackson-core-asl</artifactId><version>${jackson-core-asl.version}</version></dependency><dependency><groupId>org.codehaus.jackson</groupId><artifactId>jackson-mapper-asl</artifactId><version>${jackson-mapper-asl.version}</version></dependency><dependency><groupId>javax.servlet</groupId><artifactId>javax.servlet-api</artifactId><version>${javax.servlet-api.version}</version><scope>provided</scope></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>${log4j.version}</version></dependency><dependency><groupId>commons-fileupload</groupId><artifactId>commons-fileupload</artifactId><version>${commons-fileupload.version}</version></dependency><dependency><groupId>commons-dbcp</groupId><artifactId>commons-dbcp</artifactId><version>${commons-dbcp.version}</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><dependency><groupId>jstl</groupId><artifactId>jstl</artifactId><version>${jstl.version}</version></dependency><dependency><groupId>taglibs</groupId><artifactId>standard</artifactId><version>${standard.version}</version></dependency><dependency><groupId>commons-configuration</groupId><artifactId>commons-configuration</artifactId><version>${commons-configuration.version}</version></dependency><!-- 以下是activeMQ所需的包 --><dependency><groupId>org.apache.xbean</groupId><artifactId>xbean-spring</artifactId><version>3.16</version></dependency><dependency><groupId>com.thoughtworks.xstream</groupId><artifactId>xstream</artifactId><version>1.3.1</version></dependency><dependency>  <groupId>org.springframework</groupId>  <artifactId>spring-jms</artifactId>  <version>${spring.version}</version>  </dependency>  <dependency>  <groupId>org.apache.activemq</groupId>  <artifactId>activemq-all</artifactId>  <version>5.9.0</version>  </dependency>  </dependencies><!-- </dependencyManagement> --><build><plugins><plugin><artifactId>maven-war-plugin</artifactId><configuration><version>3.0</version></configuration></plugin></plugins></build>
</project>

下面进行订阅者模式开发

1、首先对activeMQ.xml这个文件进行修改

添加：

 <!-- 定义订阅模式（topic） --><bean id="demoTopicDestination" class="org.apache.activemq.command.ActiveMQTopic"><!-- 设置消息队列的名字 --><constructor-arg><value>activeMQ+spring(topic)</value></constructor-arg></bean>

 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"><property name="connectionFactory" ref="connectionFactory" /><property name="defaultDestination" ref="demoQueueDestination" /><property name="receiveTimeout" value="10000" /><!-- true是topic，false是queue，默认是false，此处显示写出false --><property name="pubSubDomain" value="true" /></bean>

2、添加配置文件topic_comsumer.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsd"><!-- 配置JMS连接工厂 --><bean id="consumerConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"><property name="brokerURL" value="failover:(tcp://127.0.0.1:61616)" /><property name="useAsyncSend" value="true" /><property name="clientID" value="consumerClienctConnect" /></bean><!-- 定义消息Destination --><bean id="topicDestination"  class="org.apache.activemq.command.ActiveMQTopic"><constructor-arg value="activeMQ+spring(topic)"/></bean><!-- 配置消息消费监听者 --><bean id="consumerMessageListener" class="com.cwh.activeMQ.listener.ConsumerMessageListener" /><!-- 消息订阅客户端1 --><bean id="consumerListenerClient1" class="org.springframework.jms.listener.DefaultMessageListenerContainer"><property name="connectionFactory" ref="consumerConnectionFactory" /><!-- 开启订阅模式 --><property name="pubSubDomain" value="true"/><property name="destination" ref="topicDestination" /><property name="subscriptionDurable" value="true"/><!---这里是设置接收客户端的ID，在持久化时，但这个客户端不在线时，消息就存在数据库里，直到被这个ID的客户端消费掉--><property name="clientId" value="consumerClient1"/><property name="messageListener" ref="consumerMessageListener" /><!-- 消息应答方式Session.AUTO_ACKNOWLEDGE  消息自动签收Session.CLIENT_ACKNOWLEDGE  客户端调用acknowledge方法手动签收Session.DUPS_OK_ACKNOWLEDGE 不必必须签收，消息可能会重复发送--><property name="sessionAcknowledgeMode" value="1"/></bean><!-- 消息订阅客户端2 --><bean id="consumerListenerClient2" class="org.springframework.jms.listener.DefaultMessageListenerContainer"><property name="connectionFactory" ref="consumerConnectionFactory" /><!-- 开启订阅模式 --><property name="pubSubDomain" value="true"/><property name="destination" ref="topicDestination" /><property name="subscriptionDurable" value="true"/><!---这里是设置接收客户端的ID，在持久化时，但这个客户端不在线时，消息就存在数据库里，直到被这个ID的客户端消费掉--><property name="clientId" value="consumerClient2"/><property name="messageListener" ref="consumerMessageListener" /><!-- 消息应答方式Session.AUTO_ACKNOWLEDGE  消息自动签收Session.CLIENT_ACKNOWLEDGE  客户端调用acknowledge方法手动签收Session.DUPS_OK_ACKNOWLEDGE 不必必须签收，消息可能会重复发送--><property name="sessionAcknowledgeMode" value="1"/></bean></beans>

3、添加消费者监听器

package com.cwh.activeMQ.listener;import org.apache.commons.lang.builder.ToStringBuilder;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;public class ConsumerMessageListener implements MessageListener{public void onMessage(Message message) {TextMessage tm = (TextMessage) message;try {System.out.println("---------消息消费---------");System.out.println("消息内容:\t" + tm.getText());System.out.println("消息ID:\t" + tm.getJMSMessageID());System.out.println("消息Destination:\t" + tm.getJMSDestination());System.out.println("---------更多信息---------");System.out.println(ToStringBuilder.reflectionToString(tm));System.out.println("-------------------------");} catch (JMSException e) {e.printStackTrace();}}
}

<form action="${ctxPath }activeMQ/onsendtotopic" method="post">MessageText:<textarea name="message">${time}</textarea><input type="submit" value="topic"></form>

访问http://localhost:8080/activeMQ/activeMQ/main

点击topic发布后，消费者监听器会监听到消息，查看控制台打印如下，打印出有两个订阅者

ok!这样也就完成了topic订阅者模式的简单例子