ActiveMQ的安装、运用

1、下载ActiveMQ：http://activemq.apache.org/

2、解压包，并找到activemq.bat,双击打开一个控制台，跟你打开tomcat一样的。如图：

3、在浏览器查看你的activemq是否正确启动：http://localhost:8161/admin,打开之后看到这样：  

用户名：admin  密码：admin

确定即可登录。

4、你可以新建一个Queue，在  Queue Name后面的方框中填入“MyFirstQueue”，点击正后方“Create”按钮即可。

在下面Queues：列表中，就会显示你刚刚新建的Queue：

这是已经使用过的，其实一开始数值是这样的：   

Name    ：      MyFirstQueue     

Number Of   ending Message  ： 0

Number Of   Consumers ：0

Message Enqueued：0

Message Dequeued：0

5、接下来我们通过IDEA来创建两个java类，一个是消息生产者，一个是消息消费者。

这是消息生产者：

package com.zfm.activemq;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Sender {
    private static final int SEND_NUMBER =5;
    public static void main(String[] args){
        //ConnectionFactory:连接工厂
        ConnectionFactory  connectionFactory;
        //Connection:JMS客户端到JMS Provider的连接
        Connection connection = null;
        //Session :一个会话，发送或接收消息的线程
        Session session;
        //Destination:消息的目的地
        Destination destination;
        //MessageProducer：消息产生者：发送
        MessageProducer messageProducer;
        //构造connectionFactory实例对象

        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://localhost:61616"/*ActiveMQ默认使用的TCP连接端口是61616,*/
        );

        try{
            //构造从工厂得到的连接
            connection = connectionFactory.createConnection();
            //启动
            connection.start();
            //获取操作连接
            session = connection.createSession(Boolean.TRUE,
                    Session.AUTO_ACKNOWLEDGE);
            //获取session的参数

            destination = session.createQueue("FMDemo");
            //得到消息生成者
            messageProducer = session.createProducer(destination);

            //设置·不持久化
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            //构造消息
            sendMeaasge(session,messageProducer);
            session.commit();


        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try{
                if(null!=connection){
                    connection.close();
                }
            }catch (Throwable t){

            }
        }
    }

    public static void sendMeaasge(Session session,MessageProducer producer) throws JMSException {
        for(int i=1;i<=SEND_NUMBER;i++){
            TextMessage  message = session.createTextMessage("ActiveMq 发送的消息" +i);
            //发送消息到目的地
            System.out.println("发送消息： ActiveMq发送消息：" + i);
            producer.send(message);
        }

    }
}

消息消费者：

package com.zfm.activemq;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Receiver {
    public static void main(String[] rags) throws JMSException {
        //Connection连接工厂
        ConnectionFactory connectionFactory;
        //Connection:JMS客户端到JMS provider
        Connection connection = null;
        //Session : 一个发送或接受消息的·会话
        Session session;
        //Destination:消息的目的地,消息发送给谁
        Destination destination;
        //消费者，消息接收者
        MessageConsumer messageConsumer;

        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://localhost:61616"//ActiveMQ默认使用的TCP连接端口是61616,
        );
        try{
            //从构造工厂里获得连接
            connection = connectionFactory.createConnection();
            //一定要启动
            connection.start();

            //从连接获得会话
            session = connection.createSession(Boolean.FALSE
            ,Session.AUTO_ACKNOWLEDGE);

            destination = session.createQueue("FMDemo");
            messageConsumer = session.createConsumer(destination);
            while(true){
                //设置接收者接受新消息的时间
                TextMessage message = (TextMessage)messageConsumer.receive(100000);
                if(null!=message){
                    System.out.println("收到消息" + message.getText());
                }else{
                    break;
                }
            }



        }catch (Exception e){

        }finally {
            try{
                if(null != connection) {
                    connection.close();
                }
            }catch(Throwable t){

            }
        }

    }
}

当运行Sender.java之后，刷新http://localhost:8161/admin/queues.jsp时，

Queues表中的参数有变化咯：

Name    ：     MyFirstQueue

Number Of   Pending Message  ：5

Number Of   Consumers ：0

Message  Enqueued ：5

Message  Dequeued    ：0

此时，由于消息已经发完，并且我们没有让Sender的main函数一直运行，所以，在控制台打印了：

发送消息： ActiveMq发送消息：1
发送消息： ActiveMq发送消息：2
发送消息： ActiveMq发送消息：3
发送消息： ActiveMq发送消息：4
发送消息： ActiveMq发送消息：5

Process finished with exit code 0

进程结束了。

这时我们再运行Receiver.java，而后控制台打印：

收到消息ActiveMq 发送的消息1
收到消息ActiveMq 发送的消息2
收到消息ActiveMq 发送的消息3
收到消息ActiveMq 发送的消息4
收到消息ActiveMq 发送的消息5

Process finished with exit code 0

此时，再刷新http://localhost:8161/admin/queues.jsp，此时表格数值变为：

Name    ：     MyFirstQueue

Number Of   Pending Message  ：0

Number Of   Consumers ：0

Message  Enqueued ：5

Message  Dequeued    ：5

还有一点是：当Receiver.java并没有运行结束的时候，由于我只开启了一个Receiver进程，所以这时Number Of   Consumers：1。

可见，由Sender发出的消息已经被Receiver收到了。你可以先把Sender消息发出来，这个时候只要消息已经到消息队列上了，只要你不期望发送者还要接收什么回复，你就可以把Sender停掉了。Receiver还是一样的接收到信息。

下面介绍一下ActiveMQ的几种基本通信方式：发布-订阅模式和点对点模式。

基础流程：

1、获得ActiveMQConnectionFactory。

2、利用factory获得Connection。

3、启动connection。

4、通过connection创建Session。

5、指定Session的Destination。

6、发送消息者则创建MessageProducer/接收消息者创建MessageConsumer。

7、发送和接收JMS Message。

8、关闭所有JMS资源。

ActiveMQ详情见http://shmilyaw-hotmail-com.iteye.com/blog/1897635

讲到ActiveMQ，其实他是java消息服务（JMS）的其中一种规范，JMS详情见：http://blog.csdn.net/jiuqiyuliang/article/details/46701559   和   http://blog.csdn.net/jiuqiyuliang/article/details/47160259。

什么情况下使用ActiveMQ?

1. 多个项目之间集成
   (1) 跨平台
   (2) 多语言
   (3) 多项目

2. 降低系统间模块的耦合度，解耦
   (1) 软件扩展性

3. 系统前后端隔离
   (1) 前后端隔离，屏蔽高安全区