spring 使用 ActiveMQ，JMS使用

 MS 定义了两种方式：Quere（点对点）；Topic（发布/订阅）。

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

Connection 负责创建 Session。

Session 创建 MessageProducer（用来发消息） 和 MessageConsumer（用来接收消息）。

Destination 是消息的目的地。

 

下载 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。

 

一  ActiveMQ的jms写法 

  1  消息发送类  

    

Java代码  


public class JMSsenderBean {  

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

        //连接工厂。连接工厂（ConnectionFactory）是由管理员创建，并绑定到JNDI树中。客户端使用JNDI查找连接工厂，然后利用连接工厂创建一个JMS连接。  

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");    

        // JMS连接。JMS连接（Connection）表示JMS客户端和服务器端之间的一个活动的连接，是由客户端通过调用连接工厂的方法建立的    

        Connection connection = connectionFactory.createConnection();  

        //开启连接  

        connection.start();    

        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  

        Destination destination = session.createQueue("my-queue");    

        MessageProducer producer = session.createProducer(destination);    

        for(int i=0; i<3; i++) {    

            MapMessage message = session.createMapMessage();    

            message.setLong("count", new Date().getTime());    

            Thread.sleep(1000);  

            //通过消息生产者发出消息    

            producer.send(message);    

        }    

        session.commit();    

        session.close();    

        connection.close();    

    }    

 }  

 

  2 消息接收类

  

Java代码  


import javax.jms.Connection;  

import javax.jms.ConnectionFactory;  

import javax.jms.Destination;  

import javax.jms.MapMessage;  

import javax.jms.MessageConsumer;  

import javax.jms.Session;  

  

import org.apache.activemq.ActiveMQConnectionFactory;  

//消息接收类   

public class JMSreceive {  

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

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
17ebb
"tcp://localhost:61616");    

        

        Connection connection = connectionFactory.createConnection();    

        connection.start();    

        

        final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);    

        Destination destination = session.createQueue("my-queue");    

        MessageConsumer consumer = session.createConsumer(destination);    

        /*//listener 方式  

        consumer.setMessageListener(new MessageListener() {  

      

            public void onMessage(Message msg) {  

                MapMessage message = (MapMessage) msg;  

                //TODO something....  

                System.out.println("收到消息：" + new Date(message.getLong("count")));  

                session.commit();  

            }  

        });  

        Thread.sleep(30000);  

        */    

        int i=0;    

        while(i<3) {    

            i++;    

            MapMessage message = (MapMessage) consumer.receive();    

            session.commit();    

            System.out.println("收到消息：" + new Date(message.getLong("count")));    

        }    

        

        session.close();    

        connection.close();    

    }    

}  

 

 

二 spring集成

  

   1 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"    

        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">    

    

<!-- 在非 web / ejb 容器中使用 pool 时，要手动 stop，spring 不会为你执行 destroy-method 的方法    

    <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">    

        <property name="connectionFactory">    

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

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

            </bean>    

        </property>    

    </bean>    

-->    

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

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

    </bean>   

  

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

        <constructor-arg index="0" value="my-queue" />    

    </bean>    

      

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

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

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

        <property name="messageConverter">    

            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />    

        </property>    

    </bean>    

</beans>   

 

 2   消息发送类

Java代码  


import java.util.Date;  

  

import javax.jms.JMSException;  

import javax.jms.MapMessage;  

import javax.jms.Session;  

import javax.mail.Message;  

  

import org.springframework.context.ApplicationContext;  

import org.springframework.context.support.FileSystemXmlApplicationContext;  

import org.springframework.jms.core.JmsTemplate;  

import org.springframework.jms.core.MessageCreator;  

  

public class JMSsenderBean_Spring {  

    public static void main(String[] args) {    

        ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:spring/applicationContext-jmx.xml");    

        

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

        

        jmsTemplate.send(new MessageCreator() {    

        

            public MapMessage createMessage(Session session) throws JMSException {   

                  

                MapMessage mm = session.createMapMessage();    

                mm.setLong("count", new Date().getTime());  

                  

                return mm;    

            }    

        

        });    

    }     

 }  

 3  消息接收类

  

Java代码  


import java.util.Date;  

import java.util.Map;  

  

import org.springframework.context.ApplicationContext;  

import org.springframework.context.support.FileSystemXmlApplicationContext;  

import org.springframework.jms.core.JmsTemplate;  

  

public class JMSreceive_Spring {  

    public static void main(String[] args) {    

        ApplicationContext ctx = new FileSystemXmlApplicationContext("classpath:spring/applicationContext-jmx.xml");    

        

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

        while(true) {    

            Map<String, Object> mm =  (Map<String, Object>) jmsTemplate.receiveAndConvert();    

            System.out.println("收到消息：" + new Date((Long)mm.get("count")));  

              

        }    

    }   

}