1、ActiveMq安装启动

1、从官网http://activemq.apache.org/download.html下载对应的版本，并解压
2、设置activeMQ配置环境.activemqrc ：bin/activemq setup ~/.activemqrc
3、到bin目录下执行启动脚本：bin/activemq start
4、查看端口号是否监听：netstat -an|grep 61616
5、访问activeMQ管理界面：如果是本机就是http://localhost:8161/admin

2、spring整合jms--基于ActiveMQ

1、增加依赖(gradle版本)

apply plugin: 'java'
apply plugin: 'jetty'
apply plugin: 'idea'
apply plugin: 'eclipse'

sourceSets {
    main {
        resources.srcDirs += 'src/main/webapp'
    }
    test {
        resources.srcDirs += 'src/test/webapp'
    }
}
repositories {
    mavenRepo urls: ["http://maven.lujs.cn/nexus/content/groups/public/"]
    mavenCentral()
    mavenLocal()
}

dependencies {
    spring_version = "3.0.5.RELEASE"
    spring = ["org.springframework:spring-core:$spring_version",
              "org.springframework:spring-expression:$spring_version",
              "org.springframework:spring-beans:$spring_version",
              "org.springframework:spring-aop:$spring_version",
              "org.springframework:spring-context:$spring_version",
              "org.springframework:spring-context-support:$spring_version",
              "org.springframework:spring-tx:$spring_version",
              "org.springframework:spring-orm:$spring_version",
              "org.springframework:spring-web:$spring_version",
              "org.springframework:spring-asm:$spring_version",
              "org.springframework:spring-jdbc:$spring_version",
              "org.springframework:spring-webmvc:$spring_version",
              "org.springframework:spring-test:$spring_version",
              "org.springframework:spring-jms:$spring_version",
              "aopalliance:aopalliance:1.0"]

    apache = ["org.apache.velocity:velocity:1.7",
              "org.apache.velocity:velocity-tools:2.0",
              "commons-logging:commons-logging:1.1.1",
              "commons-io:commons-io:2.0.1",
              "commons-codec:commons-codec:1.5"]

    activemq_version = "5.6.0"
    activemq = ["org.apache.activemq:activemq-web:$activemq_version",
                "org.apache.activemq:activemq-camel:$activemq_version",
                "org.apache.activemq:activemq-pool:$activemq_version",
                "org.apache.activemq:activemq-all:$activemq_version",
                "org.apache.activemq:activemq-core:$activemq_version",
                "org.apache.activemq:activemq-jaas:$activemq_version",
                "org.apache.activemq:activeio-core:3.1.4",
                "dom4j:dom4j:1.6.1",
                "commons-pool:commons-pool:1.5.6",
                "org.apache.geronimo.specs:geronimo-j2ee-management_1.1_spec:1.0.1",
                "org.apache.geronimo.specs:geronimo-jms_1.1_spec:1.1.1"
    ]

    junit = [
            "junit:junit:4.10"
    ]

    gson = ["com.google.code.gson:gson:1.7.2"]
    dubbo_version = "2.0.13"
    dubbo = ["com.alibaba:dubbo:$dubbo_version"]
    zookeeper = ["org.apache.zookeeper:zookeeper:3.4.9"]
    log4j = ["log4j:log4j:1.2.16"]
    guava = ["com.google.guava:guava:18.0"]
    apollo = ["com.ctrip.framework.apollo:apollo-client:0.9.1-SNAPSHOT"]
    netty = ["io.netty:netty-all:4.1.8.Final"]
    jsr = ["javax.annotation:jsr250-api:1.0"]
    compile spring, gson, dubbo, zookeeper, log4j, guava, apollo, netty, apache, activemq, junit
    testCompile group: 'junit', name: 'junit', version: '4.12'
}

task "create-dirs" << {
    sourceSets*.java.srcDirs*.each { it.mkdirs() }
    sourceSets*.resources.srcDirs*.each { it.mkdirs() }
    file("release").mkdir()
    file("src/main/webapp/WEB-INF").mkdirs()
}

task wrapper(type: Wrapper) {
    outputs.upToDateWhen { false }
    distributionUrl = 'http://lujs.cn/lts/static/software/gradle-1.0-milestone-4.zip'
}
stopKey = 'stop-jetty'
stopPort = 8881
httpPort = 8880

2、配置ConnectionFactory
ConnectionFactory是用于产生到JMS服务器的链接的，Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory。SingleConnectionFactory对于建立JMS服务器链接的请求会一直返回同一个链接，并且会忽略Connection的close方法调用。CachingConnectionFactory继承了SingleConnectionFactory，所以它拥有SingleConnectionFactory的所有功能，同时它还新增了缓存功能，它可以缓存Session、MessageProducer和MessageConsumer。除此之外， ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。
Spring提供的ConnectionFactory只是Spring用于管理ConnectionFactory的，真正产生到JMS服务器链接的ConnectionFactory还得是由JMS服务厂商提供，并且需要把它注入到Spring提供的ConnectionFactory中。我们这里使用的是ActiveMQ实现的JMS，所以在我们这里真正的可以产生Connection的就应该是由ActiveMQ提供的ConnectionFactory。这里MQ配置的是本地，实例如下：

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <constructor-arg value="tcp://localhost:61616"></constructor-arg>
    </bean>
    <!--<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">-->
        <!--<property name="connectionFactory" ref="connectionFactory"/>-->
    <!--</bean>-->
    <bean id="singleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="connectionFactory"></property>
    </bean>

3、配置生产者
生产者负责产生消息并发送到JMS服务器，这通常对应的是我们的一个业务逻辑服务实现类。我们可以利用Spring为我们提供的JmsTemplate类来实现的，使用JmsTemplate进行消息发送时没有指定destination的时候将使用默认的Destination。默认Destination可以通过在定义jmsTemplate bean对象时通过属性defaultDestination或defaultDestinationName来进行注入，defaultDestinationName对应的就是一个普通字符串。另一种方式是使用jms提供的messageProducer实现，在ActiveMQ中实现了两种类型的Destination，一个是点对点的ActiveMQQueue，另一个就是支持订阅/发布模式的ActiveMQTopic，本实例使用topic方式。

<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg>
            <value>test.topic1</value>
        </constructor-arg>
    </bean>

    <bean id="mqMessageProducer" class="com.test.activeMq.MqMessageProducer">
        <property name="singleConnectionFactory" ref="singleConnectionFactory"/>
    </bean>

producer实现类

public class MqMessageProducer {
//    private PooledConnectionFactory pooledConnectionFactory;
//
//    public PooledConnectionFactory getPooledConnectionFactory() {
//        return pooledConnectionFactory;
//    }
//
//    public void setPooledConnectionFactory(PooledConnectionFactory pooledConnectionFactory) {
//        this.pooledConnectionFactory = pooledConnectionFactory;
//    }

    private SingleConnectionFactory singleConnectionFactory;

    public SingleConnectionFactory getSingleConnectionFactory() {
        return singleConnectionFactory;
    }

    public void setSingleConnectionFactory(SingleConnectionFactory singleConnectionFactory) {
        this.singleConnectionFactory = singleConnectionFactory;
    }

    public void sendMessage(Destination destination, String message){
//        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        try {
            Connection connection = singleConnectionFactory.createConnection();
            Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
//            Destination destination = session.createTopic("test.topic1");
            MessageProducer producer = session.createProducer(destination);
            TextMessage msg = session.createTextMessage();
            msg.setText(message);
            producer.send(msg);
            System.out.println("往目的地"+destination.toString()+"发送消息成功："+ message);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

这里我们使用MessageProducer来发送消息。
3、配置消费者
生产者往指定目的地Destination发送消息后，接下来就是消费者对指定目的地的消息进行消费了。Spring为我们封装的消息监听容器MessageListenerContainer它负责接收信息，并把接收到的信息分发给真正的MessageListener进行处理。每个消费者对应每个目的地都需要有对应的MessageListenerContainer，另外，MessageListenerContainer还需要知道去监听哪个JMS服务器，这是通过在配置MessageConnectionFactory的时候往里面注入一个ConnectionFactory来实现的，MessageListenerContainer有三个属性必须指定，一个是表示从哪里监听的ConnectionFactory；一个是表示监听什么的Destination；一个是接收到消息以后进行消息处理的MessageListener。Spring一共为我们提供了两种类型的MessageListenerContainer，SimpleMessageListenerContainer和DefaultMessageListenerContainer。
SimpleMessageListenerContainer会在一开始的时候就创建一个会话session和消费者Consumer，并且会使用标准的JMS MessageConsumer.setMessageListener()方法注册监听器让JMS提供者调用监听器的回调函数。它不会动态的适应运行时需要和参与外部的事务管理。兼容性方面，它非常接近于独立的JMS规范，但一般不兼容Java EE的JMS限制。
大多数情况下我们还是使用的DefaultMessageListenerContainer，跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer会动态的适应运行时需要，并且能够参与外部的事务管理。它很好的平衡了对JMS提供者要求低、先进功能如事务参与和兼容Java EE环境。
定义处理消息的MessageListener
要定义处理消息的MessageListener我们只需要实现JMS规范中的MessageListener接口就可以了。MessageListener接口中只有一个方法onMessage方法，当接收到消息的时候会自动调用该方法。

public class ConsumerMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        try {
            System.out.println("接受到一条消息");
           //todo可以做一些复杂业务
            String receive = ((TextMessage)message).getText();
            System.out.println("接受到的消息是："+receive);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

spring配置如下：

<bean id="consumerMessageListener" class="com.test.activeMq.ConsumerMessageListener"/>
    <bean id="springContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="messageListener" ref="consumerMessageListener"/>
        <property name="connectionFactory" ref="singleConnectionFactory" />
        <property name="destination" ref="queueDestination" />
    </bean>

测试代码如下：

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:/WEB-INF/bean.xml")
public class Test {
    @Autowired
    private MqMessageProducer mqMessageProducer;

    @Autowired
    private MessageListenerAdapterProducer messageListenerAdapterProducer;

    @Autowired
    @Qualifier("sessionAwareQueue")
    private Destination sessionQueueDestination;

    @Autowired
    private Destination adapterQueue;

    @Autowired
    private Destination queueDestination;

//    @Autowired
//    private DefaultMessageListenerContainer springContainer;springContainer

//    @org.junit.Test
//    public void testSend(){
//        mqMessageProducer.sendMessage(queueDestination,"测试用例");
//    }

    @org.junit.Test
    public void testMessageListenerAdapter(){
        //mqMessageProducer.sendMessage(adapterQueue,"测试MessageListenerAdapter");
        messageListenerAdapterProducer.sendMessage(adapterQueue,"测试MessageListenerAdapter");
    }
}

此外，在Spring整合JMS的应用中我们在定义消息监听器的时候一共可以定义三种类型的消息监听器，分别是MessageListener、SessionAwareMessageListener和MessageListenerAdapter。MessageListener上面例子已经介绍。
SessionAwareMessageListener是Spring为我们提供的，它不是标准的JMS MessageListener。MessageListener的设计只是纯粹用来接收消息的，假如我们在使用MessageListener处理接收到的消息时我们需要发送一个消息通知对方我们已经收到这个消息了，那么这个时候我们就需要在代码里面去重新获取一个Connection或Session。SessionAwareMessageListener的设计就是为了方便我们在接收到消息后发送一个回复的消息，它同样为我们提供了一个处理接收到的消息的onMessage方法，但是这个方法可以同时接收两个参数，一个是表示当前接收到的消息Message，另一个就是可以用来发送消息的Session对象。例子：

public class ConsumerSessionAwareMessageListener implements SessionAwareMessageListener {

    private Destination destination;

    public Destination getDestination() {
        return destination;
    }

    public void setDestination(Destination destination) {
        this.destination = destination;
    }

    @Override
    public void onMessage(Message message, Session session) throws JMSException {
        System.out.println("收到一条消息");
        System.out.println("消息内容是："+((TextMessage)message).getText());
        MessageProducer producer = session.createProducer(destination);
        Message textMessage = session.createTextMessage("ConsumerSessionAwareMessageListener....");
        producer.send(textMessage);
    }
}

spring配置：

<bean id="sessionAwareQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>sessionAwareQueue</value>
        </constructor-arg>
    </bean>
    <bean id="consumerSessionAwareMessageListener" class="com.lufax.activeMq.ConsumerSessionAwareMessageListener">
        <property name="destination" ref="queueDestination"/>
    </bean>
    <bean id="sessionAwareListenerContainer"
          class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="singleConnectionFactory" />
        <property name="destination" ref="sessionAwareQueue" />
        <property name="messageListener" ref="consumerSessionAwareMessageListener" />
    </bean>

MessageListenerAdapter类实现了MessageListener接口和SessionAwareMessageListener接口，它的主要作用是将接收到的消息进行类型转换，然后通过反射的形式把它交给一个普通的Java类进行处理。
MessageListenerAdapter会把接收到的消息做如下转换：
- TextMessage转换为String对象；
- BytesMessage转换为byte数组；
- MapMessage转换为Map对象；
- ObjectMessage转换为对应的Serializable对象。
既然前面说了MessageListenerAdapter会把接收到的消息做一个类型转换，然后利用反射把它交给真正的目标处理器——一个普通的Java类进行处理（如果真正的目标处理器是一个MessageListener或者是一个SessionAwareMessageListener，那么Spring将直接使用接收到的Message对象作为参数调用它们的onMessage方法，而不会再利用反射去进行调用），那么我们在定义一个MessageListenerAdapter的时候就需要为它指定这样一个目标类。这个目标类我们可以通过MessageListenerAdapter的构造方法参数指定，也可以通过它的delegate属性来指定，如果指定的目标处理器是一个普通的Java类时Spring将利用Message进行了类型转换之后的对象作为参数通过反射去调用真正的目标处理器的处理方法，MessageListenerAdapter的defaultListenerMethod属性来决定真正的目标处理器的处理方法，当我们没有指定该属性时，Spring会默认调用目标处理器的handleMessage方法如：

<bean id="messageListenerAdapter" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
        <property name="delegate">
            <bean class="com.test.activeMq.ConsumerMessageListenerAdapter"/>
        </property>
        <!--<property name="defaultListenerMethod" value="receiveMessage"/>-->
    </bean>
<bean id="adapterQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>adapterQueue</value>
        </constructor-arg>
    </bean>
    <bean id="messageListenerAdapterContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="singleConnectionFactory"/>
        <property name="destination" ref="adapterQueue"/>
        <property name="messageListener" ref="messageListenerAdapter"/><!-- 使用MessageListenerAdapter来作为消息监听器 -->
    </bean>

public class ConsumerMessageListenerAdapter {

    public String handleMessage(String message){
        System.out.println("通过handleMessage接收到纯文本消息，消息内容是："+message);
        return "这是ConsumerMessageListenerAdapter对象handleMessage方法的返回值。";
    }

    public void receiveMessage(String message){
        System.out.println("通过receiveMessage接受到纯文本消息，消息内容是"+message);
    }
}

MessageListenerAdapter除了会自动的把一个普通Java类当做MessageListener来处理接收到的消息之外，其另外一个主要的功能是可以自动的发送返回消息。
当我们用于处理接收到的消息的方法的返回值不为空的时候，Spring会自动将它封装为一个JMS Message，然后自动进行回复。那么这个时候这个回复消息将发送到哪里呢？这主要有两种方式可以指定。
第一，可以通过发送的Message的setJMSReplyTo方法指定该消息对应的回复消息的目的地。这里我们把我们的生产者发送消息的代码做一下修改，在发送消息之前先指定该消息对应的回复目的地为一个叫responseQueue的队列目的地，具体代码如下所示：

public class MessageListenerAdapterProducer {

    @Autowired
    private Destination responseQueue;

    private SingleConnectionFactory singleConnectionFactory;

    public SingleConnectionFactory getSingleConnectionFactory() {
        return singleConnectionFactory;
    }

    public void setSingleConnectionFactory(SingleConnectionFactory singleConnectionFactory) {
        this.singleConnectionFactory = singleConnectionFactory;
    }

    public void sendMessage(Destination destination, final String message){
        System.out.println("---------------生产者发送消息-----------------");
        System.out.println("---------------生产者发了一个消息：" + message);
        Connection connection = null;
        try {
            connection = singleConnectionFactory.createConnection();
            Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
//            Destination destination = session.createTopic("test.topic1");
            MessageProducer producer = session.createProducer(destination);
            TextMessage msg = session.createTextMessage();
            msg.setJMSReplyTo(responseQueue);
            msg.setText(message);
            producer.send(msg);
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

接着定义一个叫responseQueue的队列目的地及其对应的消息监听器和监听容器。

<bean id="responseQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>responseQueue</value>
        </constructor-arg>
    </bean>

    <bean id="responseQueueListener" class="com.test.activeMq.ResponseQueueListener"></bean>
    <bean id="responseQueueMessageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="singleConnectionFactory"/>
        <property name="destination" ref="responseQueue"/>
        <property name="messageListener" ref="responseQueueListener"/>
    </bean>
    <bean id="messageListenerAdapterProducer" class="com.test.activeMq.MessageListenerAdapterProducer">
        <property name="singleConnectionFactory" ref="singleConnectionFactory"/>
    </bean>

responseQueueListener定义如下：

public class ResponseQueueListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            try {
                System.out.println("接收到发送到responseQueue的一个文本消息，内容是：" + textMessage.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

上面ConsumerMessageListenerAdapter中的handleMessage有一个非空的返回值，这样我们王adapterQueue发送一个消息的时候，
messageListenerAdapterContainer 监听到消息之后ConsumerMessageListenerAdapter的handleMessage方法处理消息，并往responseQueue发送一个消息，消息内容就是handleMessage方法返回的内容，最后responseQueueMessageListenerContainer会监听到消息，触发responseQueueListener处理消息，打印内容：
接收到发送到responseQueue的一个文本消息，内容是：XXXXX