- 浏览: 1052840 次
- 性别:
- 来自: 上海
文章分类
- 全部博客 (1441)
- 软件思想&演讲 (9)
- 行业常识 (250)
- 时时疑问 (5)
- java/guava/python/php/ruby/R/scala/groovy (213)
- struct/spring/springmvc (37)
- mybatis/hibernate/JPA (10)
- mysql/oracle/sqlserver/db2/mongdb/redis/neo4j/GreenPlum/Teradata/hsqldb/Derby/sakila (268)
- js/jquery/jqueryUi/jqueryEaseyUI/extjs/angulrJs/react/es6/grunt/zepto/raphael (81)
- ZMQ/RabbitMQ/ActiveMQ/JMS/kafka (17)
- lucene/solr/nuth/elasticsearch/MG4J (167)
- html/css/ionic/nodejs/bootstrap (19)
- Linux/shell/centos (56)
- cvs/svn/git/sourceTree/gradle/ant/maven/mantis/docker/Kubernetes (26)
- sonatype nexus (1)
- tomcat/jetty/netty/jboss (9)
- 工具 (17)
- ETL/SPASS/MATLAB/RapidMiner/weka/kettle/DataX/Kylin (11)
- hadoop/spark/Hbase/Hive/pig/Zookeeper/HAWQ/cloudera/Impala/Oozie (190)
- ios/swift/android (9)
- 机器学习&算法&大数据 (18)
- Mesos是Apache下的开源分布式资源管理框架 (1)
- echarts/d3/highCharts/tableau (1)
- 行业技能图谱 (1)
- 大数据可视化 (2)
- tornado/ansible/twisted (2)
- Nagios/Cacti/Zabbix (0)
- eclipse/intellijIDEA/webstorm (5)
- cvs/svn/git/sourceTree/gradle/jira/bitbucket (4)
- jsp/jsf/flex/ZKoss (0)
- 测试技术 (2)
- splunk/flunm (2)
- 高并发/大数据量 (1)
- freemarker/vector/thymeleaf (1)
- docker/Kubernetes (2)
- dubbo/ESB/dubboX/wso2 (2)
最新评论
1.1 JMS简介
JMS的全称是Java Message Service,即Java消息服务。它主要用于在生产者和消费者之间进行消息传递,生产者负责产生消息,而消费者负责接收消息。把它应用到实际的业务需求中的话我们可以在特定的时候利用生产者生成一消息,并进行发送,对应的消费者在接收到对应的消息后去完成对应的业务逻辑。对于消息的传递有两种类型,一种是点对点的,即一个生产者和一个消费者一一对应;另一种是发布/订阅模式,即一个生产者产生消息并进行发送后,可以由多个消费者进行接收。
1.2 Spring整合JMS
对JMS做了一个简要介绍之后,接下来就讲一下Spring整合JMS的具体过程。JMS只是一个标准,真正在使用它的时候我们需要有它的具体实现,这里我们就使用Apache的activeMQ来作为它的实现。导入jar包:
Xml代码 收藏代码
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.10</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>jsr250-api</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
</dependencies>
1.2.1 activeMQ准备
既然是使用的apache的activeMQ作为JMS的实现,那么首先我们应该到apache官网上下载activeMQ(http://activemq.apache.org/download.html),进行解压后运行其bin目录下面的activemq.bat文件启动activeMQ。
1.2.2配置ConnectionFactory
ConnectionFactory是用于产生到JMS服务器的链接的,Spring为我们提供了多个ConnectionFactory,有SingleConnectionFactory和CachingConnectionFactory。SingleConnectionFactory对于建立JMS服务器链接的请求会一直返回同一个链接,并且会忽略Connection的close方法调用。CachingConnectionFactory继承了SingleConnectionFactory,所以它拥有SingleConnectionFactory的所有功能,同时它还新增了缓存功能,它可以缓存Session、MessageProducer和MessageConsumer。这里我们使用SingleConnectionFactory来作为示例。
Xml代码 收藏代码
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"/>
Spring提供的ConnectionFactory只是Spring用于管理ConnectionFactory的,真正产生到JMS服务器链接的ConnectionFactory还得是由JMS服务厂商提供,并且需要把它注入到Spring提供的ConnectionFactory中。我们这里使用的是ActiveMQ实现的JMS,所以在我们这里真正的可以产生Connection的就应该是由ActiveMQ提供的ConnectionFactory。所以定义一个ConnectionFactory的完整代码应该如下所示:
Xml代码 收藏代码
<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>
ActiveMQ为我们提供了一个PooledConnectionFactory,通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化,这样可以大大的减少我们的资源消耗。当使用PooledConnectionFactory时,我们在定义一个ConnectionFactory时应该是如下定义:
Xml代码 收藏代码
<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<property name="maxConnections" value="10"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>
1.2.3配置生产者
配置好ConnectionFactory之后我们就需要配置生产者。生产者负责产生消息并发送到JMS服务器,这通常对应的是我们的一个业务逻辑服务实现类。但是我们的服务实现类是怎么进行消息的发送的呢?这通常是利用Spring为我们提供的JmsTemplate类来实现的,所以配置生产者其实最核心的就是配置进行消息发送的JmsTemplate。对于消息发送者而言,它在发送消息的时候要知道自己该往哪里发,为此,我们在定义JmsTemplate的时候需要往里面注入一个Spring提供的ConnectionFactory对象。
Xml代码 收藏代码
<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
在真正利用JmsTemplate进行消息发送的时候,我们需要知道消息发送的目的地,即destination。在Jms中有一个用来表示目的地的Destination接口,它里面没有任何方法定义,只是用来做一个标识而已。当我们在使用JmsTemplate进行消息发送时没有指定destination的时候将使用默认的Destination。默认Destination可以通过在定义jmsTemplate bean对象时通过属性defaultDestination或defaultDestinationName来进行注入,defaultDestinationName对应的就是一个普通字符串。在ActiveMQ中实现了两种类型的Destination,一个是点对点的ActiveMQQueue,另一个就是支持订阅/发布模式的ActiveMQTopic。在定义这两种类型的Destination时我们都可以通过一个name属性来进行构造,如:
Xml代码 收藏代码
<!--这个是队列目的地,点对点的-->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>queue</value>
</constructor-arg>
</bean>
<!--这个是主题目的地,一对多的-->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic"/>
</bean>
定义了一个ProducerService,向Destination发送纯文本消息:
Java代码 收藏代码
import javax.jms.Destination;
public interface ProducerService {
public void sendMessage(Destination destination, final String message);
}
Java代码 收藏代码
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import cn.tzz.jms.activemq.spring.ProducerService;
@Service
public class ProducerServiceImpl implements ProducerService {
@Autowired
private JmsTemplate jmsTemplate;
public void sendMessage(Destination destination, final String message) {
System.out.println("生产者发消息:" + message);
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}
}
我们可以看到在sendMessage方法体里面我们是通过jmsTemplate来发送消息到对应的Destination的。到此,我们生成一个简单的文本消息并把它发送到指定目的地Destination的生产者就配置好了。
1.2.4配置消费者
生产者往指定目的地Destination发送消息后,接下来就是消费者对指定目的地的消息进行消费了。那么消费者是如何知道有生产者发送消息到指定目的地Destination了呢?这是通过Spring为我们封装的消息监听容器MessageListenerContainer实现的,它负责接收信息,并把接收到的信息分发给真正的MessageListener进行处理。每个消费者对应每个目的地都需要有对应的MessageListenerContainer。对于消息监听容器而言,除了要知道监听哪个目的地之外,还需要知道到哪里去监听,也就是说它还需要知道去监听哪个JMS服务器,这是通过在配置MessageConnectionFactory的时候往里面注入一个ConnectionFactory来实现的。所以我们在配置一个MessageListenerContainer的时候有三个属性必须指定,一个是表示从哪里监听的ConnectionFactory;一个是表示监听什么的Destination;一个是接收到消息以后进行消息处理的MessageListener。Spring一共为我们提供了两种类型的MessageListenerContainer,SimpleMessageListenerContainer和DefaultMessageListenerContainer。
SimpleMessageListenerContainer会在一开始的时候就创建一个会话session和消费者Consumer,并且会使用标准的JMS MessageConsumer.setMessageListener()方法注册监听器让JMS提供者调用监听器的回调函数。它不会动态的适应运行时需要和参与外部的事务管理。兼容性方面,它非常接近于独立的JMS规范,但一般不兼容Java EE的JMS限制。
大多数情况下我们还是使用的DefaultMessageListenerContainer,跟SimpleMessageListenerContainer相比,DefaultMessageListenerContainer会动态的适应运行时需要,并且能够参与外部的事务管理。它很好的平衡了对JMS提供者要求低、先进功能如事务参与和兼容Java EE环境。
定义处理消息的MessageListener
要定义处理消息的MessageListener我们只需要实现JMS规范中的MessageListener接口就可以了。MessageListener接口中只有一个方法onMessage方法,当接收到消息的时候会自动调用该方法。
Java代码 收藏代码
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 textMsg = (TextMessage) message;
try {
System.out.println("\t\t接收到消息:" + textMsg.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
有了MessageListener之后我们就可以在Spring的配置文件中配置一个消息监听容器了。
Xml代码 收藏代码
<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="cn.tzz.jms.activemq.spring.ConsumerMessageListener"/>
<!-- 消息监听容器 -->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="consumerMessageListener" />
</bean>
我们可以看到我们定义了一个名叫queue的ActiveMQQueue目的地,我们的监听器就是监听了发送到这个目的地的消息。
至此我们的生成者和消费者都配置完成了,这也就意味着我们的整合已经完成了。这个时候完整的Spring的配置文件应该是这样的:
Xml代码 收藏代码
<context:annotation-config />
<context:component-scan base-package="package cn.tzz.jms.activemq.spring" />
<!-- ActiveMQ -->
<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<property name="maxConnections" value="10"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>
<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory" />
</bean>
<!--消息队列的目的地,点对点的 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>queue</value>
</constructor-arg>
</bean>
<!--这个是主题目的地,一对多的 -->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic" />
</bean>
<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="cn.tzz.jms.activemq.spring.ConsumerMessageListener"/>
<!-- 消息监听容器 -->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="consumerMessageListener" />
</bean>
测试:
Java代码 收藏代码
import javax.jms.Destination;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import cn.tzz.jms.activemq.spring.ProducerService;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("/applicationContext.xml")
public class ProducerConsumerTest {
@Autowired
private ProducerService producerService;
@Autowired
@Qualifier("queueDestination")
private Destination destination;
@Test
public void testSend() {
for (int i = 0; i < 5; i++) {
producerService.sendMessage(destination, "消息--" + i);
}
}
}
上面的测试代码中我们利用生产者发送了5个消息,消费者应该可以接收到这两个消息。运行测试代码后控制台输出如下:
生产者发消息:消息--0
接收到消息:消息--0
生产者发消息:消息--1
接收到消息:消息--1
生产者发消息:消息--2
接收到消息:消息--2
生产者发消息:消息--3
接收到消息:消息--3
生产者发消息:消息--4
接收到消息:消息--4
JMS的全称是Java Message Service,即Java消息服务。它主要用于在生产者和消费者之间进行消息传递,生产者负责产生消息,而消费者负责接收消息。把它应用到实际的业务需求中的话我们可以在特定的时候利用生产者生成一消息,并进行发送,对应的消费者在接收到对应的消息后去完成对应的业务逻辑。对于消息的传递有两种类型,一种是点对点的,即一个生产者和一个消费者一一对应;另一种是发布/订阅模式,即一个生产者产生消息并进行发送后,可以由多个消费者进行接收。
1.2 Spring整合JMS
对JMS做了一个简要介绍之后,接下来就讲一下Spring整合JMS的具体过程。JMS只是一个标准,真正在使用它的时候我们需要有它的具体实现,这里我们就使用Apache的activeMQ来作为它的实现。导入jar包:
Xml代码 收藏代码
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.10</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>jsr250-api</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
</dependencies>
1.2.1 activeMQ准备
既然是使用的apache的activeMQ作为JMS的实现,那么首先我们应该到apache官网上下载activeMQ(http://activemq.apache.org/download.html),进行解压后运行其bin目录下面的activemq.bat文件启动activeMQ。
1.2.2配置ConnectionFactory
ConnectionFactory是用于产生到JMS服务器的链接的,Spring为我们提供了多个ConnectionFactory,有SingleConnectionFactory和CachingConnectionFactory。SingleConnectionFactory对于建立JMS服务器链接的请求会一直返回同一个链接,并且会忽略Connection的close方法调用。CachingConnectionFactory继承了SingleConnectionFactory,所以它拥有SingleConnectionFactory的所有功能,同时它还新增了缓存功能,它可以缓存Session、MessageProducer和MessageConsumer。这里我们使用SingleConnectionFactory来作为示例。
Xml代码 收藏代码
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"/>
Spring提供的ConnectionFactory只是Spring用于管理ConnectionFactory的,真正产生到JMS服务器链接的ConnectionFactory还得是由JMS服务厂商提供,并且需要把它注入到Spring提供的ConnectionFactory中。我们这里使用的是ActiveMQ实现的JMS,所以在我们这里真正的可以产生Connection的就应该是由ActiveMQ提供的ConnectionFactory。所以定义一个ConnectionFactory的完整代码应该如下所示:
Xml代码 收藏代码
<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>
ActiveMQ为我们提供了一个PooledConnectionFactory,通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化,这样可以大大的减少我们的资源消耗。当使用PooledConnectionFactory时,我们在定义一个ConnectionFactory时应该是如下定义:
Xml代码 收藏代码
<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<property name="maxConnections" value="10"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>
1.2.3配置生产者
配置好ConnectionFactory之后我们就需要配置生产者。生产者负责产生消息并发送到JMS服务器,这通常对应的是我们的一个业务逻辑服务实现类。但是我们的服务实现类是怎么进行消息的发送的呢?这通常是利用Spring为我们提供的JmsTemplate类来实现的,所以配置生产者其实最核心的就是配置进行消息发送的JmsTemplate。对于消息发送者而言,它在发送消息的时候要知道自己该往哪里发,为此,我们在定义JmsTemplate的时候需要往里面注入一个Spring提供的ConnectionFactory对象。
Xml代码 收藏代码
<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
在真正利用JmsTemplate进行消息发送的时候,我们需要知道消息发送的目的地,即destination。在Jms中有一个用来表示目的地的Destination接口,它里面没有任何方法定义,只是用来做一个标识而已。当我们在使用JmsTemplate进行消息发送时没有指定destination的时候将使用默认的Destination。默认Destination可以通过在定义jmsTemplate bean对象时通过属性defaultDestination或defaultDestinationName来进行注入,defaultDestinationName对应的就是一个普通字符串。在ActiveMQ中实现了两种类型的Destination,一个是点对点的ActiveMQQueue,另一个就是支持订阅/发布模式的ActiveMQTopic。在定义这两种类型的Destination时我们都可以通过一个name属性来进行构造,如:
Xml代码 收藏代码
<!--这个是队列目的地,点对点的-->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>queue</value>
</constructor-arg>
</bean>
<!--这个是主题目的地,一对多的-->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic"/>
</bean>
定义了一个ProducerService,向Destination发送纯文本消息:
Java代码 收藏代码
import javax.jms.Destination;
public interface ProducerService {
public void sendMessage(Destination destination, final String message);
}
Java代码 收藏代码
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import cn.tzz.jms.activemq.spring.ProducerService;
@Service
public class ProducerServiceImpl implements ProducerService {
@Autowired
private JmsTemplate jmsTemplate;
public void sendMessage(Destination destination, final String message) {
System.out.println("生产者发消息:" + message);
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}
}
我们可以看到在sendMessage方法体里面我们是通过jmsTemplate来发送消息到对应的Destination的。到此,我们生成一个简单的文本消息并把它发送到指定目的地Destination的生产者就配置好了。
1.2.4配置消费者
生产者往指定目的地Destination发送消息后,接下来就是消费者对指定目的地的消息进行消费了。那么消费者是如何知道有生产者发送消息到指定目的地Destination了呢?这是通过Spring为我们封装的消息监听容器MessageListenerContainer实现的,它负责接收信息,并把接收到的信息分发给真正的MessageListener进行处理。每个消费者对应每个目的地都需要有对应的MessageListenerContainer。对于消息监听容器而言,除了要知道监听哪个目的地之外,还需要知道到哪里去监听,也就是说它还需要知道去监听哪个JMS服务器,这是通过在配置MessageConnectionFactory的时候往里面注入一个ConnectionFactory来实现的。所以我们在配置一个MessageListenerContainer的时候有三个属性必须指定,一个是表示从哪里监听的ConnectionFactory;一个是表示监听什么的Destination;一个是接收到消息以后进行消息处理的MessageListener。Spring一共为我们提供了两种类型的MessageListenerContainer,SimpleMessageListenerContainer和DefaultMessageListenerContainer。
SimpleMessageListenerContainer会在一开始的时候就创建一个会话session和消费者Consumer,并且会使用标准的JMS MessageConsumer.setMessageListener()方法注册监听器让JMS提供者调用监听器的回调函数。它不会动态的适应运行时需要和参与外部的事务管理。兼容性方面,它非常接近于独立的JMS规范,但一般不兼容Java EE的JMS限制。
大多数情况下我们还是使用的DefaultMessageListenerContainer,跟SimpleMessageListenerContainer相比,DefaultMessageListenerContainer会动态的适应运行时需要,并且能够参与外部的事务管理。它很好的平衡了对JMS提供者要求低、先进功能如事务参与和兼容Java EE环境。
定义处理消息的MessageListener
要定义处理消息的MessageListener我们只需要实现JMS规范中的MessageListener接口就可以了。MessageListener接口中只有一个方法onMessage方法,当接收到消息的时候会自动调用该方法。
Java代码 收藏代码
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 textMsg = (TextMessage) message;
try {
System.out.println("\t\t接收到消息:" + textMsg.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
有了MessageListener之后我们就可以在Spring的配置文件中配置一个消息监听容器了。
Xml代码 收藏代码
<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="cn.tzz.jms.activemq.spring.ConsumerMessageListener"/>
<!-- 消息监听容器 -->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="consumerMessageListener" />
</bean>
我们可以看到我们定义了一个名叫queue的ActiveMQQueue目的地,我们的监听器就是监听了发送到这个目的地的消息。
至此我们的生成者和消费者都配置完成了,这也就意味着我们的整合已经完成了。这个时候完整的Spring的配置文件应该是这样的:
Xml代码 收藏代码
<context:annotation-config />
<context:component-scan base-package="package cn.tzz.jms.activemq.spring" />
<!-- ActiveMQ -->
<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<property name="maxConnections" value="10"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
</bean>
<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory" />
</bean>
<!--消息队列的目的地,点对点的 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>queue</value>
</constructor-arg>
</bean>
<!--这个是主题目的地,一对多的 -->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic" />
</bean>
<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="cn.tzz.jms.activemq.spring.ConsumerMessageListener"/>
<!-- 消息监听容器 -->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="consumerMessageListener" />
</bean>
测试:
Java代码 收藏代码
import javax.jms.Destination;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import cn.tzz.jms.activemq.spring.ProducerService;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("/applicationContext.xml")
public class ProducerConsumerTest {
@Autowired
private ProducerService producerService;
@Autowired
@Qualifier("queueDestination")
private Destination destination;
@Test
public void testSend() {
for (int i = 0; i < 5; i++) {
producerService.sendMessage(destination, "消息--" + i);
}
}
}
上面的测试代码中我们利用生产者发送了5个消息,消费者应该可以接收到这两个消息。运行测试代码后控制台输出如下:
生产者发消息:消息--0
接收到消息:消息--0
生产者发消息:消息--1
接收到消息:消息--1
生产者发消息:消息--2
接收到消息:消息--2
生产者发消息:消息--3
接收到消息:消息--3
生产者发消息:消息--4
接收到消息:消息--4
发表评论
-
ActiveMQ安装与使用
2017-12-11 18:07 796一 .安装运行ActiveMQ: ... -
MQ的使用
2017-12-11 18:18 415安装环境:linux redhat activemq版本 ... -
mq常用命令
2017-12-11 18:20 506创建队列管理器 crtmqm qmgrname 删 ... -
Activemq的端口监听介绍及端口优化
2018-01-07 09:06 594在activemq的conf/activemq.x ... -
ActiveMq启动停止
2017-10-25 10:05 1270Linux操作系统 进入bin目录下,64位操作系统可以进 ... -
RabbitMq、ActiveMq、ZeroMq、kafka之间的比较,资料汇总
2017-07-21 14:21 993MQ框架非常之多,比较流行的有RabbitMq、Active ... -
使用activemq
2017-06-29 14:32 459说说在项目里是怎么使用activemq(简称为amq)进行通 ... -
activeMQ启动失败61616端口被占用问题
2017-03-27 23:19 1733通过netstat -aon | findstr " ... -
Kafka 对比 ActiveMQ
2016-12-13 13:00 1775Kafka 是LinkedIn 开发的一个高性能、分布式的 ... -
ActiveMQ的工作原理
2016-11-14 10:19 719如图所示 首先来看本地通讯的情况,应用程序A和应用程 ... -
Spring整合JMS的三种监听
2016-12-30 11:47 847在Spring整合JMS的应用中我们在定义消息监听器的时候一共 ... -
Spring整合JMS的MessageConverter
2016-12-30 11:43 458消息转换器MessageConverter Me ... -
Spring整合JMS
2016-12-30 11:44 426Spring提供了一个JmsTransac ... -
ActiveMQ入门
2017-01-01 23:34 5041.下载ActiveMQ 去官方网站下载:http://ac ... -
zmq的介绍
2015-12-11 11:52 695http://blog.csdn.net/lfsf802/ar ... -
kafka特点
2015-11-23 02:07 1121一、入门 1、简介 Kafka is a di ...
相关推荐
Spring整合JMS基于ActiveMQ实现是一项常见的企业级应用开发任务,它涉及到Spring框架、Java消息服务(JMS)以及ActiveMQ消息中间件的使用。在本文中,我们将深入探讨这三个关键概念,以及如何将它们有效地结合在一起...
标题 "Spring-ActiveMQ-jar" 指的是一个与Spring框架和ActiveMQ集成相关的Java归档(JAR)文件。这个JAR文件通常包含了Spring框架支持ActiveMQ所需的所有类和资源,使得开发者能够利用Spring的IoC(Inversion of ...
详细内容: SpringJMS整合ActiveMQ.doc 详细说明文档 apache-activemq-5.8.0-bin.zip ActiveMQ安装包 JMSTest.rar MyEclipse8.5下web工程
综上所述,Spring整合JMS和ActivemQ提供了一套完整的解决方案,帮助开发者轻松地在应用中实现消息的发送和接收。通过这种方式,可以构建出高可用、松耦合、可扩展的分布式系统,提高系统的稳定性和响应速度。在实际...
ActiveMQ是Apache下的开源项目,完全支持JMS1.1和J2EE1.4规范的JMS Provider实现。 特点: 1、支持多种语言编写客户端 2、对spring的支持,很容易和spring整合 3、支持多种传输协议:TCP,SSL,NIO,UDP等 4、支持...
<artifactId>activemq-client ``` 接下来,配置ActiveMQ连接信息。在`application.properties`或`application.yml`中,设置以下内容: ```properties # application.properties 示例 spring.activemq.broker-url...
7. **activemq-all-5.15.0.jar**: 这是ActiveMQ的全功能JAR包,包含了所有需要的组件和服务,用于在Spring中直接使用ActiveMQ。 整合Spring和ActiveMQ的步骤通常包括以下几步: 1. 添加相关依赖到项目中,即上述...
《ActiveMQ与Spring整合——深度解析5.5.0版本》 在Java消息服务(Java Message Service,简称JMS)领域,Apache ActiveMQ是广泛使用的开源消息代理和集成框架。它支持多种协议,如OpenWire、AMQP、STOMP、MQTT等,...
标题中的“Spring整合Blazeds实现ActiveMQ JMS消息服务”指的是在Java应用程序中使用Spring框架与Blazeds(一个Flex和Java之间的消息传递中间件)集成,通过ActiveMQ(一个流行的开源JMS提供商)来实现消息队列服务...
以上就是ActiveMQ与Spring整合的基本流程。在实际项目中,你可能还需要处理事务管理、监听消费者、错误处理等复杂场景。通过这种方式,你可以构建一个健壮的、可扩展的消息传递系统,提升应用的可靠性和可维护性。 ...
"spring-activeMQ-demo"是一个基于Spring的示例项目,用于演示如何将Spring与ActiveMQ结合使用。项目结构简单明了,主要涉及以下关键组件: 1. **Maven配置**:`mvn clean install` 是使用Maven构建工具进行全新...
总结起来,"apache-activemq-5.3.0-bin.zip"包含了使用Apache ActiveMQ所需的所有组件,开发者可以通过解压、配置和启动服务,将其整合到自己的应用程序中,实现高效、可靠的消息传递。同时,通过持续监控和调优,...
**ActiveMQ** 是Apache软件基金会开发的一个开源消息代理和队列服务器,它是JMS的一个实现。ActiveMQ支持多种协议,如OpenWire、STOMP、AMQP、MQTT和WS-Notification,以及多种语言的客户端,包括Java、C++、Python...
将Spring与ActiveMQ整合,可以轻松地在Spring应用中实现消息队列的功能,提高系统的可扩展性和可靠性。 首先,让我们了解Spring框架如何支持消息传递。Spring提供了JmsTemplate类,这是一个模板类,用于简化发送和...
在Spring整合JMS的过程中,我们通常会使用`DefaultMessageListenerContainer`或`SimpleMessageListenerContainer`类来消费消息。这些容器提供了与JMS提供者(如ActiveMQ、RabbitMQ等)的连接管理,并且支持事务性...
《ActiveMQ与Spring整合——深度解析activemq-spring-1.3.jar》 在Java消息服务(JMS)领域,Apache ActiveMQ是一款广泛应用的消息中间件,它支持多种协议,如OpenWire、AMQP、STOMP等,为分布式系统提供可靠的消息...
总结来说,Spring整合JMS和ActiveMQ的过程包括:配置ConnectionFactory,定义Destination,创建MessageListener容器,以及使用JmsTemplate发送消息。通过这种方式,你可以构建一个健壮的、异步的消息传递系统,提高...
<artifactId>activemq-client <version>5.x.x ``` 接下来,配置Spring的XML上下文文件,定义ConnectionFactory、目的地以及JmsTemplate等: ```xml <beans xmlns="http://www.springframework.org/schema/...
在 Spring Boot 中整合 ActiveMQ,可以使用 Spring 的 JmsTemplate 或者 @JmsListener 注解来发送和接收消息,实现应用间的异步通信和解耦。 整合以上技术的过程通常包括以下步骤: - 添加对应库的依赖,如 spring...
除了基本的发送和接收消息,ActiveMQ与Spring整合还可以实现更复杂的功能,比如消息确认(acknowledgement)、事务支持、消息分发策略(例如负载均衡)以及消息优先级等。此外,你还可以使用ActiveMQ的管理控制台来...