使用 ActiveMQ 示例

红薯 发布于 2011/05/08 07:07
阅读 5K+
收藏 58

企业中各项目中相互协作的时候可能用得到消息通知机制。比如有东西更新了,可以通知做索引。

在 Java 里有 JMS 的多个实现。其中 apache 下的 ActiveMQ 就是不错的选择。还有一个比较热的是 RabbitMQ (是 erlang 语言实现的)。这里示例下使用 ActiveMQ

用 ActiveMQ 最好还是了解下 JMS

JMS 公共 点对点域 发布/订阅域
ConnectionFactory QueueConnectionFactory TopicConnectionFactory
Connection QueueConnection TopicConnection
Destination Queue Topic
Session QueueSession TopicSession
MessageProducer QueueSender TopicPublisher
MessageConsumer QueueReceiver TopicSubscriber

JMS 定义了两种方式:Quere(点对点);Topic(发布/订阅)。

ConnectionFactory 是连接工厂,负责创建Connection。

Connection 负责创建 Session。

Session 创建 MessageProducer(用来发消息) 和 MessageConsumer(用来接收消息)。

Destination 是消息的目的地。

详细的可以网上找些 JMS 规范(有中文版)。

下载 apache-activemq-5.3.0。http://activemq.apache.org/download.html,解压,然后双击 bin/activemq.bat。运行后,可以在 http://localhost:8161/admin 观察。也有 demo, http://localhost:8161/demo。把 activemq-all-5.3.0.jar 加入 classpath。

Jms 发送 代码:

  1. public static void main(String[] args) throws Exception {  
  2.     ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();  
  3.   
  4.     Connection connection = connectionFactory.createConnection();  
  5.     connection.start();  
  6.   
  7.     Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  
  8.     Destination destination = session.createQueue("my-queue");  
  9.   
  10.     MessageProducer producer = session.createProducer(destination);  
  11.     for(int i=0; i<3; i++) {  
  12.         MapMessage message = session.createMapMessage();  
  13.         message.setLong("count"new Date().getTime());  
  14.         Thread.sleep(1000);  
  15.         //通过消息生产者发出消息  
  16.         producer.send(message);  
  17.     }  
  18.     session.commit();  
  19.     session.close();  
  20.     connection.close();  
  21. }  

Jms 接收代码:

  1. public static void main(String[] args) throws Exception {  
  2.     ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();  
  3.   
  4.     Connection connection = connectionFactory.createConnection();  
  5.     connection.start();  
  6.   
  7.     final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  
  8.     Destination destination = session.createQueue("my-queue");  
  9.   
  10.     MessageConsumer consumer = session.createConsumer(destination);  
  11.     /*//listener 方式 
  12.     consumer.setMessageListener(new MessageListener() { 
  13.  
  14.         public void onMessage(Message msg) { 
  15.             MapMessage message = (MapMessage) msg; 
  16.             //TODO something.... 
  17.             System.out.println("收到消息:" + new Date(message.getLong("count"))); 
  18.             session.commit(); 
  19.         } 
  20.  
  21.     }); 
  22.     Thread.sleep(30000); 
  23.     */  
  24.     int i=0;  
  25.     while(i<3) {  
  26.         i++;  
  27.         MapMessage message = (MapMessage) consumer.receive();  
  28.         session.commit();  
  29.   
  30.         //TODO something....  
  31.         System.out.println("收到消息:" + new Date(message.getLong("count")));  
  32.     }  
  33.   
  34.     session.close();  
  35.     connection.close();  
  36. }  

启动 JmsReceiver 和 JmsSender 可以在看输出三条时间信息。当然 Jms 还指定有其它格式的数据,如 TextMessage

结合 Spring 的 JmsTemplate 方便用:

xml:

  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.         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">  
  4.   
  5. <!-- 在非 web / ejb 容器中使用 pool 时,要手动 stop,spring 不会为你执行 destroy-method 的方法  
  6.     <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  
  7.         <property name="connectionFactory">  
  8.             <bean class="org.apache.activemq.ActiveMQConnectionFactory">  
  9.                 <property name="brokerURL" value="tcp://localhost:61616" />  
  10.             </bean>  
  11.         </property>  
  12.     </bean>  
  13. -->  
  14.     <bean id="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
  15.         <property name="brokerURL" value="tcp://localhost:61616" />  
  16.     </bean>  
  17.     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
  18.         <property name="connectionFactory" ref="jmsFactory" />  
  19.         <property name="defaultDestination" ref="destination" />  
  20.         <property name="messageConverter">  
  21.             <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />  
  22.         </property>  
  23.     </bean>  
  24.   
  25.     <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">  
  26.         <constructor-arg index="0" value="my-queue" />  
  27.     </bean>  
  28.   
  29. </beans>  

sender:

  1. public static void main(String[] args) {  
  2.     ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:app*.xml");  
  3.   
  4.     JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");  
  5.   
  6.     jmsTemplate.send(new MessageCreator() {  
  7.   
  8.         public Message createMessage(Session session) throws JMSException {  
  9.             MapMessage mm = session.createMapMessage();  
  10.             mm.setLong("count"new Date().getTime());  
  11.             return mm;  
  12.         }  
  13.   
  14.     });  
  15. }  

receiver:

  1. public static void main(String[] args) {  
  2.     ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:app*.xml");  
  3.   
  4.     JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");  
  5.     while(true) {  
  6.         Map<String, Object> mm =  (Map<String, Object>) jmsTemplate.receiveAndConvert();  
  7.         System.out.println("收到消息:" + new Date((Long)mm.get("count")));  
  8.     }  
  9. }  

注意:直接用 Jms 接口时接收了消息后要提交一下,否则下次启动接收者时还可以收到旧数据。有了 JmsTemplate 就不用自己提交 session.commit() 了。如果使用了 PooledConnectionFactory 要把 apache-activemq-5.3.0\lib\optional\activemq-pool-5.3.0.jar 加到 classpath

加载中
0
黑马王子
黑马王子

不错,学习下。

0
xmdeepdata
xmdeepdata

不知道有没人在生产环境用过,说说经验。

0
陌上草
@红薯大哥,,oschina里面的问题怎么收藏啊??点击收藏后在自己的空间里看不到啊~~
0
uemc
uemc

为什么我这只能创建和接收queue 类型的消息,而对于topic类型的消息只能发送,接收的时候

messageConsumer.receive(0);

卡在这句代码不动了。

0
uemc
uemc

@红薯 老大,能补充下topic 的例子吗!我做的JMS 只能发送接收队列消息,能发送主题消息,但不能接收主题消息。而且不能加接收的消息监听。

http://www.oschina.net/question/168245_45256

 

求解!

0
依然范特西
依然范特西

很不错的入门文章,希望老大继续分享MQ方面的文章。

0
weiweihong
weiweihong

@红薯  老大, 我想在加载spring的时候就启动ActiveMQ,我在xml 头文件里面加上xmlns:amq="http://activemq.apache.org/schema/core"

xsi:schemaLocation="http://activemq.apache.org/schema/core    
          http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd">

<amq:broker useJmx="false" persistent="false">
  <amq:transportConnectors>
   <amq:transportConnector uri="tcp://localhost:61616"></amq:transportConnector>
  </amq:transportConnectors>
 </amq:broker>

jar包有的,但报错,Configuration problem: Unable to locate NamespaceHandler for namespace [http://activemq.apache.org/schema/core]  认不到这个,, 我也在eclipse--window--preferences--xml catalog 里面add了activemq的namespace和schema.. 这是怎么回事呢?

0
像風一样自由
像風一样自由

不错正在学习中

0
砍材农夫
砍材农夫
不知道 配置数据库 怎么配置
smith789
smith789
在activemq.xml里配置下persistenceAdapter,然后再数据库持久化的bean,具体的格式忘了,可以百度下。当然前提数据库实例配置正确并已启动。
0
夏夜流星
夏夜流星
我想问一下红薯大大,如果想让onMessage中接收消息过后就去执行一个5分钟左右的任务,此时再来一个消息,如何处理这两个消息的并行处理呢?我这边onMessage好像是串行处理的!  
返回顶部
顶部