MessageQueue二之ActiveMQ

介绍

它是一个消息中间件，支持多语言协议，使用非常简单，另外它集成了管理平台，对高可用和持久化的支持也非常好。

部署

Replicated LevelDB Store（热备）:http://blog.csdn.net/eric520zenobia/article/details/52370609

Replicated LevelDB Store + BROKER CLUSTER(集群):http://www.cnblogs.com/guozhen/p/6209973.html

下文中的示例都是单机部署，只需要从http://activemq.apache.org/下载，解压后执行对应版本的activemq.bat脚本即可

消息模式

点对点

每个消息只有一个消费者（一旦被消费，消息就从队列中移除）
生产者与消费者没有时间依赖（不管消费者是否运行，都不影响生产者把消息发送到队列）
消费者在消息接收成功后需要响应消息队列

发布订阅

每个消息可以有多个消费者
生产者与消费者有时间依赖（如果订阅者不在线，此时发布者发布的消息，订阅者都将丢失）

JAVA使用示例

以下示例需要用到解压目录中的jar包：activemq-all-5.14.5.jar

HelloWord

发送消息：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Send {
    //连接账号
    private String userName = "";
    //连接密码
    private String password = "";
    //连接地址
    private String brokerURL = "tcp://192.168.0.130:61616";
    //connection的工厂
    private ConnectionFactory factory;
    //连接对象
    private Connection connection;
    //一个操作会话
    private Session session;
    //目的地，其实就是连接到哪个队列，如果是点对点，那么它的实现是Queue，如果是订阅模式，那它的实现是Topic
    private Destination destination;
    //生产者，就是产生数据的对象
    private MessageProducer producer;
    
    public static void main(String[] args) {
        PTPSend send = new PTPSend();
        send.start();
    }
    
    public void start(){
        try {
            //根据用户名，密码，url创建一个连接工厂
            factory = new ActiveMQConnectionFactory(userName, password, brokerURL);
            
            //从工厂中获取一个连接
            connection = factory.createConnection();
            
            //测试过这个步骤不写也是可以的，但是网上的各个文档都写了
            connection.start();
            
            //创建一个session
            //第一个参数: 为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
            //第二个参数：为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
            //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
            //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。（要写在接收端里面，不是写在发送端的）
            //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            
            //创建一个到达的目的地，其实想一下就知道了，activemq不可能同时只能跑一个队列吧，这里就是连接了一个名为"text-msg"的队列，这个会话将会到这个队列，当然，如果这个队列不存在，将会被创建
            destination = session.createQueue("HelloWord");
            
            //点对点与订阅模式唯一不同的地方，就是这一行代码，点对点创建的是Queue，而订阅模式创建的是Topic
            //destination = session.createTopic("topic");

            
            //从session中，获取一个消息生产者
            producer = session.createProducer(destination);
            
            //设置生产者的模式，有两种可选
            //DeliveryMode.PERSISTENT 当activemq关闭的时候，队列数据将会被保存
            //DeliveryMode.NON_PERSISTENT 当activemq关闭的时候，队列里面的数据将会被清空
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            
            //创建一条消息，当然，消息的类型有很多，如文字，字节，对象等,可以通过session.create..方法来创建出来
            TextMessage textMsg = session.createTextMessage("你好，activeMQ！");
            //纯字符串的数据
            //session.createTextMessage();
            //序列化的对象
            //session.createObjectMessage();
            //流，可以用来传递文件等
            //session.createStreamMessage();
            //用来传递字节
            //session.createBytesMessage();
            //这个方法创建出来的就是一个map，可以把它当作map来用，当你看了它的一些方法，你就懂了
            //session.createMapMessage();
            
            for(int i = 0 ; i < 10 ; i ++){
                //设置该消息的超时时间
                //producer.setTimeToLive(i * 1000);
                
                //发送消息
                producer.send(textMsg);
            }
            
            System.out.println("发送消息成功");
            
            //即便生产者的对象关闭了，程序还在运行哦
            producer.close();
            
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

View Code

接收消息：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Receive {
    //连接账号
    private String userName = "";
    //连接密码
    private String password = "";
    //连接地址
    private String brokerURL = "tcp://192.168.0.130:61616";
    //connection的工厂
    private ConnectionFactory factory;
    //连接对象
    private Connection connection;
    //一个操作会话
    private Session session;
    //目的地，其实就是连接到哪个队列，如果是点对点，那么它的实现是Queue，如果是订阅模式，那它的实现是Topic
    private Destination destination;
    //消费者，就是接收数据的对象
    private MessageConsumer consumer;
    
    
    public static void main(String[] args) {
        PTPReceive receive = new PTPReceive();
        receive.start();
    }
    
    public void start(){
        try {
            //根据用户名，密码，url创建一个连接工厂
            factory = new ActiveMQConnectionFactory(userName, password, brokerURL);
            
            //从工厂中获取一个连接
            connection = factory.createConnection();
            
            //测试过这个步骤不写也是可以的，但是网上的各个文档都写了
            connection.start();
            
            //创建一个session
            //第一个参数: 为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
            //第二个参数：为false，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
            //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
            //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
            //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            
            //创建一个到达的目的地，其实想一下就知道了，activemq不可能同时只能跑一个队列吧，这里就是连接了一个名为"text-msg"的队列，这个会话将会到这个队列，当然，如果这个队列不存在，将会被创建
            destination = session.createQueue("HelloWord");
            
            //点对点与订阅模式唯一不同的地方，就是这一行代码，点对点创建的是Queue，而订阅模式创建的是Topic
            //destination = session.createTopic("topic");
            
            //过期失败的消息所在队列
            //destination = session.createQueue("ActiveMQ.DLQ");
            
            //根据session，创建一个接收者对象
            consumer = session.createConsumer(destination);
            
            
            //被动模式：实现一个消息的监听器
            //实现这个监听器后，以后只要有消息，就会通过这个监听器接收到
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        //获取到接收的数据
                        String text = ((TextMessage)message).getText();
                        //获取javaBean
                        //Person p = (Person) ((ObjectMessage)message).getObject();
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            
            
            
            /*主动模式：当程序有着处理消息的能力时，主动去接收一条消息进行处理
            if(当程序有能力处理){//当程序有能力处理时接收
                    Message receive = consumer.receive();
　　　　　　　　　　　//这个可以设置超时时间，超过则不等待消息　
　　　　　　　　　　　 recieve.receive(10000);
                    //其实receive是一个阻塞式方法，一定会拿到值的
                    if(null != receive){
                        String text = ((TextMessage)receive).getText();
                        receive.acknowledge();
                        System.out.println(text);
                    }else{
                        //没有值嘛
                        //
                    }
                }
            */
            
            //关闭接收端，也不会终止程序哦
            //consumer.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

View Code

整合spring

配置spring*.xml文件：

　　 <!-- 配置connectionFactory -->
    <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://127.0.0.1:61616</value>
                </property>
            </bean>
        </property>
        <property name="maxConnections" value="100"></property>
    </bean>

    <!-- Spring JMS Template -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory">
            <ref local="jmsFactory" />
        </property>
        <property name="defaultDestinationName" value="HelloWord" />
        <!-- 区别它采用的模式为false是p2p为true是订阅 -->
        <property name="pubSubDomain" value="true" />
    </bean>

    <!-- 发送消息的目的地（一个队列） -->
    <bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">
        <!-- 设置消息队列的名字 -->
        <constructor-arg index="0" value="HelloWord" />
    </bean>

代码示例：

import javax.jms.JMSException;
import javax.jms.Session;

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

public class HelloActiveMQ {

    /**
     * @param args
     * jmsTemplate和destination都是在spring配置文件中进行配制的
     * Sender只使用了配置文件中的jmsFactory，jmsTemplate，还有destination这三个属性
     */
    public static void main(String[] args) {
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext-jms.xml");
        JmsTemplate template = (JmsTemplate) applicationContext.getBean("jmsTemplate");
        Destination destination = (Destination) applicationContext.getBean("destination");
        template.send(destination, new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage("Hello ActiveMQ！");
            }
        });
        System.out.println("成功发送了一条消息");
        
        
        while (true) {
            TextMessage txtmsg = (TextMessage) jmsTemplate.receive(destination);
            System.out.println("接收消息内容为: " + txtmsg.getText());
        }   
    }
}

后台管理地址

http://127.0.0.1:8161/admin/