简介

ActiveMQ 是开源的JMS实现，Geronimo应用服务器就是使用的ActiveMQ提供JMS服务。ActiveMQ5.0相比以前版本提供了一些非常有用的新功能：

1.  AMQ Message Store (Faster Persistence!)

2.  Cursors (To handle very large number of storedmessages)

3.  Blob Messages

4.  Command Agent

5.  Enterprise Integration Patterns via Camel Integration

6.  Logging a warning if you forget to start a Connection

7.  Message Transformation

8.  Mirrored Queues

9.  Flow Control

鉴于目前关于ActiveMQ5.0的文章比较少，故准备写一系列ActiveMQ的使用方面的文章。本篇先从安装开始。

安装

1.  在http://activemq.apache.org/download.html 下载5.0.0发行包，解压到需要安装ActiveMQ的文件夹，记为/path/to/activemq。

2.  unix环境activemq文件夹需要执行权限，执行如下命令  chmod -R755 /path/to/activemq 

启动

1.  window环境运行/path/to/activemq/bin/activemq.bat

2.  unix环境运行/path/to/activemq/bin/activemq

测试

ActiveMQ默认使用的TCP连接端口是61616, 通过查看该端口的信息可以测试ActiveMQ是否成功启动

1.  window环境运行  netstat -an|find "61616"

2.  unix环境运行netstat -an|grep 61616

监控

ActiveMQ5.0版本默认启动时，启动了内置的jetty服务器，提供一个demo应用和用于监控ActiveMQ的admin应用。

admin：http://127.0.0.1:8161/admin/

demo：http://127.0.0.1:8161/demo/

点击demo应用中的“ Market data publisher ”，就会发一些测试的消息。转到admin页面的topics menu下面（queue和topic的区别见 http://andyao.iteye.com/blog/153173 ），可以看到消息在增长。

配置

ActiveMQ5.0的配置文件在/path/to/activemq/conf目录下面。主要配置文件为activemq.xml，具体的配置将在后续文章中详细说明。

不同的需求，不同的环境，需要不同的features，当然需要不同的配置。在这里我只写了最基本的配置，算是抛砖了，希望引出更多关于ActiveMQ的高级配置。
假设已经正确安装ActiveMQ5.0，同时及其IP地址为192.168.1.148，具体使用时可以改为自己的IP。下面讲解的配置实现的features如下：

1.      客户端可以通过tcp://192.168.1.148连接ActiveMQ。

2.      消息持久化保存，重启服务器不会丢失消息。

3.      可以通过http://192.168.1.148:8161/admin监控ActiveMQ服务器

配置

ActiveMQ默认使用的是XML格式配置，从4.0版本开始用MBean的方式实现XML配置，配置文件在${activemq.home}/conf目录下，文件名为activemq.xml。最新的默认配置见
http://svn.apache.org/repos/asf/activemq/trunk/assembly/src/release/conf/activemq.xml。下面为本篇文章使用的配置，及重要部分的解释。

Xml代码

1.      <beans

2.      xmlns="http://www.springframework.org/schema/beans"

3.      xmlns:amq="http://activemq.org/config/1.0"

4.      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd

5.      http://activemq.org/config/1.0 http://activemq.apache.org/schema/activemq-core.xsd

6.      http://activemq.apache.org/camel/schema/spring>

7.

8.      <!-- persistent="true"表示要持久化存储消息，和子元素persistenceAdapter结合使用-->

9.      <!-- dataDirectory默认的存储持久化数据的目录-->

10.  <!-- brokerName设置broker的name，在注意在网络上必须是唯一的-->

11.  <!--更多参考http://activemq.apache.org/xbean-xml-reference-50.html#XBeanXMLReference5.0-brokerelement -->

12.  <broker xmlns="http://activemq.org/config/1.0" brokerName="192.168.1.148" persistent ="true" dataDirectory="${activemq.base}/data" useShutdownHook="false">

13.

14.  <!-- Destination specific policies using destination names or wildcards -->

15.  <!-- wildcards意义见http://activemq.apache.org/wildcards.html -->

16.  <destinationPolicy>

17.  <policyMap>

18.  <policyEntries>

19.  <!--这里使用了wildcards，表示所有以EUCITA开头的topic -->

20.  <policyEntry topic="EUCITA.>" producerFlowControl="false" memoryLimit="10mb">

21.  <!--分发策略-->

22.  <dispatchPolicy>

23.  <!--按顺序分发-->

24.  <strictOrderDispatchPolicy/>

25.  </dispatchPolicy>

26.  <!--恢复策略-->

27.  <subscriptionRecoveryPolicy>

28.  <!--只恢复最后一个message -->

29.  <lastImageSubscriptionRecoveryPolicy/>

30.  </subscriptionRecoveryPolicy>

31.  </policyEntry>

32.  </policyEntries>

33.  </policyMap>

34.  </destinationPolicy>

35.

36.  <!-- The transport connectors ActiveMQ will listen to -->

37.  <transportConnectors>

38.  <transportConnector name="openwire" uri="tcp://192.168.1.148:61616" discoveryUri="multicast://default"/>

39.  <transportConnector name="ssl"     uri="ssl://192.168.1.148:61617"/>

40.  <transportConnector name="stomp"   uri="stomp://192.168.1.148:61613"/>

41.  <transportConnector name="xmpp"    uri="xmpp://192.168.1.148:61222"/>

42.  </transportConnectors>

43.

44.  <!--消息持久化方式-->

45.  <persistenceAdapter>

46.  <amqPersistenceAdapter directory="${activemq.base}/data"/>

47.  </persistenceAdapter>

48.  </broker>

49.

50.  <!-- lets create a command agent to respond to message based admin commands on the ActiveMQ.Agent topic -->

51.  <commandAgent xmlns="http://activemq.org/config/1.0"/>

52.

53.  <!-- An embedded servlet engine for serving up the Admin console -->

54.  <jetty xmlns="http://mortbay.com/schemas/jetty/1.0">

55.  <connectors>

56.  <nioConnector port="8161" />

57.  </connectors>

58.

59.  <handlers>

60.  <webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true" />

61.  <webAppContext contextPath="/demo" resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true" />

62.  </handlers>

63.  </jetty>

64.  </beans>

注释

关于XML配置中元素的具体信息可以参考http://activemq.apache.org/xbean-xml-reference-50.html下面介绍本篇配置使用的一些重要元素。

DispathPolicy

ActiveMQ支持3中不同的分发策略（避免翻译了以后误解，这里用原文）：

1.      <roundRobinDispatchPolicy>：Simple dispatch policy that sends a message to every subscriptionthat matches the message.

2.      <simpleDispatchPolicy>：Simple dispatch policy that sends a message to every subscriptionthat matches the message.

3.      <strictOrderDispatchPolicy>：Dispatch policy that causes every subscription to see messages inthe same order.

SubscriptionRecoveryPolicy

ActiveMQ支持6种恢复策略，可以自行选择使用不同的策略

1.      <fixedCountSubscriptionRecoveryPolicy>： keep a fixed count of last messages.

2.      <fixedSizedSubscriptionRecoveryPolicy>：keep a fixed amount of memory available in RAM for messagehistory which is evicted in time order.

3.      <lastImageSubscriptionRecoveryPolicy>：only keep the last message.

4.      <noSubscriptionRecoveryPolicy>：disable recovery of messages.

5.      <queryBasedSubscriptionRecoveryPolicy>：perform a user specific query mechanism to load any messages theymay have missed.

6.      <timedSubscriptionRecoveryPolicy>：keep a timed buffer of messages around in memory and use that torecover new subscriptions.

PersistenceAdapter

http://activemq.apache.org/persistence讲解了关于persistence的信息。ActiveMQ5.0使用AMQ Message Store 持久化消息，这种方式提供了很好的性能（The AMQ Message Store is an embeddable transactional message storagesolution that is extremely fast and reliable.） 默认使用该存储方式即可，如果想使用JDBC来存储，可以查找文档配置。

Summary

本篇文章只提供了基本配置信息。如果需要更多的文章，可以查看ActiveMQ的文档。

讲了安装和简单的配置，下一篇将介绍和Sping的整合，以及多个queue，多个topic，多个producer，多个consumer的配置，使用。

通过一个实例介绍使用spring发送,消费topic,queue类型消息的方法. 不懂topic和queue的google 之.

如图示, TOPIC和QUEUE分别代表一个topic和一个queue消息通道.

1.      TopicMessageProducer向topic发送消息, TopicConsumerA和TopicConsumerB则从topic消费消息.

2.      QueueMessageProducer向Queue发送消息, QueueConsumer从Queue中消费消息

Spring整合JMS

就像对orm, web的支持一样, spring同样支持jms, 为整合jms到已有的项目提供了很多便利的方法. 本篇主要讲实战, 是所以先从配置开始,spring配置jms基本上需要8个部分.

1.      ConnectionFactory. 和jms服务器的连接, 可以是外部的jms server, 也可以使用embeddedActiveMQ Broker.

2.      Destination. 有topic和queue两种方式.

3.      JmsTemplate. spring提供的jms模板.

4.      MessageConverter. 消息转换器.

5.      MessageProducer. 消息生产者.

6.      MessageConsumer. 消息消费者.

7.      MessageListener. 消息监听器

8.      MessageListenerContainer. 消息监听容器

下面以实例的方式介绍上面8个部分.

1. ConnectionFactory

Xml代码

1.      <amq:connectionFactory id="jmsConnectionFactory" brokerURL="vm://localhost" />

brokerURL是指要连接的activeMQ server的地址,activeMQ提供了多种brokerURL, 集体可参见文档.一般我们使用嵌套的ActiveMQ server. 配置如下, 这个配置使用消息的存储机制, 服务器重启也不会丢失消息.

Xml代码

1.      <!--  embedded ActiveMQ Broker -->

2.      <amq:broker useJmx="false" persistent="true">

3.      <amq:persistenceAdapter>

4.      <amq:amqPersistenceAdapter directory="d:/amq"/>

5.      </amq:persistenceAdapter>

6.      <amq:transportConnectors>

7.      <amq:transportConnector uri="tcp://localhost:61616" />

8.      <amq:transportConnector uri="vm://localhost:0" />

9.      </amq:transportConnectors>

10.  </amq:broker>

2. Destination

在实例中我们使用了两种destination

Xml代码

1.      <!--  ActiveMQ destinations  -->

2.      <!--使用topic方式-->

3.      <amq:topic name="TOPIC" physicalName="JMS-TEST-TOPIC" />

4.      <!--使用Queue方式-->

5.      <amq:queue name="QUEUE" physicalName="JMS-TEST-QUEUE" />

3. JmsTemplate

Xml代码

1.      <!--  Spring JmsTemplate config -->

2.      <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

3.      <property name="connectionFactory">

4.      <!--  lets wrap in a pool to avoid creating a connection per send -->

5.      <bean class="org.springframework.jms.connection.SingleConnectionFactory">

6.      <property name="targetConnectionFactory" ref="jmsConnectionFactory" />

7.      </bean>

8.      </property>

9.      <!-- custom MessageConverter -->

10.  <property name="messageConverter" ref="defaultMessageConverter" />

11.  </bean>

4. MessageConverter

MessageConverter实现的是org.springframework.jms.support.converter.MessageConverter接口, 提供消息的转换功能. DefaultMessageConverter的实现见附件.

Xml代码

1.      <bean id="defaultMessageConverter" class="com.andyao.activemq.DefaultMessageConverter" />

5. MessageProducer

实例拥有两个消息生产者, 消息生产者都是POJO, 实现见附件.

Xml代码

1.      <!-- POJO which send Message uses  Spring JmsTemplate -->

2.      <bean id="topicMessageProducer" class="com.andyao.activemq.TopicMessageProducer">

3.      <property name="template" ref="jmsTemplate" />

4.      <property name="destination" ref="TOPIC" />

5.      </bean>

6.      <bean id="queueMessageProducer" class="com.andyao.activemq.QueuMessageProducer">

7.      <property name="template" ref="jmsTemplate" />

8.      <property name="destination" ref="QUEUE" />

9.      </bean>

6. MessageConsumer

TOPIC通道有两个消息消费者, QUEUE有一个消息消费者

Xml代码

1.      <!--  Message Driven POJO (MDP) -->

2.      <!-- consumer1 for topic a -->

3.      <bean id="topicConsumerA" class="com.andyao.activemq.TopicConsumerA" />

4.      <!-- consumer2 for topic a -->

5.      <bean id="topicConsumerB" class="com.andyao.activemq.TopicConsumerB" />

6.      <!-- consumer for queue -->

7.      <bean id="queueConsumer" class="com.andyao.activemq.QueueConsumer" />

7. MessageListener

每一个消息消费者都对应一个MessageListener

Xml代码

1.      <bean id="topicListenerA" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

2.      <constructor-arg ref="topicConsumerA" />

3.      <!--  may be other method -->

4.      <property name="defaultListenerMethod" value="receive" />

5.      <!-- custom MessageConverter define -->

6.      <property name="messageConverter" ref="defaultMessageConverter" />

7.      </bean>

8.

9.      <bean id="topicListenerB" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

10.  <constructor-arg ref="topicConsumerB" />

11.  <!--  may be other method -->

12.  <property name="defaultListenerMethod" value="receive" />

13.  <!-- custom MessageConverter define -->

14.  <property name="messageConverter" ref="defaultMessageConverter" />

15.  </bean>

16.

17.  <bean id="queueListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

18.  <constructor-arg ref="queueConsumer" />

19.  <!--  may be other method -->

20.  <property name="defaultListenerMethod" value="receive" />

21.  <!-- custom MessageConverter define -->

22.  <property name="messageConverter" ref="defaultMessageConverter" />

23.  </bean>

8. MessageListenerContainer

有几个MessageListener既有几个MessageListenerContainer

Xml代码

1.      <bean id="topicListenerContainerA" class="org.springframework.jms.listener.DefaultMessageListenerContainer">

2.      <property name="connectionFactory" ref="jmsConnectionFactory" />

3.      <property name="destination" ref="TOPIC" />

4.      <property name="messageListener" ref="topicListenerA" />

5.      </bean>

6.

7.      <bean id="topicListenerContainerB" class="org.springframework.jms.listener.DefaultMessageListenerContainer">

8.      <property name="connectionFactory" ref="jmsConnectionFactory" />

9.      <property name="destination" ref="TOPIC" />

10.  <property name="messageListener" ref="topicListenerB" />

11.  </bean>

12.

13.  <bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">

14.  <property name="connectionFactory" ref="jmsConnectionFactory" />

15.  <property name="destination" ref="QUEUE" />

16.  <property name="messageListener" ref="queueListener" />

17.  </bean>

Summary

写spring配置文件的时候, 要把MessageProducer,MessageConsumer,MessageListener,MessageListenerContainer几个地方弄清楚:

1.      可以有一个或者多个消息生产者向同一个destination发送消息.

2.      queue类型的只能有一个消息消费者.

3.      topic类型的可以有多个消息消费者.

4.      每个消费者对应一个MessageListener和一个MessageListenerContainer.

activemq上一次已经讲解了安装、启动、数据库的持久化配置等。

这次主要记录下，如何跟spring结合使用，如何发送消息以及进行消费。

消息产生者向JMS发送消息的步骤

(1)创建连接使用的工厂类JMSConnectionFactory
(2)使用管理对象JMS ConnectionFactory建立连接Connection
(3)使用连接Connection 建立会话Session
(4)使用会话Session和管理对象Destination创建消息生产者MessageSender
(5)使用消息生产者MessageSender发送消息

消息消费者从JMS接受消息的步骤

(1)创建连接使用的工厂类JMS ConnectionFactory
(2)使用管理对象JMS ConnectionFactory建立连接Connection
(3)使用连接Connection 建立会话Session
(4)使用会话Session和管理对象Destination创建消息消费者MessageReceiver
(5)使用消息消费者MessageReceiver接受消息，需要用setMessageListener将MessageListener接口绑定到MessageReceiver

1.下面是使用了JTA配置、以及mq进行数据库持久化的spring相关配置。

Xml代码

1.  <p> <!--建立mqfactory<strong>-->

2.  </strong><bean id="xaMQFactory"

3.  class="org.apache.activemq.ActiveMQXAConnectionFactory">

4.  <property name="brokerURL" value="tcp://localhost:61616" />

5.  </bean>

6.  <!--建立connectionfactory，<strong>用jta的事务进行管理-->

7.  </strong><bean id="connectionFactory"

8.  class="com.atomikos.jms.AtomikosConnectionFactoryBean"

9.  init-method="init" destroy-method="close">

10. <property name="uniqueResourceName">

11. <value>QUEUE_BROKER</value>

12. </property>

13. <property name="xaConnectionFactory">

14. <ref bean="xaMQFactory"/>

15. </property>

16. </bean>

17. <!--建立发送队列，以queuem模式-->

18. <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

19. <constructor-arg index="0" value="solrMessageQueue"/>

20. </bean>

21. <!--jdbcTemplate 用于发送消息，由于是使用数据库持久化-->

22.

23. <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">

24. <property name="dataSource">

25. <ref bean="dataSource"/>

26. </property>

27. </bean>

28.

29. <!--数据库连接信息,使用了Atomikos的JTA事务机制-->

30. <bean id="dataSource" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close">

31. <property name="uniqueResourceName">

32. <value>ZZGRID_MAIN_JDBC_RESOURCE</value>

33. </property>

34. <property name="xaDataSourceClassName" value="oracle.jdbc.xa.client.OracleXADataSource" />

35. <property name="xaProperties">

36. <props>

37. <prop key="user">username</prop>

38. <prop key="password">password</prop>

39. <prop key="URL" >jdbc:oracle:thin:@localhost:1521:oracle</prop>

40. </props>

41. </property>

42. </bean></p>

2.不使用JTA，只进行数据库持久化的Spring配置

Xml代码

1.  <!--建立connectionfactory-->

2.  <span style="color: #000000;"> <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">

3.  <property name="brokerURL" value="tcp://localhost:61616"/>

4.  </bean>

5.  </span>

6.  <!--建立发送队列，以queue模式-->

7.  <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

8.  <constructor-arg index="0" value="solrMessageQueue"/>

9.  </bean>

10. <!--jdbcTemplate 用于发送消息，由于是使用数据库持久化-->

11. <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">

12. <property name="dataSource">

13. <ref bean="dataSource"/>

14. </property>

15. </bean>

16. <!--数据库连接信息-->

17. <bean id="dataSource"

18. class="com.mchange.v2.c3p0.ComboPooledDataSource" destroy-method="close">

19. <property name="driverClass"

20. value="oracle.jdbc.driver.OracleDriver" />

21. <property name="jdbcUrl" value="jdbc:oracle:thin:@localhost:1521:oracle"/>

22. <property name="user" value="username" />

23. <property name="password" value="password" />

24. <property name="acquireIncrement" value="5" />

25. <property name="acquireRetryAttempts" value="30" />

26. <property name="acquireRetryDelay" value="1000" />

27. <property name="idleConnectionTestPeriod" value="60" />

28. <property name="testConnectionOnCheckin" value="true" />

29. <property name="automaticTestTable" value="C3P0Table" />

30. </bean>

3.使用MQ默认的硬盘读写，结合spring的配置

Xml代码

1.  <!--建立connectionfactory-->

2.  <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">                 <property name="brokerURL" value="tcp://localhost:61616"/>   </bean> <!--建立发送队列，以queue模式--><bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

3.  <constructor-arg index="0" value="solrMessageQueue"/>

4.  </bean>

5.

6.  <!--建立发送队列，以queue模式-->

7.  <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

8.  <constructor-arg index="0" value="solrMessageQueue"/>

9.  </bean>

10.

11.

12. <!--jdbcTemplate 用于发送消息-->

13. <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">

14. <span style="color: #000000;"> <property name="connectionFactory" ref="connectionFactory"/></span>

15. </bean>

MQ消息体（body）――JMS中定义了5种消息体：ByteMessage、MapMessage、ObjectMessage、StreamMessage和TextMessage。这里演示相对最复杂的 ObjectMessage。

消息发送主体

Java代码

1.  package xxxx;

2.  import javax.jms.ConnectionFactory;

3.  import javax.jms.Destination;

4.

5.  import org.springframework.beans.BeansException;

6.  import org.springframework.context.ApplicationContext;

7.  import org.springframework.context.ApplicationContextAware;

8.  import org.springframework.jms.core.JmsTemplate;

9.  import org.springframework.stereotype.Component;

10.

11. import com.xxx.jms.SolrMessage;

12. import com.xxx.jms.SolrMessageConvert;

13. @Component("activeMQMessageSender")

14. public class ActiveMQMessageSender implements ApplicationContextAware{

15.

16. private ApplicationContext appContext;

17.

18. @Override

19. public void setApplicationContext(ApplicationContext appContext)

20. throws BeansException {

21. this.appContext=appContext;

22. }

23.

24. public void send(SolrMessage solrMessage) {

25. String isSenderSolrMsg = "true";

26. if (isSenderSolrMsg != null && Boolean.valueOf(isSenderSolrMsg)){

27. JmsTemplate template = new JmsTemplate();

28. template.setConnectionFactory((ConnectionFactory) appContext.getBean("connectionFactory"));

29. template.setDefaultDestination((Destination) appContext.getBean("queueDestination"));

30. template.setSessionTransacted(true);//如果不使用事务，这里修改成false

31. template.setMessageConverter(new SolrMessageConvert());

32. template.convertAndSend(solrMessage);

33. }

34. }}

消息对象

Java代码

1.  package com.xxx.jms;

2.

3.  public class SolrMessage {

4.  private Long id;

5.  private String type;

6.  private String mode;

7.  private String idCardNo;

8.

9.  public Long getId() {

10. return id;

11. }

12.

13. public void setId(Long id) {

14. this.id = id;

15. }

16.

17. public String getType() {

18. return type;

19. }

20.

21. public void setType(String type) {

22. this.type = type;

23. }

24.

25. public String getMode() {

26. return mode;

27. }

28.

29. public void setMode(String mode) {

30. this.mode = mode;

31. }

32.

33. public String getIdCardNo() {

34. return idCardNo;

35. }

36.

37. public void setIdCardNo(String idCardNo) {

38. this.idCardNo = idCardNo;

39. }

40.

41. public SolrMessage(String idCardNo, String type) {

42. this.idCardNo = idCardNo;

43. this.mode = Mode.DELETE.toString();

44. this.type = type;

45. }

46.

47. public SolrMessage(Long id, String type, String mode) {

48. this.id = id;

49. this.type = type;

50. this.mode = mode;

51. }

52.

53. public SolrMessage() {

54.

55. }}

msg消息和send消息的转化类

Java代码

1.  package com.xxx.jms;

2.

3.  import javax.jms.JMSException;

4.  import javax.jms.MapMessage;

5.  import javax.jms.Message;

6.  import javax.jms.Session;

7.

8.  import org.springframework.jms.support.converter.MessageConversionException;

9.  import org.springframework.jms.support.converter.MessageConverter;

10.

11. public class SolrMessageConvert implements MessageConverter {

12.

13. @Override

14. public Object fromMessage(Message message) throws JMSException,

15. MessageConversionException {

16. SolrMessage solrMessage = new SolrMessage();

17. if (message != null) {

18. MapMessage msg = (MapMessage) message;

19. solrMessage.setId(msg.getLong("id"));

20. solrMessage.setIdCardNo(msg.getString("idCardNo"));

21. solrMessage.setMode(msg.getString("mode"));

22. solrMessage.setType(msg.getString("type"));

23. }

24.

25. return solrMessage;

26. }

27.

28. @Override

29. public Message toMessage(Object object, Session session)

30. throws JMSException, MessageConversionException {

31. MapMessage mapMessage = session.createMapMessage();

32. if (null != object) {

33. SolrMessage solrMessage = (SolrMessage) object;

34.

35. if (null != solrMessage.getId()) {

36. mapMessage.setLong("id", solrMessage.getId());

37. }

38. if (null != solrMessage.getType()) {

39. mapMessage.setString("type", solrMessage.getType());

40. }

41. if (null != solrMessage.getMode()) {

42. mapMessage.setString("mode", solrMessage.getMode());

43. }

44. if (null != solrMessage.getIdCardNo()) {

45. mapMessage.setString("idCardNo", solrMessage.getIdCardNo());

46. }

47. }

48.

49. return mapMessage;

50. }}

mq消息的消费也是类似的，主要涉及的spring配置如下

Xml代码

1.  <!--消息转化类，用于send消息和mqmsg消息的转化-->

2.  t;bean id="messageConvert" class="com.xxx.SolrMessageConvert" />

3.  <!--消息的消费-->

4.  t;bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

5.  <constructor-arg>

6.  <bean class="com.xxx.SolrRegisterListener"/>

7.  </constructor-arg>

8.  <property name="defaultListenerMethod" value="onRegister"/>

9.  <property name="messageConverter" ref="messageConvert"/>

10. </bean>

11. <!--接收mq消息的监听器,用于检测mq消息的消费-->

12. <bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">

13. <property name="connectionFactory" ref="connectionFactory"/>

14. <property name="destination" ref="queueDestination"/>

15. <property name="messageListener" ref="messageListener"/>

16. </bean>

消息消费类（接收类）

Java代码

1.  package com.xxx;

2.

3.  import org.springframework.beans.factory.annotation.Autowired;

4.  import com.xxx.solr.domain.SolrMessage;

5.

6.

7.  public class SolrRegisterListener {

8.

9.  public void onRegister(SolrMessage solrMessage){

10. System.out.println(solrMessage.getId() + "  "

11. + solrMessage.getMode() + " "

12. + solrMessage.getType() + " "

13. + solrMessage.getIdCardNo());

14. /**具体的业务逻辑

15. **对接收到的消息进行消费

16. **/

17. }}

什么是JMS？

JMS即java消息服务，JMS通过消息的形式来降低组件之间的耦合度。

JMS由两部分组成消息发送者和消息监听者。

JMS的运用场景？

用户系统负责维护用户信息，文档系统负责维护文档信息，但是当用户删除的时候，需要将他所撰写的文档信息也删除的时候，在用户管理模块调用文档管理模块的接口，会造成用户模块和业务模块紧耦合。

这个时候可以使用JMS技术来将紧耦合转化为松耦合，具体做法是用户系统在删除，修改用户的时候往JMS服务器发送更新消息，又业务系统监听这些消息，然后按照自己的业务逻辑来进行相应的处理。

即组件A做了一件事情往消息服务器上发送了一个通知，组件B监听到了消息，处理自己的业务逻辑。

详细步骤：

1：配置消息服务器：配置JMS需要两个对象connectionFactory和 destination。

connectionFactory使用jboss自带的TopicConnectionFactory。

destination可以使用自定义的。

kiral-jms-service.xml   注意：文件名称一定要是-service.xml结尾。这个文件放在部署目录下。

xml 代码

1.  <!---->< xml   version = "1.0"  encoding = "UTF-8" ?>

2.  < server >

3.  < mbean   code ="org.jboss.mq.server.jmx.Topic"

4.  name ="jboss.mq.destination:service=Topic,name=kiralJms" >

5.  < depends  optional-attribute-name = "DestinationManager" > jboss.mq:service= DestinationManager depends >

6.  < depends  optional-attribute-name = "SecurityManager" > jboss.mq:service =SecurityManager depends >

7.  < attribute   name ="SecurityConf" >

8.  < security>

9.  <role   name = "guest"   read = "true"  write = "true" />

10. <role   name = "publisher"   read = "true"  write = "true"   create = "false" />

11. <role   name = "durpublisher"   read = "true"  write = "true"   create = "true" />

12. security >

13. attribute >

14. mbean >

15. server >

2：配置发送消息端

bean-jms.xml

xml 代码

1.  <!---->xml   version = "1.0"  encoding = "GB2312" ?>

2.  < beans >

3.  < bean   id ="jmsConnectionFactory"

4.  class= "org.springframework.jndi.JndiObjectFactoryBean" >

5.  <property   name = "jndiName" >

6.  < value > TopicConnectionFactory value >

7.  property >

8.  bean >

9.

10. < bean   id ="destination"

11. class= "org.springframework.jndi.JndiObjectFactoryBean" >

12. <property   name = "jndiName" >

13. < value > topic/kiralJms value >

14. property >

15. bean >

16.

17. <!---->

18. < bean   id ="jmsTemplate"

19. class= "org.springframework.jms.core.JmsTemplate" >

20. <property   name = "connectionFactory" >

21. < bean

22. class = "org.springframework.jms.connection.SingleConnectionFactory">

23. < property   name = "targetConnectionFactory"

24. ref = "jmsConnectionFactory"   />

25. bean >

26. property >

27. bean >

28.

29. <!----> <!---->

30. < bean   id ="messageProducer"

31. class= "jms.MessageProducer" >

32. <property   name = "template"   ref ="jmsTemplate"   />

33. <property   name = "destination"   ref ="destination"   />

34. bean >

35. beans >

java 代码

1.  import  javax.jms.Destination;

2.  import  javax.jms.JMSException;

3.  import  javax.jms.Message;

4.  import  javax.jms.Session;

5.

6.  import  org.springframework.jms.core.JmsTemplate;

7.  import org.springframework.jms.core.MessageCreator;

8.

9.  /***********************************************************

10. * 消息发送者

11. *

12. * @作者：kiral

13. * @日期：2007-7-3

14. **********************************************************/

15. public   class  MessageProducer {

16.

17. public   void send( final  String message) {

18. template.send(destination, new  MessageCreator() {

19. public  Message createMessage(Session session)  throws JMSException {

20. Message m = session.createTextMessage(message);

21. return  m;

22. }

23. });

24. }

25.

26. private JmsTemplate template;

27.

28. private Destination destination;

29.

30. public   void setDestination(Destination destination) {

31. this.destination = destination;

32. }

33.

34. public   void setTemplate(JmsTemplate template) {

35. this.template = template;

36. }

37.

38. }

发送方调用send方法发送消息。

3：配置消息接收者

xml 代码

1.  <!---->< xml   version = "1.0"  encoding = "UTF-8" ?>

2.  <!---->

3.  < beans >

4.  < bean   id ="jmsConnectionFactory"

5.  class= "org.springframework.jndi.JndiObjectFactoryBean" >

6.  <property   name = "jndiName" >

7.  < value > TopicConnectionFactory value >

8.  property >

9.  < bean >

10. < bean   id ="destination"

11. class= "org.springframework.jndi.JndiObjectFactoryBean" >

12. <property   name = "jndiName" >

13. < value > topic/kiralJms value >

14. property >

15. < bean >

16.

17. <!---->

18. < bean   id ="messageListener"

19. class= "jms.MessageConsumer" >

20. <property   name = "worksheetService"   ref ="worksheetService" > property >

21. < bean >

22.

23. <!---->

24. < bean   id ="listenerContainer"

25. class= "org.springframework.jms.listener.DefaultMessageListenerContainer">

26. <property   name = "connectionFactory"   ref ="jmsConnectionFactory"   />

27. <property   name = "destination"   ref ="destination"   />

28. <property   name = "messageListener"   ref ="messageListener"   />

29. < bean >

30. < beans >

java 代码

1.  import  javax.jms.Message;

2.  import  javax.jms.MessageListener;

3.

4.  import org.kiral.flow.service.WorksheetService;

5.

6.  /*******************************************************************************

7.  * 消息接收者

8.  *

9.  * @作者：kiral

10. * @日期：2007-7-3

11. ******************************************************************************/

12. public   class  MessageConsumer implements  MessageListener {

13.

14. private WorksheetService worksheetService;

15.

16. public  WorksheetService getWorksheetService() {

17. return  worksheetService;

18. }

19.

20. public   void setWorksheetService(WorksheetService worksheetService) {

21. this.worksheetService = worksheetService;

22. }

23.

24. public   void onMessage(Message message) {

25. System.out.println(message);

26. worksheetService.updateRole();

27. }

28.

29. }

1.1 JMS与ActiveMQ特性

JMS始终在JavaEE五花八门的协议里，WebService满天飞的时候占一位置，是因为：

·        它可以把不影响用户执行结果又比较耗时的任务（比如发邮件通知管理员）异步的扔给JMS 服务端去做，而尽快的把屏幕返还给用户。

·        服务端能够多线程排队响应高并发的请求。

·        可以在Java世界里达到最高的解耦。客户端与服务端无需直连，甚至无需知晓对方是谁、在哪里、有多少人，只要对流过的信息作响应就行了，在企业应用环境复杂时作用明显。

ActiveMQ的特性：

·        完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，也是Apache Geronimo默认的JMS provider。

·        POJO withdout EJB Container，不需要实现EJB繁琐复杂的Message Bean接口和配置。

·        Spring Base，可以使用Spring的各种特性如IOC、AOP。

·        Effective，基于Jencks的JCA Container实现pool connection，controltransactions and manage security。

1.2SpringSide 的完全POJO的JMS方案

SpringSide2.0在BookStore示例中，演示了用户下订单时，将发通知信到用户邮箱的动作，通过JMS交给JMS服务端异步完成，避免了邮件服务器的堵塞而影响用户的下订。

全部代码于examples\bookstore\src\java\org\springside\bookstore\components\activemq目录中。

一个JMS场景通常需要三者参与：

·        一个POJO的的Message Producer，负责使用Spring的JMS Template发送消息。

·        一个MessageConverter，负责把Java对象如订单(Order)转化为消息，使得Producer能够直接发送POJO。

·        一个MDP MessageConsumer，负责接收并处理消息。

SpringSide 2.0采用了ActiveMQ4.1-incubator-SNAPSHOT 与Spring 2.0 集成，对比SS1.0M3，有三个值得留意的地方，使得代码中几乎不见一丝JMS的侵入代码：

1.      采用Spring2.0的Schema式简化配置。

2.      实现MessageConverter转化消息与对象，使得Producer能够直接发送POJO而不是JMS Message。

3.      使用了Spring的DefaultMessageListenerContainer与MessageListenerAdapter，消息接收者不用实现MessageListener接口。

2.引入ActiveMQ的XSD

ActiveMQ4.1 响应Spring 2.0号召，支持了引入XML Schemanamespace的简单配置语法，简化了配置的语句。

在ApplicationContext.xml(Spring的配置文件)中引入ActiveMQ的XML Scheam 配置文件),如下:

<beans

xmlns="http://www.springframework.org/schema/beans"

xmlns:amq="http://activemq.org/config/1.0"

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

http://activemq.org/config/1.0http://people.apache.org/repository/org.apache.activemq/xsds/activemq-core-4.1-incubator-SNAPSHOT.xsd">

由于ActiveMQ4.1 SnapShot的那个XSD有部分错误，因此使用的是自行修改过的XSD。

先在ClassPath根目录放一个修改过的activemq-core-4.1-incubator-SNAPSHOT.xsd。

在ClassPath 下面建立META-INF\spring.schemas 内容如下。这个spring.schemas是spring自定义scheam的配置文件,请注意"http:\://"部分写法

http\://people.apache.org/repository/org.apache.activemq/xsds/activemq-core-4.1-incubator-SNAPSHOT.xsd=/activemq-core-4.1-incubator-SNAPSHOT.xsd

3. 配置方案

3.1 基础零件

1. 配置ActiveMQ Broker

暂时采用在JVM中嵌入这种最简单的模式，  当spring初始化时候，ActiveMQ embedded Broker 就会启动了。

<!--  lets create an embedded ActiveMQ Broker -->

<amq:broker useJmx="false" persistent="false">

<amq:transportConnectors>

　　　　          <amq:transportConnector uri="tcp://localhost:0"/>

　       </amq:transportConnectors>

　</amq:broker>

2. 配置(A)ConnectionFactory

由于前面配置的Broker是JVM embedded 所以URL为:vm://localhost

<!--  ActiveMQ connectionFactory to use  -->

<amq:connectionFactory id="jmsConnectionFactory"brokerURL="vm://localhost"/>

3 配置(B)Queue

<!--  ActiveMQ destinations to use  -->

<amq:queue name="destination"physicalName="org.apache.activemq.spring.Test.spring.embedded"/>

4. 配置(C)Converter

   配置Conveter，使得Producer能够直接发送Order对象，而不是JMS的Message对象。

<!--  OrderMessage converter  -->

<bean id="orderMessageConverter"class="org.springside.bookstore.components.activemq.OrderMessageConverter"/>

3.2  发送端

1 配置JmsTemplate

Spring提供的Template，绑定了(A)ConnectionFactory与(C)Converter。

<!--  Spring JmsTemplate config -->

<bean id="jmsTemplate"class="org.springframework.jms.core.JmsTemplate">

<property name="connectionFactory">

<!--  lets wrap in a pool to avoid creating aconnection per send -->

<bean class="org.springframework.jms.connection.SingleConnectionFactory">

<property name="targetConnectionFactory"ref="jmsConnectionFactory"/>

</bean>

</property>

<!-- custom MessageConverter -->

<property name="messageConverter"ref="orderMessageConverter"/>

</bean>

2.Producer

消息发送者，使用JmsTemplate发送消息，绑定了JmsTemplate(含A、C)与(B)Queue。

<!-- POJO which send Message uses  Spring JmsTemplate，绑定JMSTemplate与Queue -->

<bean id="orderMessageProducer"class="org.springside.bookstore.components.activemq.OrderMessageProducer">

<property name="template"ref="jmsTemplate"/>

<property name="destination"ref="destination"/>

</bean>

3.3 接收端

  1.接收处理者(MDP)

使用Spring的MessageListenerAdapter，指定负责处理消息的POJO及其方法名，绑定(C)Converter。

<!-- Message Driven POJO (MDP)，绑定Converter-->

<beanid="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">

<constructor-arg>

<beanclass="org.springside.bookstore.components.activemq.OrderMessageConsumer">

<propertyname="mailService" ref="mailService"/>

</bean>

</constructor-arg>

<!-- may be other method -->

<propertyname="defaultListenerMethod" value="sendEmail"/>

<!--custom MessageConverter define -->

<propertyname="messageConverter" ref="orderMessageConverter"/>

</bean>

2. listenerContainer

负责调度MDP， 绑定(A) connectionFactory, (B)Queue和MDP。

<!--  this is the attendant message listener container-->

<beanid="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">

<propertyname="connectionFactory" ref="jmsConnectionFactory"/>

<propertyname="destination" ref="destination"/>

<propertyname="messageListener" ref="messageListener"/>

</bean>

什么是消息
消息是一个用于在组件和应用程序之间通讯的的方法。消息之间的传递是点对点的。任何终端之间都可以相互接受和发送消息。并且每个终端都必须遵守如下的规则
 -> 创建消息 -> 发送消息 -> 接收消息 -> 读取消息
为什么要使用消息
理由很简单，消息是一个分布式的低耦合通讯方案。A发送一个消息到一个agent ,B作为接受者去agent上获取消息。但是A,B不需要同时到agent上去注册。agent作为一个中转为A,B提供搞效率的通讯服务。
开发者的关注点
走到这里，我也不想去解释jms spec上那些抽象且复杂的概念了，说的很白，1年多了我自己也没弄懂是个什么东西，也没时间从头到尾去仔细的看，同时我认为没必要，我所关注的是如何让jms跑起来，并且工作正常，所以spec只是个字典，当我需要用的时候才去查。
开发者的jms环境
遵守简单明了的原则，所谓jms环境只是2个对象
1> ConnectionFactory
2> Destination 通常Provider会提供JNDI的对象获取，具体方法可以去Privider的网站上搜索jndi support

下面我以jbossMq为介质跑一个简单的jms，为了保证jms的本质清晰，我没有使用jbossMq的Api，而是直接调用的jms Api.

java 代码

1.  package com.javaeye.jms.jboss;

2.

3.  import javax.jms.Connection;

4.  import javax.jms.ConnectionFactory;

5.  import javax.jms.Destination;

6.  import javax.jms.JMSException;

7.  import javax.jms.MessageConsumer;

8.  import javax.jms.MessageProducer;

9.  import javax.jms.Queue;

10. import javax.jms.QueueSender;

11. import javax.jms.Session;

12. import javax.jms.TextMessage;

13. import javax.naming.Context;

14. import javax.naming.InitialContext;

15. import javax.naming.NamingException;

16.

17. public class JbossNativeJmsImpl {

18.

19. /**

20. * @author zuly

21. *

22. * following jms ptp domain, use an simple text message to test

23. *

24. * A jms ptp sender will following the steps below!

25. *     1> get an ConnectionFactory use JNDI Lookup Or Initial it yourself

26. *     2> use this ConnectionFactory to start a jms connection

27. *       [spec to jms 1.1 apito get the main idea of it ]

28. *     3> use connection to create a jms session

29. *     4> get a queue destination / messege agent

30. *     5> start the Producer[jms1.1 spec] by a session

31. *     6> get messege Object or initial it yourself by implements the messegeor

32. *       it's sub interfaces

33. *     7> call sender or send it selfing

34. *     8> finallized the connection object or it will throw a warning to you!

35. *

36. * @param messege

37. * @throws NamingException

38. * @throws JMSException

39. */

40. public void sendingProcessing(String messege) throws NamingException, JMSException{

41. Context ctx = new InitialContext();

42. ConnectionFactory cf = (ConnectionFactory) ctx.lookup("java:JmsXA");

43. Connection conn = cf.createConnection();

44. Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

45. Destination dest = (Queue) ctx.lookup("queue/A");

46. MessageProducer msgp = session.createProducer(dest);

47. QueueSender sender = (QueueSender) msgp;

48. TextMessage msg = session.createTextMessage();

49. msg.setText(messege);

50. sender.send(msg);

51. conn.close();

52. }

53.

54.

55.

56. /**

57. * @author zuly

58. *

59. * following jms ptp domain, use an simple text message to test

60. *

61. * A jms ptp retriver will following the steps below!

62. *     1> get an ConnectionFactory use JNDI Lookup Or Initial it yourself

63. *     2> use this ConnectionFactory to start a jms connection

64. *       [spec to jms 1.1 apito get the main idea of it ]

65. *     3> use connection to create a jms session

66. *     4> get a queue destination / messege agent

67. *     5> retrive a consumer from session

68. *     6> start the jms connection to retrivte the message

69. *     7> get message from consumer

70. *

71. * @return textMessege

72. * @throws NamingException

73. * @throws JMSException

74. */

75. public String retriveingProcessing() throws NamingException, JMSException{

76. Context ctx = new InitialContext();

77. ConnectionFactory cf = (ConnectionFactory) ctx.lookup("java:JmsXA");

78. Connection conn = cf.createConnection();

79. Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

80. Destination dest = (Queue) ctx.lookup("queue/A");

81. MessageConsumer msgconsumer = session.createConsumer(dest);

82. //MessageListener ml = new JmsListenner();

83. //msgconsumer.setMessageListener(ml);

84. conn.start();

85. TextMessage msg = (TextMessage) msgconsumer.receive();

86. conn.close();

87. System.out.println("messege is" + msg.getText());

88. return msg.getText();

89. }

90. }

<pointto="" point=""></point> <point to=""point=""></point> 注意retrive函数中comment的掉的两行，消息Listener的作用是实现异步通讯，但是它有一个约定，必须和发送者
保持物理上的分离，针对于jboss而言，就要求这个Listener必须跑在容器外面。这是一个很搞的问题，每天Jms的邮件列表里面都有无数的这样的 问题发过来。但是回复的人很少。我自己也从来不回复。 其实我也不清楚写这篇文章到底是出于什么目的，怕只是让这么一个简单的问题有一个回答而已。
把下面这个程序跑起来就可以异步接受消息了。

java 代码

1.  package com.javaeye.jms.jboss;

2.

3.  import java.util.Properties;

4.

5.  import javax.jms.Connection;

6.  import javax.jms.ConnectionFactory;

7.  import javax.jms.Destination;

8.  import javax.jms.JMSException;

9.  import javax.jms.MessageConsumer;

10. import javax.jms.MessageListener;

11. import javax.jms.Session;

12. import javax.naming.Context;

13. import javax.naming.InitialContext;

14. import javax.naming.NamingException;

15.

16. import com.javaeye.spring.services.jms.mdp.JmsListenner;

17.

18. public class JbossJmsAsync {

19.

20. /**

21. * @param args

22. * @throws NamingException

23. * @throws JMSException

24. */

25. public static void main(String[] args) throws NamingException, JMSException {

26. Properties pops = new Properties();

27. pops.setProperty("jboss.bind.address", "0.0.0.0");

28. pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");

29. pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");

30. pops.setProperty("java.naming.provider.url", "localhost");

31. Context ctx = new InitialContext(pops);

32. ConnectionFactory cf = (ConnectionFactory) ctx.lookup("ConnectionFactory");

33. Connection conn = cf.createConnection();

34. Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

35. Destination dest = (Destination) ctx.lookup("queue/A");

36. MessageConsumer msgConsumer = session.createConsumer(dest);

37. MessageListener ml = new JmsListenner();

38. msgConsumer.setMessageListener(ml);

39. conn.start();

40. }

41.

42. }

javaeye的主题好像是spring，为了迎合领导，下面我把这套东西跑在spring里面。同时我发现spring对jms的包装真的简单，而且还提供了一个模版，虽然这个模版的接口是在是很罗唆。
ps:今天是第1次用spring在reference里找了半天找不到方法注入的办法，于是google了一个注入办法，不合理的地方请大家指出。首先我通过方法来注入ConnectionFactory和Destination这两个对象来支撑jms环境

java 代码

1.  package com.javaeye.spring.services.jms.mdp;

2.

3.  import java.util.Properties;

4.

5.  import javax.jms.ConnectionFactory;

6.  import javax.jms.Destination;

7.  import javax.jms.Queue;

8.  import javax.naming.Context;

9.  import javax.naming.InitialContext;

10. import javax.naming.NamingException;

11.

12. public class UserJmsTransactionUtil {

13.

14. private String connectionFactoryJndiLookUp;

15.

16. private String destinationJndiLookUp;

17.

18. private String localConnectionFactoryJndiLookUp;

19.

20. private String containerType;

21.

22.

23. public String getConnectionFactoryJndiLookUp() {

24. return connectionFactoryJndiLookUp;

25. }

26.

27.

28.

29. public void setConnectionFactoryJndiLookUp(String connectionFactoryJndiLookUp) {

30. this.connectionFactoryJndiLookUp = connectionFactoryJndiLookUp;

31. }

32.

33.

34.

35. public String getDestinationJndiLookUp() {

36. return destinationJndiLookUp;

37. }

38.

39.

40.

41. public void setDestinationJndiLookUp(String destinationJndiLookUp) {

42. this.destinationJndiLookUp = destinationJndiLookUp;

43. }

44.

45.

46.

47. public ConnectionFactory getConnectionFactory() throws NamingException{

48. Context ctx = new InitialContext();

49. ConnectionFactory cf = (ConnectionFactory) ctx.lookup(connectionFactoryJndiLookUp);

50. return cf;

51. }

52.

53.

54. public Destination getJmsDestination() throws NamingException{

55. Context ctx = new InitialContext();

56. Destination dest = (Queue) ctx.lookup(destinationJndiLookUp);

57. return dest;

58. }

59.

60.

61. public ConnectionFactory getQueueConnectionFactory() throws NamingException{

62. Properties pops = new Properties();

63. pops.setProperty("jboss.bind.address", "0.0.0.0");

64. pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");

65. pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");

66. pops.setProperty("java.naming.provider.url", "localhost");

67. Context ctx = new InitialContext(pops);

68. ConnectionFactory cf = (ConnectionFactory) ctx.lookup(localConnectionFactoryJndiLookUp);

69. return cf;

70. }

71.

72.

73. public Destination getLocalJmsDestination() throws NamingException{

74. Properties pops = new Properties();

75. pops.setProperty("jboss.bind.address", "0.0.0.0");

76. pops.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");

77. pops.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");

78. pops.setProperty("java.naming.provider.url", "localhost");

79. Context ctx = new InitialContext(pops);

80. Destination dest = (Destination) ctx.lookup(destinationJndiLookUp);

81. return dest;

82. }

83.

84.

85.

86. public String getLocalConnectionFactoryJndiLookUp() {

87. return localConnectionFactoryJndiLookUp;

88. }

89.

90.

91.

92. public void setLocalConnectionFactoryJndiLookUp(

93. String localConnectionFactoryJndiLookUp) {

94. this.localConnectionFactoryJndiLookUp = localConnectionFactoryJndiLookUp;

95. }

96. }

发送端的配置如下

xml 代码

1.  <beans>

2.  <bean id="userJmsUtil" class="com.javaeye.spring.services.jms.mdp.UserJmsTransactionUtil">

3.  <property name="connectionFactoryJndiLookUp" value="java:JmsXA"><!---->property>

4.  <property name="destinationJndiLookUp" value="queue/A"><!---->property>

5.  <property name="localConnectionFactoryJndiLookUp" value="ConnectionFactory"><!---->property>

6.  <!---->bean>

7.

8.  <bean id="connectionFactory" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">

9.  <property name="targetObject" ref="userJmsUtil"><!---->property>

10. <property name="targetMethod" value="getConnectionFactory"><!---->property>

11. <!---->bean>

12.

13. <bean id="queue" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">

14. <property name="targetObject" ref="userJmsUtil"><!---->property>

15. <property name="targetMethod" value="getJmsDestination"><!---->property>

16. <!---->bean>

17.

18. <bean id="jmsQueue" class="org.springframework.jms.core.JmsTemplate">

19. <property name="connectionFactory" ref="connectionFactory"><!---->property>

20. <property name="defaultDestination" ref="queue"><!---->property>

21. <property name="messageConverter">

22. <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"><!---->bean>

23. <!---->property>

24. <!---->bean>

25. <!---->beans>

ps:javaeye的模版工具bug还真多,不管了.

如果使用Listenner的化,一样需要遵守发送者和接收者物理隔离的原则,我的做法是把发送者配到一个xml中,在把接受者配到另外一个xml中去,发送的配置绑定到容器里,接收者的跑在本地.否则spring初始化是过不去的.

下面这个程序是发送消息的程序.使用了spring的模版,发条消息比new个对象还简单.同时spring还提供了适配器的接口,一样通过声明式的配 置,这样可以在同一个接口里发送各种类型的消息了.同时支持事务,我还不知道这个有什么用呵呵,第1次使用嘛!但是就使用上来说,spring是最简单 的.2者都只需要注入一个对象而已.

java 代码

1.  @Test public void send(){

2.  ApplicationContext ac = new FileSystemXmlApplicationContext("jms.xml");

3.  BeanFactory bf = ac;

4.  JmsTemplate jt = (JmsTemplate) bf.getBean("jmsQueue");

5.  jt.convertAndSend("2132134");

6.  }

接收端的配置如下

xml 代码

1.  xml version="1.0" encoding="UTF-8"?>

2.  >

3.  <beans>

4.

5.  <bean id="listenner" class="com.javaeye.spring.services.jms.mdp.JmsListenner"><!---->bean>

6.

7.  <bean id="userJmsUtil" class="com.javaeye.spring.services.jms.mdp.UserJmsTransactionUtil">

8.  <property name="connectionFactoryJndiLookUp" value="java:JmsXA"><!---->property>

9.  <property name="destinationJndiLookUp" value="queue/A"><!---->property>

10. <property name="localConnectionFactoryJndiLookUp" value="ConnectionFactory"><!---->property>

11. <!---->bean>

12.

13. <bean id="localConnectionFactory" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">

14. <property name="targetObject" ref="userJmsUtil"><!---->property>

15. <property name="targetMethod" value="getQueueConnectionFactory"><!---->property>

16. <!---->bean>

17.

18. <bean id="localDestination" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">

19. <property name="targetObject" ref="userJmsUtil"><!---->property>

20. <property name="targetMethod" value="getLocalJmsDestination"><!---->property>

21. <!---->bean>

22.

23. <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">

24. <property name="concurrentConsumers" value="5"><!---->property>

25. <property name="connectionFactory" ref="localConnectionFactory"><!---->property>

26. <property name="destination" ref="localDestination"><!---->property>

27. <property name="messageListener" ref="listenner"><!---->property>

28. <!---->bean>

29. <!---->beans>

接收端由于需要从jbossmq里取ConnectionFactory和Destination,所以,我调用的是userJmsUtil的localLookup.这个函数的作用等同于发送者的那个函数,只不过前者是容器外获取,而后者是容器内的而已.

java 代码

1.  package com.javaeye.spring.services.jms.mdp;

2.

3.  import javax.jms.JMSException;

4.  import javax.jms.Message;

5.  import javax.jms.MessageListener;

6.  import javax.jms.TextMessage;

7.

8.  public class JmsListenner implements MessageListener {

9.

10. public void onMessage(Message message) {

11. try {

12. TextMessage msg = (TextMessage) message;

13. System.out.println(msg.getText());

14. } catch (JMSException e) { e.printStackTrace(); }

15. }

16.

17. }

先到[url=http://www.activemq.org]http://www.activemq.org[/url]（官网）上面去下载ActiveMQ5.0.1. 把一些毕要的jar包加入到项目的lib目录下面.

至于jar包的添加在官方网上都有介绍.

官方网上的资料信息会给你的开发带来很大的帮助.这是个人的感触虽然是全英文的.但哪些英语并不难.

下面是Spring的配置文件.

Xml代码

1.  <beans xmlns="http://www.springframework.org/schema/beans"

2.  xmlns:amq="http://activemq.apache.org/schema/core"

3.  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

4.  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schem

5.  /beans/spring-beans-2.0.xsd

6.  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd

7.  http://activemq.apache.org/camel/schema/spring http://activemq.apache.org/camel/schema/spring

8.  /camel-spring.xsd">

9.

10. <!-- 数据源 -->

11. <bean id="dataSource"

12. class="org.apache.commons.dbcp.BasicDataSource"

13. destroy-method="close">

14. <property name="driverClassName"

15. value="oracle.jdbc.driver.OracleDriver" />

16. <property name="url"

17. value="jdbc:oracle:thin:@192.168.0.100:1521:ora10" />

18. <property name="username" value="ceipportal" />

19. <property name="password" value="ceipportal" />

20. </bean>

21.

22. <!-- Spring中的jdbc模型 -->

23. <bean id="jdbcTemplate"

24. class="org.springframework.jdbc.core.JdbcTemplate">

25. <property name="dataSource">

26. <ref bean="dataSource" />

27. </property>

28. </bean>

29.

30. <!-- 配置connectionFactory -->

31. <bean id="jmsFactory"

32. class="org.apache.activemq.pool.PooledConnectionFactory"

33. destroy-method="stop">

34. <property name="connectionFactory">

35. <bean

36. class="org.apache.activemq.ActiveMQConnectionFactory">

37. <property name="brokerURL">

38. <value>tcp://localhost:61616</value>

39. </property>

40. </bean>

41. </property>

42. </bean>

43.

44. <!-- Spring JMS Template -->

45. <bean id="myJmsTemplate"

46. class="org.springframework.jms.core.JmsTemplate">

47. <property name="connectionFactory">

48. <ref local="jmsFactory" />

49. </property>

50. <property name="defaultDestinationName" value="subject" />

51. <property name="messageConverter" ref="conversion" />

52. <!-- 区别它采用的模式为false是p2p为true是订阅 -->

53. <property name="pubSubDomain" value="false" />

54. </bean>

55.

56. <!-- 读取信息 -->

57. <bean id="consumer" class="org.spring.activeDemo01.Consumer">

58. <property name="jdbcTemplate" ref="jdbcTemplate" />

59. <property name="jmsTemplate" ref="myJmsTemplate" />

60. </bean>

61.

62. <!-- 发送信息 -->

63. <bean id="producer" class="org.spring.activeDemo01.Producer">

64. <property name="jmsTemplet" ref="myJmsTemplate" />

65. <!--  <property name="destination" ref="destination"/>-->

66. <!-- <property name="destinationName" value="subject"/>    -->

67. </bean>

68.

69. <!-- 消息监听     -->

70. <bean id="listenerContainer"

71. class="org.springframework.jms.listener.DefaultMessageListenerContainer">

72. <property name="concurrentConsumers" value="5" />

73. <property name="connectionFactory" ref="jmsFactory" />

74. <property name="destinationName" value="subject" />

75. <property name="messageListener" ref="consumer" />

76. </bean>

77.

78. <!-- 消息转换     -->

79. <bean id="conversion"

80. class="org.spring.activeDemo01.PaySettlementCoverter" />

81.

82. <!-- POJO类 -->

83. <bean id="hello" class="org.spring.activeDemo01.Hello" />

84. </beans>

发送信息类

Java代码

1.  package org.spring.activeDemo01;

2.  import org.springframework.jms.core.JmsTemplate;

3.

4.  public class Producer{

5.  private JmsTemplate jmsTemplet = null ;

6.  public void setJmsTemplet(JmsTemplate jmsTemplet) {

7.  this.jmsTemplet = jmsTemplet;

8.  }

9.

10. public void simpleSend(Hello h) {

11. this.jmsTemplet.setDeliveryPersistent(true);

12. this.jmsTemplet.convertAndSend(h);

13. }

14. }

接收信息类

Java代码

1.  public class Consumer implements MessageListener{

2.  private JdbcTemplate  jdbcTemplate;

3.  private JmsTemplate jmsTemplate;

4.  private static final String insert_sql = "insert into jms_queue_send (id, MessageID, MessageDetails) values (seq_jms_queue_id.nextval, ?,?)";

5.  public JdbcTemplate getJdbcTemplate() {

6.  return jdbcTemplate;

7.  }

8.

9.  public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {

10. this.jdbcTemplate = jdbcTemplate;

11. }

12.

13. public void onMessage(Message arg0) {

14.

15. ActiveMQObjectMessage msg = (ActiveMQObjectMessage)arg0;

16. try {

17. System.out.print("From -->"+msg.getStringProperty("id"));

18. System.out.println("-->"+msg.getStringProperty("hello"));

19.

20. /*String insert_sql = "insert into jms_queue_send (id, MessageID, MessageDetails)" +

21. " values (seq_jms_queue_id.nextval, '"+msg.getJMSMessageID()+"', '"+msg.getStringProperty("hello")+"')";

22. jdbcTemplate.execute(insert_sql);*/

23.

24. } catch (JMSException e) {

25. e.printStackTrace();

26. }

27. }

28.

29. public JmsTemplate getJmsTemplate() {

30. return jmsTemplate;

31. }

32.

33. public void setJmsTemplate(JmsTemplate jmsTemplate) {

34. this.jmsTemplate = jmsTemplate;

35. }

36.

37. }

消息转换类

Java代码

1.  package org.spring.activeDemo01;

2.

3.  import javax.jms.JMSException;

4.  import javax.jms.Message;

5.  import javax.jms.ObjectMessage;

6.  import javax.jms.Session;

7.  import org.springframework.jms.support.converter.MessageConversionException;

8.  import org.springframework.jms.support.converter.MessageConverter;

9.

10. public class PaySettlementCoverter implements MessageConverter{

11. public Object fromMessage(Message message) throws JMSException,

12. MessageConversionException {

13. ObjectMessage objMsg = (ObjectMessage)message;

14. Hello hello = new Hello();

15. hello.setId(objMsg.getStringProperty("id"));

16. hello.setSaying(objMsg.getStringProperty("hello"));

17. return hello;

18. }

19.

20. public Message toMessage(Object obj, Session session) throws JMSException,

21. MessageConversionException {

22. Hello hello = (Hello)obj;

23. ObjectMessage objMsg = session.createObjectMessage();

24. objMsg.setJMSCorrelationID("123654");

25. objMsg.setJMSReplyTo(objMsg.getJMSDestination());

26. objMsg.setStringProperty("id",hello.getId());

27. objMsg.setStringProperty("hello",hello.getSaying());

28. return objMsg;

29. }

30. }

POJO类

Java代码

1.  public class Hello {

2.  /**

3.  *

4.  */

5.  private static final long serialVersionUID = 1L;

6.  private String id;

7.  private String saying;

8.  public String getId() {

9.  return id;

10. }

11. public void setId(String id) {

12. this.id = id;

13. }

14. public String getSaying() {

15. return saying;

16. }

17. public void setSaying(String saying) {

18. this.saying = saying;

19. }

20. }

测试类

Java代码

1.  import org.springframework.context.support.ClassPathXmlApplicationContext;

2.

3.  public class TestSender {

4.

5.  /**

6.  * @param args

7.  */

8.  public static void main(String[] args) {

9.  ClassPathXmlApplicationContext c = new ClassPathXmlApplicationContext("appliactionContext.xml");

10. Producer d = (Producer)c.getBean("producer");

11. Hello h = (Hello)c.getBean("hello");

12. h.setId("123456789");

13. h.setSaying("Hello World.....!!! ");

14. d.simpleSend(h);

15. }

16.

Spring与Jms结合的实例

博客分类：

·        JMS

另见：

ActiveMQ5.0实战一: 安装配置ActiveMQ5.0

ActiveMQ5.0实战二: 基本配置

http://www.iteye.com/topic/234101

================================================

环境：apache-activemq-5.4.2 ,spring2.5,jdk1.6

首先启动消息代理：在apache-activemq-5.4.2\bin下
运行
activemq.bat。

example1：

package jms.activemq.myexample.spring;

import java.util.Date;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.springframework.jms.core.MessageCreator;

public class MyMessageCreator implements MessageCreator {

/**
  * 消息序号
*/
 private int msgNo;

public MyMessageCreator(int no) {
  this.msgNo = no;
 }

@Override
 public Message createMessage(Session session) throws JMSException {
  TextMessage textMsg = session.createTextMessage();
  textMsg.setText(new Date() + "第" + this.msgNo + "条消息发出");

return textMsg;
 }

}

package jms.activemq.myexample.spring;

import javax.jms.*;

/**
 * Text消息监听
*
 * @author xiaochuanyu
 *
 */
public class TextListener implements MessageListener {

/**
  * Casts the message to a TextMessage and displays its text.
  *
  * @param message
  *            theincoming message
  */
 public void onMessage(Message message) {
  TextMessage msg = null;

try {
   if (message instanceof TextMessage) {
    msg = (TextMessage) message;
    System.out.println("Reading message: " +msg.getText());
   } else {
    System.out.println("Message of wrong type: "
      + message.getClass().getName());
   }
  } catch (JMSException e) {
   System.out.println("JMSException in onMessage(): "+ e.toString());
  } catch (Throwable t) {
   System.out.println("Exception in onMessage():" +t.getMessage());
  }
 }
}

package jms.activemq.myexample.spring;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SpringJmsTestMain {

public static void main(String[] args) throwsInterruptedException {

ApplicationContext context = newClassPathXmlApplicationContext(
    new String[] { "SpringJms/SpringJms.xml" });

SpringPublisher publisher = (SpringPublisher)context
    .getBean("springPublisher");
  publisher.start();
 }
}

package jms.activemq.myexample.spring;

import javax.jms.Destination;

import org.springframework.jms.core.JmsTemplate;

public class SpringPublisher {
 /**
  * Jms模板
*/
 private JmsTemplate template;

/**
  * Topic
  */
 private Destination topic;

public JmsTemplate getTemplate() {
  return template;
 }

public void setTemplate(JmsTemplate template) {
  this.template = template;
 }

public Destination getTopic() {
  return topic;
 }

public void setTopic(Destination topic) {
  this.topic = topic;
 }

/**
  * Start
  *
  * @throws InterruptedException
  */
 public void start() throws InterruptedException {

int messageCount = 10;

while ((--messageCount) > 0) {
   sendMessage(messageCount);
   Thread.sleep(1000);
  }
 }

/**
  * 消息发送
*/
 protected void sendMessage(int msgNO) {

this.template.send(this.topic, newMyMessageCreator(msgNO));
 }
}

<?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/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">

<!-- jms 连接工厂 -->
    <bean id="jmsFactory"class="org.apache.activemq.ActiveMQConnectionFactory">
        <propertyname="brokerURL">
           <value>tcp://localhost:61616</value>
        </property>
    </bean>

<!-- jms 连接池 -->
   
    <!-- 
    <bean id="pooledJmsFactory"class="org.apache.activemq.pool.PooledConnectionFactory">
        <propertyname="connectionFactory">
            <reflocal="jmsFactory" />
        </property>
    </bean>
    -->

<!-- jms Topic -->
    <bean id="myTopic"class="org.apache.activemq.command.ActiveMQTopic"
        autowire="constructor">
        <constructor-argvalue="STOCKS.JAVA" />
    </bean>

<!-- 消息监听器 -->
    <bean id="myTextListener"class="jms.activemq.myexample.spring.TextListener">
    </bean>

<!-- jms Consumer -->
    <bean id="javaConsumer"
       class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <propertyname="connectionFactory" ref="jmsFactory" />
        <propertyname="destination" ref="myTopic" />
        <propertyname="messageListener" ref="myTextListener" />
    </bean>

<!-- jms 模板 -->
    <bean id="jmsTemplate"class="org.springframework.jms.core.JmsTemplate">
        <propertyname="connectionFactory">
            <reflocal="jmsFactory" />
        </property>
    </bean>

<!-- 消息发布器 -->
    <bean id="springPublisher"class="jms.activemq.myexample.spring.SpringPublisher">
        <propertyname="template">
            <reflocal="jmsTemplate" />
        </property>
        <property name="topic">
            <reflocal="myTopic" />
        </property>
    </bean>
</beans>

===================================================

example2：

package stujms.p2ptxt;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

/**
* 消息发送者
*
* @author leizhimin 2009-8-13 17:01:48
*/
public class MySender {
        public static void main(String[]args) {
               ApplicationContext ctx = newClassPathXmlApplicationContext("/applicationContext.xml");
               JmsTemplate template = (JmsTemplate) ctx.getBean("jmsTemplate");
               Destination destination = (Destination) ctx.getBean("destination");

template.send(destination, new MessageCreator() {
                       public Message createMessage(Session session) throws JMSException {
                               return session.createTextMessage("发送消息：Hello ActiveMQ Text Message！");
                       }
               });
               System.out.println("成功发送了一条JMS消息");
        }
}
package stujms.p2ptxt;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;

/**
* 消息接收者
*
* @author leizhimin 2009-8-13 17:02:04
*/
public class MyReceiver {
        public static void main(String[]args) throws JMSException {
               ApplicationContext ctx = newClassPathXmlApplicationContext("/applicationContext.xml");
               JmsTemplate template = (JmsTemplate) ctx.getBean("jmsTemplate");
               Destination destination = (Destination) ctx.getBean("destination");
               while (true) {
                       TextMessage txtmsg = (TextMessage) template.receive(destination);
                       if (null != txtmsg)
                               System.out.println("收到消息内容为: " +txtmsg.getText());
                       else
                               break;
               }
        }
}
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xmlns:context="http://www.springframework.org/schema/context"
            xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd">

<!-- 配置JMS连接工厂 -->
        <beanid="connectionFactory"class="org.apache.activemq.spring.ActiveMQConnectionFactory">
               <property name="brokerURL"value="tcp://localhost:61616"/>
        </bean>

<!-- 配置JMS模版 -->
        <bean id="jmsTemplate"class="org.springframework.jms.core.JmsTemplate">
               <property name="connectionFactory"ref="connectionFactory"/>
        </bean>

<!-- 发送消息的目的地（一个队列） -->
        <bean id="destination"class="org.apache.activemq.command.ActiveMQQueue">
               <!-- 设置消息队列的名字 -->
               <constructor-arg index="0" value="HelloWorldQueue"/>
        </bean>
</beans>
运行发送端三次：
成功发送了一条JMS消息

第一: 在applicationContext.xml

Xml代码

1.  <?xml version="1.0" encoding="UTF-8"?>

2.  <!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">

3.

4.  <beans>

5.  <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

6.  <property name="brokerURL">

7.  <value>tcp://192.168.3.48:61616</value>

8.  </property>

9.

10. </bean>

11.

12. <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

13. <property name="connectionFactory">

14. <ref bean="connectionFactory"/>

15. </property>

16. </bean>

17.

18. <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">

19. <constructor-arg index="0">

20. <value>HelloWorldQueue</value>

21. </constructor-arg>

22. </bean>

23. </beans>

2.写发送方

Java代码

1.  package ch13.JMS;

2.

3.  import javax.jms.Destination;

4.  import javax.jms.JMSException;

5.  import javax.jms.Message;

6.  import javax.jms.Session;

7.

8.  import org.springframework.context.ApplicationContext;

9.  import org.springframework.context.support.ClassPathXmlApplicationContext;

10. import org.springframework.jms.core.JmsTemplate;

11. import org.springframework.jms.core.MessageCreator;

12.

13. public class HelloWorldSender

14. {

15. public static void main(String args[]) throws Exception

16. {

17. ApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "ch13/JMS/applicationContext.xml" });

18. JmsTemplate jmsTemplate = (JmsTemplate) context.getBean("jmsTemplate");

19. Destination destination = (Destination) context.getBean("destination");

20. jmsTemplate.send

21. (

22. destination, new MessageCreator()

23. {

24. public Message createMessage(Session session) throws JMSException

25. {

26. return session.createTextMessage("大家好这个是测试！");

27. }

28. }

29. );

30. }

31.

32. }

3.写接收方

Java代码

1.  package ch13.JMS;

2.

3.  import javax.jms.Destination;

4.  import javax.jms.TextMessage;

5.

6.  import org.springframework.context.ApplicationContext;

7.  import org.springframework.context.support.ClassPathXmlApplicationContext;

8.  import org.springframework.jms.core.JmsTemplate;

9.

10. public class HelloWorldReciver

11. {

12.

13. public static void main(String args[]) throws Exception

14. {

15. ApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "ch13/JMS/applicationContext.xml" });

16. JmsTemplate jmsTemplate = (JmsTemplate) context.getBean("jmsTemplate");

17. Destination destination = (Destination) context.getBean("destination");

18. System.out.println("will wait:" + jmsTemplate.getReceiveTimeout()+ " seconds for message");

19. TextMessage msg = (TextMessage) jmsTemplate.receive(destination);

20. System.out.println("reviced msg is:" + msg.getText());

21. }

22.

23. }

ActiveMQ+Spring2.5(转)

今天看书看了下JMS，初步有了个了解，具体的JMS知识还需要进一步的深入学习。
J2EE中，JMS（Java Message System）提供了一种异步处理机制的实现。JMS通过异步的、非阻塞的消息传递，将消息的生产者和使用者松散的联系在一起。对于使用者，它无所谓是谁 产生了消息或者是在何时产生的。这就能够建立一种动态的、灵活的可靠的系统。所谓可靠，因为JMS将每个消息都保存起来，只有确保本消息处理后才会完全放 弃。否则，将会反复提交处理。这种可靠的机制使得JMS能够成功的在证券、银行系统中得到广泛应用。
JMS中的消息类型有两种：Topic和Queue。Topic的操作是使用发布/订阅（publish/subscribe）的方式；Queue的操作是点对点（ponit to point）的方式。
·publish/subscribe：发布者（publisher）发布消息到Topic，订阅者（subsribe）从Topic订阅消息，订阅者的数量是不受限制的。
· ponit to point：点对点传输消息，建立在消息队列Queue的基础上，每个消息使用者只对应一个消息队列，不像publish/subscribe那样可以有很多消息使用者。
JMS在消息到达消息使用者，有两种——同步和异步。
·同步是指消息使用者明确地主动地调用从Queue或Topic中得到消息，一直进行循环直至一个消息到达，或者也可以设定超时时间。很明显这个方法是比较耗费CPU资源的。
·异步接受是指消息到达时，主动通知消息使用者，消息使用者可以实现message listener接口。这样每当消息到达时，JMS provider 会通过调用这个listener的onMessage方法来传输这个消息。
参考spring的资料，写了个Spring+JMS的例子。
1.建立工程spring_jms，导入工程相应的包，所用的包都是ActiveMQ自带的，在apache-activemq-5.2.0\lib\optional下可以找到。

2.applicationContext.xml

Java代码

1.  <?xml version="1.0" encoding="UTF-8"?>

2.  <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

3.  xmlns:context="http://www.springframework.org/schema/context"

4.  xsi:schemaLocation="http://www.springframework.org/schema/beans

5.  http://www.springframework.org/schema/beans/spring-beans-2.5.xsd

6.  http://www.springframework.org/schema/context

7.  http://www.springframework.org/schema/context/spring-context-2.5.xsd">

8.

9.  <!-- 配置JMS连接工厂 -->

10. <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">

11. <property name="brokerURL" value="tcp://localhost:61616"/>

12. </bean>

13.

14. <!-- 配置JMS模版 -->

15. <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

16. <property name="connectionFactory" ref="connectionFactory"/>

17. </bean>

18.

19. <!-- 发送消息的目的地（一个队列） -->

20. <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">

21. <!-- Set the Queue Name -->

22. <constructor-arg index="0" value="HelloWorldQueue"/>

23. </bean>

24. </beans>

3.消息发送程序HelloWorldSender

Java代码

1.  package com.springjms;

2.

3.  import org.springframework.context.ApplicationContext;

4.  import org.springframework.context.support.ClassPathXmlApplicationContext;

5.  import org.springframework.jms.core.JmsTemplate;

6.  import org.springframework.jms.core.MessageCreator;

7.

8.  import javax.jms.Destination;

9.  import javax.jms.JMSException;

10. import javax.jms.Message;

11. import javax.jms.Session;

12.

13. /**

14. *

15. * @author Wind

16. *

17. */

18. public class HelloWorldSender {

19. public static void main(String[] args) {

20. ApplicationContext ctx = new ClassPathXmlApplicationContext("/applicationContext.xml");

21. JmsTemplate template = (JmsTemplate) ctx.getBean("jmsTemplate");

22. Destination destination = (Destination) ctx.getBean("destination");

23.

24. template.send(destination, new MessageCreator() {

25. public Message createMessage(Session session) throws JMSException {

26. return session.createTextMessage("Hello World,ActiveMQ Spring JMS Message！");

27. }

28. });

29. System.out.println("JMS Message Sent");

30. }

31. }

4.消息接收程序HelloWorldReceiver

Java代码

1.  package com.springjms;

2.

3.  import org.springframework.context.ApplicationContext;

4.  import org.springframework.context.support.ClassPathXmlApplicationContext;

5.  import org.springframework.jms.core.JmsTemplate;

6.

7.  import javax.jms.Destination;

8.  import javax.jms.JMSException;

9.  import javax.jms.TextMessage;

10.

11. /**

12. *

13. * @author Wind

14. *

15. */

16. public class HelloWorldReceiver {

17. public static void main(String[] args) throws JMSException {

18. ApplicationContext ctx = new ClassPathXmlApplicationContext("/applicationContext.xml");

19. JmsTemplate template = (JmsTemplate) ctx.getBean("jmsTemplate");

20. Destination destination = (Destination) ctx.getBean("destination");

21. while (true) {

22. TextMessage txtmsg = (TextMessage) template.receive(destination);

23. if (null != txtmsg){

24. System.out.println("Received Message: " + txtmsg.getText());

25. }else{

26. break;

27. }

28. }

29. }

30. }

5.启动ActiveMQ，先运行发送者HelloWorldSender，然后运行HelloWorldSenderReceiver，控制台输出如下：

Java代码

1.  Received Message: Hello World,ActiveMQ Spring JMS Message！

1.       集成环境

Spring采用2.5.6版本，ActiveMQ使用的是5.4.2，从apache站点可以下载。本文是将Spring集成ActiveMQ来发送和接收JMS消息。

2.       集成步骤

将下载的ActiveMQ解压缩后文件夹如下

activemq-all-5.4.2.jar是activemq的所有的类jar包。lib下面是模块分解后的jar包。将lib下面的

Java代码

1.  /lib/activation-1.1.jar

2.  /lib/activemq-camel-5.4.2.jar

3.  /lib/activemq-console-5.4.2.jar

4.  /lib/activemq-core-5.4.2.jar

5.  /lib/activemq-jaas-5.4.2.jar

6.  /lib/activemq-pool-5.4.2.jar

7.  /lib/activemq-protobuf-1.1.jar

8.  /lib/activemq-spring-5.4.2.jar

9.  /lib/activemq-web-5.4.2.jar

文件全部拷贝到项目中。

而Spring项目所需要的jar包如下

Java代码

1.  /lib/spring-beans-2.5.6.jar

2.  /lib/spring-context-2.5.6.jar

3.  /lib/spring-context-support-2.5.6.jar

4.  /lib/spring-core-2.5.6.jar

5.  /lib/spring-jms-2.5.6.jar

6.  /lib/spring-tx.jar

当然还需要一些其他的jar文件

Java代码

1.  /lib/geronimo-j2ee-management_1.1_spec-1.0.1.jar

2.  /lib/jms-1.1.jar

3.  /lib/log4j-1.2.15.jar

4.  /lib/slf4j-api-1.6.1.jar

5.  /lib/slf4j-nop-1.6.1.jar

项目的依赖jar都准备好后就可以写配置文件了。

Spring配置文件

配置文件内容如下

Java代码

1.  <?xml version="1.0" encoding="UTF-8"?>

2.  <beans xmlns="http://www.springframework.org/schema/beans"

3.  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"

4.  xsi:schemaLocation="http://www.springframework.org/schema/beans

5.  http://www.springframework.org/schema/beans/spring-beans-2.5.xsd

6.  http://www.springframework.org/schema/context

7.  http://www.springframework.org/schema/context/spring-context-2.5.xsd"

8.  default-autowire="byName">

9.

10.

11. <!-- 配置connectionFactory -->

12. <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory"

13. destroy-method="stop">

14. <property name="connectionFactory">

15. <bean class="org.apache.activemq.ActiveMQConnectionFactory">

16. <property name="brokerURL">

17. <value>tcp://127.0.0.1:61616</value>

18. </property>

19. </bean>

20. </property>

21. <property name="maxConnections" value="100"></property>

22. </bean>

23.

24. <!-- Spring JMS Template -->

25. <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

26. <property name="connectionFactory">

27. <ref local="jmsFactory" />

28. </property>

29. <property name="defaultDestinationName" value="subject" />

30. <!-- 区别它采用的模式为false是p2p为true是订阅 -->

31. <property name="pubSubDomain" value="true" />

32. </bean>

33.

34. <!-- 发送消息的目的地（一个队列） -->

35. <bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">

36. <!-- 设置消息队列的名字 -->

37. <constructor-arg index="0" value="subject" />

38. </bean>

39.

40. <!-- 消息监听     -->

41. <bean id="listenerContainer"

42. class="org.springframework.jms.listener.DefaultMessageListenerContainer">

43. <property name="concurrentConsumers" value="10" />

44. <property name="connectionFactory" ref="jmsFactory" />

45. <property name="destinationName" value="subject" />

46. <property name="messageListener" ref="messageReceiver" />

47. <property name="pubSubNoLocal" value="false"></property>

48. </bean>

49.

50. <bean id="messageReceiver"

51. class="com.liuyan.jms.consumer.ProxyJMSConsumer">

52. <property name="jmsTemplate" ref="jmsTemplate"></property>

53. </bean>

54. </beans>

编写代码

消息发送者：这里面消息生产者并没有在Spring配置文件中进行配置，这里仅仅使用了Spring中的JMS模板和消息目的而已。

Java代码

1.  public class HelloSender {

2.

3.  /**

4.  * @param args

5.  */

6.  public static void main(String[] args) {

7.  ApplicationContext applicationContext = new ClassPathXmlApplicationContext(

8.  new String[] { "classpath:/spring/applicationContext-jms.xml" });

9.

10. JmsTemplate template = (JmsTemplate) applicationContext

11. .getBean("jmsTemplate");

12.

13. Destination destination = (Destination) applicationContext

14. .getBean("destination");

15.

16. template.send(destination, new MessageCreator() {

17. public Message createMessage(Session session) throws JMSException {

18. return session

19. .createTextMessage("发送消息：Hello ActiveMQ Text Message！");

20. }

21. });

22. System.out.println("成功发送了一条JMS消息");

23.

24. }

25.

26. }

消息接收

Java代码

1.  /**

2.  * JMS消费者

3.  *

4.  *

5.  * <p>

6.  * 消息题的内容定义

7.  * <p>

8.  * 消息对象 接收消息对象后： 接收到的消息体* <p>

9.  */

10. public class ProxyJMSConsumer {

11.

12. public ProxyJMSConsumer() {

13.

14. }

15.

16. private JmsTemplate jmsTemplate;

17.

18. public JmsTemplate getJmsTemplate() {

19. return jmsTemplate;

20. }

21.

22. public void setJmsTemplate(JmsTemplate jmsTemplate) {

23. this.jmsTemplate = jmsTemplate;

24. }

25.

26. /**

27. * 监听到消息目的有消息后自动调用onMessage(Message message)方法

28. */

29. public void recive() {

30. ApplicationContext applicationContext = new ClassPathXmlApplicationContext(

31. new String[] { "classpath:/spring/applicationContext-jms.xml" });

32.

33. Destination destination = (Destination) applicationContext

34. .getBean("destination");

35.

36. while (true) {

37.

38. try {

39. TextMessage txtmsg = (TextMessage) jmsTemplate

40. .receive(destination);

41. if (null != txtmsg) {

42. System.out.println("[DB Proxy] " + txtmsg);

43. System.out.println("[DB Proxy] 收到消息内容为: "

44. + txtmsg.getText());

45. } else

46. break;

47. } catch (Exception e) {

48. e.printStackTrace();

49. }

50.

51. }

52. }

53.

54. }

这里边也是并不是直接使用Spring来初始化建立消息消费者实例，而是在此消费者注入了JMS模板而已。

写一个main入口，初始化消息消费者

Java代码

1.  public class JMSTest {

2.

3.  /**

4.  * @param args

5.  */

6.  public static void main(String[] args) {

7.  ApplicationContext applicationContext = new ClassPathXmlApplicationContext(

8.  new String[] { "classpath:/spring/applicationContext-jms.xml" });

9.

10. ProxyJMSConsumer proxyJMSConsumer = (ProxyJMSConsumer) applicationContext

11. .getBean("messageReceiver");

12.

13. System.out.println("初始化消息消费者");

14. }

15.

16. }

Spring2.5整合ActiveMQ 5.2（P2P文本消息）

2009-08-13 17:36:52

标签：休闲ActiveMQ职场

版权声明：原创作品，如需转载，请与作者联系。否则将追究法律责任。

Spring2.5整合ActiveMQ 5.2（P2P文本消息）

想找个Spring整合ActiveMQ可运行的实例，快速看看效果，可是很失望，网上例子都是抄来抄去，没源码（或者有源码运行不了），真TM没意思，看文档我自己实现个，不能运行你砍我！

不想把例子搞的太复杂，越简单越好！

项目环境：

JDK1.5

ActiveMQ5.2

所用的包都是ActiveMQ自带的。引用的包如下图：

package stujms.p2ptxt; import org.springframework.context.ApplicationContext; importorg.springframework.context.support.ClassPathXmlApplicationContext; importorg.springframework.jms.core.JmsTemplate; importorg.springframework.jms.core.MessageCreator; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; /**
* 消息发送者*
* @author leizhimin 2009-8-13 17:01:48
*/ publicclass MySender {publicstaticvoid main(String[] args) {
                ApplicationContextctx = new ClassPathXmlApplicationContext("/applicationContext.xml");
                JmsTemplatetemplate = (JmsTemplate) ctx.getBean("jmsTemplate");
                Destinationdestination = (Destination) ctx.getBean("destination");

template.send(destination,new MessageCreator() {public Message createMessage(Sessionsession) throwsJMSException {return session.createTextMessage("发送消息：Hello ActiveMQ Text Message！");
                        }
                });
                System.out.println("成功发送了一条JMS消息");
        }
}

package stujms.p2ptxt; importorg.springframework.context.ApplicationContext; importorg.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.jms.core.JmsTemplate; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.TextMessage; /**
* 消息接收者*
* @author leizhimin 2009-8-13 17:02:04
*/ publicclass MyReceiver {publicstaticvoid main(String[] args) throws JMSException {
                ApplicationContextctx = newClassPathXmlApplicationContext("/applicationContext.xml");
                JmsTemplatetemplate = (JmsTemplate) ctx.getBean("jmsTemplate");
                Destinationdestination = (Destination) ctx.getBean("destination");while (true) {
                        TextMessagetxtmsg = (TextMessage) template.receive(destination);if (null != txtmsg)
                                System.out.println("收到消息内容为: " + txtmsg.getText());else break;
                }
        }
}

<?xmlversion="1.0"encoding="UTF-8"?> <beansxmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context-2.5.xsd"> <!--配置JMS连接工厂--><beanid="connectionFactory"class="org.apache.activemq.spring.ActiveMQConnectionFactory"> <propertyname="brokerURL"value="tcp://localhost:61616"/> </bean><!--配置JMS模版--><beanid="jmsTemplate"class="org.springframework.jms.core.JmsTemplate"> <propertyname="connectionFactory"ref="connectionFactory"/> </bean><!--发送消息的目的地（一个队列）--><beanid="destination"class="org.apache.activemq.command.ActiveMQQueue"> <!--设置消息队列的名字--><constructor-argindex="0"value="HelloWorldQueue"/> </bean></beans>

运行发送端三次：

成功发送了一条JMS消息Process finished with exit code 0

然后再运行接收端一次：

收到消息内容为: 发送消息：Hello ActiveMQ Text Message！ 收到消息内容为: 发送消息：Hello ActiveMQ Text Message！ 收到消息内容为: 发送消息：Hello ActiveMQ Text Message！

继续测试发现，接收端接收一条消息后不退出程序，而是继续等待，一旦有消息发送过来，就获取到，然后输出！

发一张图看看：

ActiveMQ与Spring整合

ActiveMQ 是Apache出品， 是最流行​​和最强大的开源消息总线。 同时完全支持 JMS 1.1和J2EE 1.4规范。

ActiveMQ 特性

1.      支持多种编程语言和协议编写客户端。

2.      在JMS客户端和消息代理完全支持企业集成模式

3.      完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)。

4.      对Spring的支持, ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性。

5.      测试通过常见的J2EE服务器。如：Geronimo, JBoss 4, GlassFish, WebLogic。

6.      支持多种传送协议：in-VM,TCP, SSL, NIO, UDP, JGroups,JXTA。

7.      支持通过JDBC和journal提供高速的消息持久化。

8.      从设计上保证了高性能的集群,客户端-服务器,点对点。

9.      支持Ajax

10.  支持与Axis的整合

11.  可以很容易得调用内嵌JMSprovider,进行测试。

通过网络搜索及个人理解，整理出ActiveMQ与Spring整合的文章。

相关jar包

activemq-core-5.5.1.jar
geronimo-j2ee-management_1.1_spec-1.0.1.jar
geronimo-jms_1.1_spec-1.1.1.jar
geronimo-jta_1.0.1B_spec-1.0.1.jar
kahadb-5.5.1.jar
xbean-spring-3.7.jar
commons-beanutils.jar
commons-codec.jar
commons-collections.jar
commons-fileupload.jar
commons-httpclient.jar
commons-io.jar
commons-lang.jar
commons-logging.jar
commons-validator.jar
dom4j-1.6.1.jar
javaee.jar
jsf-api.jar
jsf-impl.jar
jstl.jar
log4j-1.2.15.jar
slf4j-api-1.5.8.jar
slf4j-log4j12-1.5.8.jar
aspectjrt.jar
aspectjweaver.jar
spring.jar

spring.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" xmlns:p="http://www.springframework.org/schema/p" xmlns:tx="http://www.springframework.org/schema/tx"     xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context" xmlns:util="http://www.springframework.org/schema/util" xmlns:jee="http://www.springframework.org/schema/jee"     xmlns:amq="http://activemq.apache.org/schema/core" xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-2.5.xsd
      http://www.springframework.org/schema/tx
       http://www.springframework.org/schema/tx/spring-tx-2.5.xsd
      http://www.springframework.org/schema/aop
      http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
      http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context-2.5.xsd
      http://www.springframework.org/schema/util
      http://www.springframework.org/schema/util/spring-util-2.5.xsd
      http://www.springframework.org/schema/jee
       http://www.springframework.org/schema/jee/spring-jee-2.5.xsd
      http://activemq.apache.org/schema/core
      http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd"> 　　<!-- Annotation Config -->
　　<context:annotation-config/>
　　<!-- Compoent Scan -->
　　<context:component-scan base-package="com.uu.web.*"/>
　　<!-- Property Placeholder-->
　　<context:property-placeholder location="classpath:config.properties"/>
　　<!-- Aop Config -->
　　<aop:aspectj-autoproxy/>
<!-- <beanid="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource"destroy-method="close"> 　　　　<property name="driverClassName"value="${dirver}"/>
　　　　<property name="url"value="${url}"/>
　　　　<property name="username"value="${username}"/>
　　　　<property name="password"value="${password}"/>
　　</bean>
　　-->
<!-- 　　　　使用spring的listenerContainer 消息用持久化保存，服务器重启不会丢失
　　　　也可以配置在${ACTIVEMQ_HOME}/conf/activemq.xml内
　　　　消息的保存方式文件持久化和数据库持久化
　　　　此配置是文件持久化
　　-->
　　<!-- Embedded ActiveMQ Broker-->
　　<amq:broker useJmx="false" persistent="true">
　　　　<amq:persistenceAdapter>
　　　　　　<amq:amqPersistenceAdapter directory="G:/amq"/>
　　　　　　<!-- 使用数据库持久化 -->
　　　　　　<!--<amq:jdbcPersistenceAdapterdataSource="#mysql-ds" />-->
　　　　</amq:persistenceAdapter>
　　　　<amq:transportConnectors>
　　　　　　<amq:transportConnector uri="tcp://localhost:61616" />
　　　　</amq:transportConnectors>
　　</amq:broker>
　　<!--ActiveMQconnectionFactory -->
　　<amq:connectionFactory id="jmsConnectionFactory" brokerURL="vm://localhost" />
   
　　<!-- ActiveMQdestinations -->
<!-- 　　　　TOPIC：发布订阅消息
　　　　　　　　无状态，不保证每条消息被消费
　　　　　　　　只有监听该TOPIC地址才能收到消息并消费，否则该消息将会丢失
　　　　　　　　一对多的发布接受策略，可以同时消费多个消息
　　-->
　　<amq:topic name="TOPIC" physicalName="JMS-TOPIC" /> <!-- 　　　　QUEUE：　点对点
　　　　　　　　　消息数据被持久化，每条消息都能被消费
　　　　　　　　　没有监听QUEUE地址也能被消费，数据不会丢失
　　　　　　　　　一对一的发布接受策略，保证数据完整
-->
<amq:queue name="QUEUE"physicalName="JMS-QUEUE" /> 　　<!-- ConnectionFactory -->
　　<bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory" destroy-method="destroy">
　　　　<property name="targetConnectionFactory" ref="jmsConnectionFactory"/>
　　</bean>
　　<!-- 添加事务 -->
　　<bean id="jmsTransactionManager" class="org.springframework.jms.connection.JmsTransactionManager" >
　　　　<property name="connectionFactory" ref="singleConnectionFactory"/>
　　</bean>
　　<!-- Spring JmsTemplate config -->
　　<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
　　　　<!-- lets wrap in a pool to avoid creating aconnection per send -->
　　　　<property name="connectionFactory" ref="singleConnectionFactory"/>
　　　　<!-- custom MessageConverter-->
　　　　<property name="messageConverter" ref="defaultMessageConverter" />
　　　　<property name="sessionTransacted" value="true"/>
　　</bean>
　　<!-- converter -->
　　<bean id="defaultMessageConverter" class="com.uu.activemq.DefaultMessageConverter" />
   
　　<!-- 生产消息配置 -->
　　<!-- POJO which send Messageuses Spring JmsTemplate -->
　　<bean id="topicMessageProducer" class="com.uu.activemq.TopicMessageProducer">
　　　　<property name="template" ref="jmsTemplate" />
　　　　<property name="destination" ref="TOPIC" />
　　</bean>
　　<bean id="queueMessageProducer" class="com.uu.activemq.QueueMessageProducer">
　　　　<property name="template" ref="jmsTemplate" />
　　　　<property name="destination" ref="QUEUE" />
　　</bean>
　　<!-- 消费消息 配置 -->
　　<!-- Message Driven POJO (MDP) -->
　　<!-- consumer1 for topic-->
　　<bean id="topicConsumer" class="com.uu.activemq.TopicConsumer" />

　　<!-- consumer for queue -->
　　<bean id="queueConsumer" class="com.uu.activemq.QueueConsumer" />
   
　　<!-- Message Listener for -->
　　<bean id="topicListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
　　　　<constructor-arg ref="topicConsumer" />
　　　　<!-- 指定消费消息的方法 -->
　　　　<property name="defaultListenerMethod" value="receive" />
　　　　<!-- custom MessageConverterdefine -->
　　　　<property name="messageConverter" ref="defaultMessageConverter" />
　　</bean>
　　<bean id="queueListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
　　　　<constructor-arg ref="queueConsumer" />
　　　　<!-- 指定消费消息的方法 -->
　　　　<property name="defaultListenerMethod" value="receive" />
　　　　<!-- custom MessageConverterdefine -->
　　　　<property name="messageConverter" ref="defaultMessageConverter" />
　　</bean>
　　<!-- listener container，MDP无需实现接口 -->
　　<bean id="topicListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
　　　　<property name="connectionFactory" ref="jmsConnectionFactory" />
　　　　<property name="destination" ref="TOPIC" />
　　　　<property name="messageListener" ref="topicListener" />
　　</bean>
　　<bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
　　　　<property name="connectionFactory" ref="jmsConnectionFactory" />
　　　　<property name="destination" ref="QUEUE" />
　　　　<property name="messageListener" ref="queueListener" />
　　　　<property name="transactionManager" ref="jmsTransactionManager"/>
　　　　<property name="sessionTransacted" value="true"/>
　　　　<property name="concurrentConsumers" value="5"/>
　　</bean>
</beans>

消息转换类

DefaultMessageConverter.java

1 package com.uu.activemq;
2
3 import java.io.ByteArrayInputStream;
4 import java.io.ByteArrayOutputStream;
5 import java.io.IOException;
6 import java.io.ObjectInputStream;
7 import java.io.ObjectOutputStream;
8 import java.util.HashMap;
9
10 importjavax.jms.JMSException;
11 importjavax.jms.Message;
12 importjavax.jms.ObjectMessage;
13 importjavax.jms.Session;
14
15 importorg.apache.activemq.command.ActiveMQObjectMessage;
16 importorg.apache.commons.logging.Log;
17 importorg.apache.commons.logging.LogFactory;
18 importorg.springframework.jms.support.converter.MessageConverter;
19
20 /** 21 * 消息转换类 22*/
23 @SuppressWarnings("unchecked")
24 public class DefaultMessageConverter implements MessageConverter {
25    
26     private static final Log log =LogFactory.getLog(DefaultMessageConverter.class);
27
28     public Message toMessage(Object obj, Session session)throws JMSException {
29        if (log.isDebugEnabled()) {
30            log.debug("toMessage(Object, Session) - start");
31        }
32
33// check Type 34         ActiveMQObjectMessage objMsg =(ActiveMQObjectMessage) session.createObjectMessage();
35        HashMap<String, byte[]> map = new HashMap<String, byte[]>();
36        try {
37// POJO must implements Seralizable 38             ByteArrayOutputStream bos = new ByteArrayOutputStream();
39            ObjectOutputStream oos = newObjectOutputStream(bos);
40            oos.writeObject(obj);
41            map.put("POJO", bos.toByteArray());
42            objMsg.setObjectProperty("Map", map);
43
44        } catch (IOException e) {
45            log.error("toMessage(Object, Session)", e);
46        }
47        return objMsg;
48     }
49
50    
51     public Object fromMessage(Message msg) throws JMSException {
52        if (log.isDebugEnabled()) {
53            log.debug("fromMessage(Message) - start");
54        }
55
56        if (msg instanceofObjectMessage) {
57            HashMap<String, byte[]> map =(HashMap<String, byte[]>)((ObjectMessage) msg).getObjectProperty("Map");
58            try {
59 //POJO must implements Seralizable 60                 ByteArrayInputStream bis = new ByteArrayInputStream(map.get("POJO"));
61                 ObjectInputStream ois = new ObjectInputStream(bis);
62                 Object returnObject =ois.readObject();
63                 returnreturnObject;
64            } catch (IOException e) {
65                log.error("fromMessage(Message)",e);
66
67            } catch (ClassNotFoundException e) {
68                log.error("fromMessage(Message)", e);
69            }
70
71            return null;
72        } else {
73            throw newJMSException("Msg:[" + msg + "] is not Map");
74        }
75     }
76 }

接受Queue方式消费

QueueConsumer.java

1 package com.uu.activemq;
2
3 import java.util.Map;
4 import java.util.Set;
5
6 public classQueueConsumer
7 {
8     public void receive(Map<String, Object> message)
9     {
10        Set<String> set = message.keySet();
11        String str = "";
12        for(String key : set)
13        {
14            str += key + "_" + message.get(key) + ":Queue";
15        }
16        System.out.println(str);
17     }
18 }

接受Topic方式消费

TopicConsumer.java

1 package com.uu.activemq;
2
3 import java.util.Map;
4 import java.util.Set;
5 import java.util.concurrent.ExecutorService;
6 import java.util.concurrent.Executors;
7
8 /** 9* 创建10个线程池
10 * 1.使用线程异步消息处理 11 * 2.不使用线程，那么消息等待上一个消息处理完成后才继续 12 * 注:如果担心据同步问题，那么使用第2种方法
13*/
14 public class TopicConsumer
15 {
16     protected staticExecutorService exec = Executors.newFixedThreadPool(10);
17    
18     public voidreceive(Map<String, Object> message)
19     {
20/*exec.submit(new Runnable(){ 21 22 public void run()
23 {
24 25 }
26 });*/
27        Set<String> set = message.keySet();
28        String str = "";
29        for(String key : set)
30        {
31            str += key + "_" + message.get(key) + ":Topic";
32        }
33        System.out.println(str);
34     }
35 }

生产Queue方式的消息

QueueMessageProducer.java

1 package com.uu.activemq;
2
3 import java.util.Map;
4
5 import javax.jms.Queue;
6
7 import org.springframework.jms.core.JmsTemplate;
8
9 public classQueueMessageProducer {
10
11     private JmsTemplate template;
12
13     private Queue destination;
14
15     public voidsetTemplate(JmsTemplate template) {
16        this.template = template;
17     }
18
19     public voidsetDestination(Queue destination) {
20        this.destination = destination;
21     }
22
23     public voidsend(Map<String,Object> message) {
24         template.convertAndSend(this.destination, message);
25     }
26
27 }

生产Topic方式的消息

TopicMessageProducer.java

1 package com.uu.activemq;
2
3 import java.util.Map;
4
5 import javax.jms.Topic;
6
7 import org.springframework.jms.core.JmsTemplate;
8
9 public classTopicMessageProducer {
10    
11     private JmsTemplate template;
12
13     private Topic destination;
14
15     public voidsetTemplate(JmsTemplate template) {
16        this.template = template;
17     }
18
19     public voidsetDestination(Topic destination) {
20        this.destination = destination;
21     }
22
23     public voidsend(Map<String,Object> message) {
24        template.convertAndSend(this.destination,message);
25     }
26 }

写一个测试类

1 package com.uu.activemq;
2
3 import java.util.LinkedHashMap;
4 import java.util.Map;
5
6 importorg.springframework.context.ApplicationContext;
7 importorg.springframework.context.support.FileSystemXmlApplicationContext;
8
9 /** 10 * 解决耗时的数据操作 11 * 发送消息不等待返回，继续执行 12*/
13 public class MainTest
14 {
15     public static void main(String[] args)
16     {
17        ApplicationContext wac = newFileSystemXmlApplicationContext("classpath:spring.xml");
18        
19        TopicMessageProducer topicMessageProducer  = (TopicMessageProducer)wac.getBean("topicMessageProducer");
20        
21        QueueMessageProducer queueMessageProducer  = (QueueMessageProducer)wac.getBean("queueMessageProducer");
22        
23        Map<String, Object> message = newLinkedHashMap<String, Object>();
24        message.put("test", "ActiveMQ");
25        
26        queueMessageProducer.send(message);
27        topicMessageProducer.send(message);
28        System.out.println("完成");
29     }
30 }

结果先输出完成然后输出传递的参数

Spring整和ActiveMQ

博客分类：

·        spring

ActiveMQSpringBeanJMSApache

问题一：为什么要整合ActiveMQ？

传统的JDBC代码在处理连接、语句、结果集和异常时是多么冗长和繁杂你一定不会忘记，传统的JMS继承了JDBC的“关荣传统”。发送一个简单的消息，要几行代码呢？请仔细数数吧。

Java代码

1.      public void sendMessage() throws JMSException {

2.      ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(

3.      "tcp://localhost:61616");

4.      Connection connection = null;

5.      Session session=null;

6.      try {

7.

8.      connection = (Connection) connectionFactory.createConnection();//创建连接

9.      session = (Session) connection.createSession(false,

10.  Session.AUTO_ACKNOWLEDGE);//创建会话

11.  Destination destination = session.createQueue("myQueue");

12.  MessageProducer producer = session.createProducer(destination);

13.  TextMessage message = session.createTextMessage(expectedBody);

14.  message.setStringProperty("headname", "remoteB");

15.  producer.send(message);

16.  connection.close();

17.  } catch (Exception e) {

18.  e.printStackTrace();

19.  }finally{

20.  try {

21.  if(session!=null){

22.  session.close();

23.  }

24.  if(connection!=null){

25.  connection=null;

26.  }

27.  } catch (Exception e) {

28.

29.  }

30.  }

31.  }

传统接受消息而是类似的代码，其实我们的目的就是发送和接受消息。幸运的是Spring为我们提供了大量的模板。项目一期用的较多的是JdbCTemplate，spring也为我们提供了JMSTemplate模板。

问题二：JMSTemplate模板该如何配置呢？

类似于jdbcTemplate，首先要配置一个ConnectionFactory，我们采用ActiveMQ5.2作为消息服务器。之后要开始配置JmsTemplate模板了。最后是配置消息目标了。消息分为队列和主题两大类，因此要配置两个消息目标了。

Java代码

1.      <!-- ActiveMQ -->

2.

3.      <!--配置JMS连接工厂-->

4.      <bean id="JmsConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">

5.      <property name="brokerURL" value="tcp://localhost:61616"/>

6.      </bean>

7.      <!--配置JMS模版-->

8.      <bean id="JmsTemplate" class="org.springframework.jms.core.JmsTemplate">

9.      <property name="connectionFactory" ref="JmsConnectionFactory"/>

10.  </bean>

11.  <!--发送消息的目的地（队列）-->

12.  <bean id="QueueDestination" class="org.apache.activemq.command.ActiveMQQueue">

13.  <!--设置消息队列的名字-->

14.  <constructor-arg index="0" value="HelloWorldQueue"/>

15.  </bean>

16.  <!--发送消息的目的地（主题）-->

17.  <bean id="TopicDestination" class="org.apache.activemq.command.ActiveMQTopic">

18.  <!--设置消息主题的名字-->

19.  <constructor-arg index="0" value="FlexTopic"/>

20.  </bean>

问题三：如何使用JmsTemplate发送消息呢?

spring的beanfactory得到一个jmsTemplate的实例和消息目标的实例，发送消息，够简单的吧。看看代码：

Java代码

1.      JmsTemplate template = (JmsTemplate) SpringContext.getBean("JmsTemplate");

2.      ActiveMQTopic destination=(ActiveMQTopic)SpringContext.getBean("TopicDestination");

3.      template.send((javax.jms.Destination) destination, new MessageCreator(){

4.      public Message createMessage(Session session) throws JMSException {

5.      return session.createTextMessage("hello");

6.      }

7.

8.      });

问题四：上面的代码能不能在简单些？

很多时候，发送消息的目标都是默认的，因此是不是可以在jmsTemplate中设置一个默认的消息目标呢？答案是肯定的。

Java代码

1.      <!--配置JMS模版-->

2.      <bean id="JmsTemplate" class="org.springframework.jms.core.JmsTemplate">

3.      <property name="connectionFactory" ref="JmsConnectionFactory"/>

4.      <property name="defaultDestination" ref="TopicDestination"></property>

5.      </bean>

发送消息的时候，不指定目标，spring就会调用默认的目标了。

Java代码

1.      JmsTemplate template = (JmsTemplate) SpringContext.getBean("JmsTemplate");

2.      template.send( new MessageCreator(){

3.      public Message createMessage(Session session) throws JMSException {

4.      return session.createTextMessage("hello");

5.      }

6.      });

问题四：jmsTemplate怎么接受信息？

jmsTemplate接收消息十分的简单，只需要调用template.receive()方法，receive方法是同步的，默认情况下，对 receive()方法的调用会造成阻塞，知道消息到达目标----如果必要，永远等下去。为了避免对消息内容等待，可以配置jmsTemplate时，通过设置receiveTimeout属性来指定接收消息超时时间。下面的配置将接收消息的超时时间设置为一分钟（60000毫秒）。

Xml代码

1.      <!--配置JMS模版-->

2.      <bean id="JmsTemplate" class="org.springframework.jms.core.JmsTemplate">

3.      <property name="connectionFactory" ref="JmsConnectionFactory"/>

4.      <property name="defaultDestination" ref="TopicDestination"></property>

5.      <property name="receiveTimeout" value="60000"></property>

6.      </bean>

template.receive()会从默认目标接收消息，如果你希望指定一个目标，可以传一个目标。如：template.receive("myQueue").

同步接收消息并不是spring唯一的选择，消息监听器可以实现异步（下篇文章将会介绍消息驱动）。

问题五：怎么自动将消息转化为Java对象？

转化器在很多组件中都是必不缺少的东西。Spring挺过MessageConverter接口提供了对消息转换的支持。

Java代码

1.      public class MyMessageConverter implements MessageConverter {

2.      @Override

3.      public Object fromMessage(Message arg0) throws JMSException,

4.      MessageConversionException {

5.      // TODO Auto-generated method stub

6.      return null;

7.      }

8.      @Override

9.      public Message toMessage(Object arg0, Session arg1) throws JMSException,

10.  MessageConversionException {

11.  // TODO Auto-generated method stub

12.  return null;

13.  }

14.  }

MessageConverter接口的两个方法简单明了。在发送端toMessage会将java对象转化为消息，在接收端fromMessage会将消息转化为java对象。

下面的代码简单的实现了MessageConverter的两个接口。

Java代码

1.      public class MyMessageConverter implements MessageConverter {

2.      @Override

3.      public Object fromMessage(Message message) throws JMSException,

4.      MessageConversionException {

5.      if(!(message instanceof MapMessage)){

6.      throw new MessageConversionException("Messae is not MapMessage");

7.      }

8.      MapMessage mapMessage=(MapMessage)message;

9.      MessageObj messageObj=new MessageObj();

10.  messageObj.setId(mapMessage.getString("id"));

11.  messageObj.setInfo(mapMessage.getString("info"));

12.  return messageObj;

13.  }

14.  @Override

15.  public Message toMessage(Object obj, Session session) throws JMSException,

16.  MessageConversionException {

17.  if(!(obj instanceof MessageObj)){

18.  throw new MessageConversionException("obj is not MessageObj");

19.  }

20.  MessageObj messageObj=(MessageObj)obj;

21.  MapMessage mapMessage=session.createMapMessage();

22.  mapMessage.setString("id", messageObj.getId());

23.  mapMessage.setString("info", messageObj.getInfo());

24.  return mapMessage;

25.  }

26.  }

此时，发送和接收消息要换成template.convertAndSend(message);template.receiveAndConvert();

可是jmsTemplate如何知道消息转换器呢？需要在配置jmsTemplate的时候，加上messageConverter属性。

Java代码

1.      <!--配置JMS模版-->

2.      <bean id="JmsTemplate" class="org.springframework.jms.core.JmsTemplate">

3.      <property name="connectionFactory" ref="JmsConnectionFactory"/>

4.      <property name="defaultDestination" ref="TopicDestination"></property>

5.      <property name="receiveTimeout" value="60000"></property>

6.      <property name="messageConverter" ref="messageObj"></property>

7.      </bean>

博客分类：

·        spring

SpringBeanJavaquartzJMS

应用场景：每月10号，我们都会收到本月薪资的邮件；托管行会在每天早上发送可用头寸表。这些事情都不是用户动作引起的，而是系统根据任务调度规划引发的一些动作。

1 使用Java Timer调度任务

从jdk1.3开始，java.util.Timer类提供了基本的调度功能。这个类允许你调用一个任务(java.util.TimerTask子类定义)按任意周期运行。

Spring通过TimerFactoryBean向应用程序上下文提供对java Timer的支持。TimerFactoryBean是Spring的一个工厂类，用于在应用程序上下文(运行TimerTask的)中生成JavaTimer。下面展示了TimerFactoryBean是如何工作的。

创建一个定时任务

Java代码

1.      public class ScanFileTask extends TimerTask {

2.

3.      public void run() {

4.      System.out.println("开始扫描文件了");

5.      }

6.

7.      }

run()方法定义了当任务运行时该做什么。

在spring的配置文件中配置该bean。

Xml代码

1.      <bean id="scanFileTask" class="org.utmost.jms.ScanFileTask"/>

spring的 ScheduledTimerTask定义了定时器任务运行周期period。delay属性指定当任务第一次运行之前等待多长时间。timerTask告诉ScheduledTimerTask执行那个TimerTask。

Xml代码

1.      <bean id="testTask" class="org.springframework.scheduling.timer.ScheduledTimerTask">

2.      <property name="timerTask" ref="scanFileTask"></property>

3.      <property name="period" value="40000"/>

4.      <property name="delay" value="10000"/>

5.      </bean>

最后一步就是启动定时器了。

Xml代码

1.      <bean class="org.springframework.scheduling.timer.TimerFactoryBean">

2.      <property name="scheduledTimerTasks">

3.      <list>

4.      <ref bean="testTask"/>

5.      </list>

6.      </property>

7.      </bean>

scheduledTimerTasks是一个需要启动定时器的列表。

东子破解jms7月2号相关推荐

1. 10月28号安全焦点DDOS事件背后内幕串串烧

   文章整理:安全焦点论坛 作者:cloudx 发布时间:2006-11-09 主题:讲述安焦被DDOS背后的故事及部分安全知识!同时奉送ASP.NET备份db日志获得webshell小知识,小小伎俩,牛 ...

2. 10月28号安全焦点DDOS事件背后内幕串串烧～～

   主题:讲述安焦被DDOS背后的故事及部分安全知识!同时奉送ASP.NET备份db日志获得webshell小知识,小小伎俩,牛人莫笑话俺了. 本贴纯属娱乐大众!慰安下各位fans在安焦被D 9天以来无法 ...

3. 2017年6月16号课堂笔记

   2017年6月16号 星期五 空气质量:中度污染~轻度污染 内容:jQuery:remove,bind,attr,on和live,同辈和父辈节点的操作, keyup/keypress/keydown, ...

4. 2017年06月07号课堂笔记

   2017年06月07号 晴 空气质量:中度污染->良 内容:jQuery第二节:基本选择器,层次选择器,属性选择器, 基本过滤选择器,可见性过滤选择器 备注:代课老师:李老师 一.基本选择器 1 ...

5. 2017年4月21号课堂笔记

   2017年4月21号 星期五 空气质量:良 内容:XML,Dom4j解析XML文件,单例 一.XML 老师代码: <?xml version="1.0" encoding=& ...

6. 2017年07月03号课堂笔记

   2017年07月03号 星期一  多云 空气质量:轻度污染~中度污染 内容:MySQL第四节课 in和not in:两个表的内连接:exists和not exsits的使用:all,any和some: ...

7. 泰山行宫碧霞元君祠_临清市泰山行宫碧霞元君祠5月4号（农历三月三十日）举行大型泰山奶奶接驾法会...

   临清是泰山奶奶的娘家,每年的四月泰山奶奶要回临清省亲 临清市道教协会定于2019年农历三月三十(5月4号星期六)于临清桑树园泰山行宫碧霞元君祠举行大型泰山奶奶迎鸾接驾庙会.届时,将有架鼓会.云龙会.狮 ...

8. 2008年9月3号，星期三，晴。日日行，不怕千万里；常常做，不怕千万事。 ——《格言联璧•处事》

   2008年9月3号,星期三,晴. 日日行,不怕千万里:常常做,不怕千万事. --<格言联璧•处事> 今天是我博士生涯的第59天,昨天导师对我们那个项目整体篇中技术创新部分的PPT不很满意, ...

9. 2016年9月下旬校园招聘面经（美团、百度、58同城、华为、微店 11月10号更新）

   以下是近些天来的面试情况,大概就面了这几家,还挺顺利的,找工作太累了,缓了两天才坚持写完了,希望能帮助到近期奋斗在一线的找工作的朋友们,加油! 美团(后台开发工程师, offer) 很喜欢美团,希望有 ...

10. 7 月 24 号张小龙内部讲座《通过微信谈产品》

    7 月 24 号张小龙内部讲座<通过微信谈产品>有哪些亮点? 相关问题: 张小龙 2012 年 7 月 24 日的演讲,哪些值得称道或不足?  请选择一个修改理由...1. 不构成问题,需 ...

最新文章

1. CIO无法忽视的三大云角色
2. VS2012+EF6+Mysq
3. Linux驱动设计——字符杂项设备
4. 开头和结尾标记 在不同系统_写人作文的写作技巧。含思维导图、开头结尾写法及范文分析...
5. 这个女生躲在衣柜等男友回家，结果竟是......
6. 苹果要悄悄对这个产品动手了？你们最期盼的NFC功能也要来？
7. 苹果自动驾驶测试车近距离实拍，头顶新添传感器和硬件
8. linux下eclipse进行ndk调试，超简单，写的超清晰
9. android之wifi开发
10. python通过多进程实行多任务
11. eclipse基础实用教程
12. 协方差的计算公式例子_协方差公式_协方差的计算公式例子
13. win10位置定位服务器,win10定位设置在哪里进行设置|win10定位的设置方法详解
14. verilog符号 与或非 异或_与（）、非（~）、或（|）、异或（^）
15. VR全景图拍摄制作之无人机航拍
16. MySQL中文存到数据库是,springMVC保存数据到mysql数据库中文乱码问题解决方法
17. 塞班系统是如何没落的
18. GetFirmwareEnvironmentVariable失败返回错误码1314
19. Hadoop Steaming编程示例
20. 微信小程序被投诉怎么办？小妙招教给你

热门文章

1. mysql一张表建多个字段比较合适呢，答案来嘞
2. 【脚本小子狂喜】日常实用脚本推荐
3. 带你十分钟做出一个自己的手机APP
4. C++ 利用管道操作本机命令行命令
5. 在终端输入“python”或者”pip“时出现python’不是内部或外部命令，也不是可运行的程序 或批处理文件
6. C讲坛之猜数字游戏的实现
7. SQL Server认证培训与考试
8. shell脚本：备份数据库、代码上线
9. C语言:请自定义函数求两个数的和、差、积、商!
10. js去掉前后空格的函数_Node.js十年，你大爷还是你大爷