ActiveMQ基本使用

消息队列，目前在实际的开发项目中应用十分广泛。本文主要介绍入门级的ActiveMQ的基本使用以及相关的概念。

一、JMS

 　　全称 Java Message Service，即Java消息服务。JMS是一套Java的面向消息中间件的API接口规范，用于在不同应用程序中异步的发送消息。JMS本身语言无关，绝大多数的消息中间件厂商都提供了对JMS的支持。基于JMS实现的消息中间件，也叫做JMS Provider。

　　消息服务，传递的载体自然是消息（Message）。在JMS中，消息主体可以简单分为几个类型：简单文本(TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)。

　　JMS中，有一整套的名词提供，下面简单说说相关的名词以及解释：

　　1、Destination

　　　　目的地。JMS Provider(消息中间件)进行维护，用于对消息（Message）对象进行管理。MessageProducer需要指定Destination才能进行发送消息，MessageConsumer需要指定Destination才能进项消息的接收消费。

　　2、Producer

　　　　消息生产者，负责发送消息到Destination目的地。应用接口为MessageProducer。

　　3、Consumer

　　　　消息接收者，负责接收消费指定Destination的消息，应用接口为MessageConsumer。

　　4、Message

　　　　消息体，一般常用的有：TextMessage、ObjectMessage、BytesMessage。

　　5、ConnectionFactory

　　　　连接工厂。用于创建连接信息的。

　　6、Connection

　　　　连接。用于和ActiveMQ服务端建立连接，一般由连接工厂创建。

　　7、Session

　　　　会话。Session是操作消息的接口。可以通过session创建生产者、消费者、消息等信息。Session支持事务特征，当需要批处理（发送或者接收）消息的时候，可以将这些操作放到一个事务中进行。

　　8、Queue和Topic

　　　　Queue - 队列目的地。Topic - 主题目的地。都是Destination的子接口。

　　　　Queue：一般队列中的一条消息，默认的只能被一个消费者消费。消费完成即删除。

　　　　Topic：消息会发送给所有订阅的消费者。消息不会持久化，也即如果发消息时不存在订阅关系，则消息直接丢弃。

　　9、PTP

　　　　point to point，点对点模型。针对Queue实现的消息处理方式。

　　10、Pub/Sub

　　　　Publish & Subscribe，发布订阅模型。针对Topic实现的消息处理方式。

二、ActiveMQ简介

 　　ActiveMQ是纯Java编写的消息中间件服务，完全支持JMS规范。支持多种语言编写客户端：C、C++、C#、Java、PHP、Python等。应用协议包括：OpenWire、STOMP、WS-Notification、MQTT以及AMQP。对Spring的支持非常好，可以很容易的集成到现有的Spring系统中去使用。在消息的持久化上，支持jdbc和journal两种方式的使用。另外，在集群搭建上，也比较容易。上手难度比较低，适合大多数的中小型项目使用。

三、ActiveMQ安装

 　　1、下载安装包

　　　　ActiveMQ官网下载包，注意的是，ActiveMQ 5.10.x以上版本需要JDK1.8的环境。其他只需要1.7环境即可。

　　2、 上传linux服务器

　　　　本文下载上传的是最新的 5.15.9版本。

　　3、解压安装文件

　　　　tar -zxf apache-activemq-5.15.9-bin.tar.gz

　　4、检查权限

　　　　ls -al apache-activemq-5.15.9/bin

　　　　如果权限不足的话，会无法执行，修改文件权限：chmod 755 activemq

　　5、移动到集中目录（可选）

　　　　cp -r apache-activemq-5.15.9 /usr/local/activemq

　　6、ActiveMQ配置文件简介

　　　　配置文件目录为${activemq_home}/conf，对配置文件的修改，都必须重新启动ActiveMQ才能生效。

　　　　6.1、activemq.xml

　　　　　　就是Spring配置文件，配置了MQ使用的默认的对象组件。

　　　　　　broker -  ActiveMQ的实例标签，配置的内容基本在此标签内部

　　　　　　destinationPolicy - 配置目的地的规则信息

　　　　　　persistenceAdapter - 配置持久化策略

　　　　　　systemUsage - 内存信息设置

　　　　　　transportConnectors - 配置连接端口信息，一般Java使用最多的就是openwire协议，也就是基于tcp的协议访问，默认开放端口61616，可自定义修改。

　　　　6.2、jetty.xml

　　　　　　ActiveMQ默认控制台的配置文件，也是个Spring的配置文件。一般在标签 jettyPort 中可以修改控制台的访问端口，默认是8161。控制台管理的时候需要用户名密码登录，默认为admin:admin。

　　7、启动或停止ActiveMQ

　　　　启动：${activemq_home}/bin/activemq start

　　　　重启：${activemq_home}/bin/activemq restart

　　　　停止：${activemq_home}/bin/activemq stop

　　8、测试ActiveMQ

　　　　8.1、查看是否成功启动：jps命令，查看是否有activemq.jar的信息：

　　　　8.2、检查进程信息：ps -ef|grep activemq

　　　　8.3、ActiveMQ控制台

　　　　一般，使用浏览器访问ActiveMQ控制台管理信息，地址格式：http://ip:port/admin

　　　　其中端口设置见上述 6.2节中提及。默认为8161。访问正确，出现下面界面几位成功启动：

　　9、查看ActiveMQ状态

　　${activemq_home}/bin/activemq status

 　　

四、ActiveMQ基本使用

 　　上述安装完ActiveMQ之后，就可以使用Java代码去进行访问操作啦。下面开始介绍使用。

 　　编写生产者：

package com.cfang.mq.simpleCase;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class SimpleProducer {

	public static void main(String[] args) {
		SimpleProducer simpleProducer = new SimpleProducer();
		simpleProducer.sendMsg("我有一只小毛驴");
	}
	
	public void sendMsg(String msg) {
		ConnectionFactory factory = null;	//连接工厂
		Connection connection = null;		//连接对象
		Session session = null;				//session会话
		Destination destination = null;		//目的地
		MessageProducer producer = null;	//生产者
		Message message = null;				//消息
		try {
			//创建连接工厂，前两个参数是做安全认证使用，本例中尚未开启。
			factory = new ActiveMQConnectionFactory(null, null, "tcp://172.31.31.160:61616");
			//通过工厂创建连接对象
			connection = factory.createConnection();
			//启动连接。生产者通常来说不是必须显式启动的，在发送消息的时候，会检测是否启动，未启动的话会先进行启动操作。
			connection.start();
			/**
			 * 	根据连接对象信息，创建session会话信息。
			 * 	第一个参数为是否开启事务特性。
			 * 		false - 不开启事务。使用比较多的配置。
			 * 		true - 开启事务。如果开启事务，这第二个参数默认无效了，建议还是写成Session.SESSION_TRANSACTED
			 * 	第二个参数表示消息确认机制。
			 * 		AUTO_ACKNOWLEDGE - 自动消息确认。消息消费者接受处理消息后，自动发送确认信息
			 * 		CLIENT_ACKNOWLEDGE - 手动确认。消息消费者在接受处理消息后，必须手动发起确认ack信息
			 * 		DUPS_OK_ACKNOWLEDGE - 有副本的手动确认机制。
			 */
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			//创建目的地，参数是目的地名称，也即队列名。
			destination = session.createQueue("tp_simple_queue");
			//创建消息生产者，参数为目的地，也可以不指定，在发送消息的时候再指定
			producer = session.createProducer(destination);
			//创建消息
			message = session.createTextMessage(msg);
			//发送到ActiveMQ指定的目的地中
			producer.send(message);
			System.out.println("=====send msg ok!=====");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			// 回收资源
			if(producer != null){ // 回收消息发送者
				try {
					producer.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
			if(session != null){ // 回收会话对象
				try {
					session.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
			if(connection != null){ // 回收连接对象
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}
}

　　编写消费者

package com.cfang.mq.simpleCase;

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.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class SimpleConsumer {

	public static void main(String[] args) {
		SimpleConsumer simpleConsumer = new SimpleConsumer();
		System.out.println("=====receive msg: " + simpleConsumer.receiveMsg());
	}
	
	public String receiveMsg() {
		String result = "";
		ConnectionFactory factory = null;	//连接工厂
		Connection connection = null;		//连接对象
		Session session = null;				//session会话
		Destination destination = null;		//目的地
		MessageConsumer consumer = null;	//生产者
		Message message = null;				//消息
		try {
			factory = new ActiveMQConnectionFactory(null, null, "tcp://172.31.31.160:61616");
			connection = factory.createConnection();
			//不同于生产者存在自动启动机制，消息的消费者必须显式的手动启动连接
			connection.start();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			//创建目的地，参数是目的地名称，也即队列名。
			destination = session.createQueue("tp_simple_queue");
			//创建消费者，参数为目的地，也可以不指定，在发送消息的时候再指定
			consumer = session.createConsumer(destination);
			//接收一条消息
			message = consumer.receive();
			//手动确认
//			message.acknowledge();
			result = ((TextMessage)message).getText();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			// 回收资源
			if(consumer != null){ // 回收消息发送者
				try {
					consumer.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
			if(session != null){ // 回收会话对象
				try {
					session.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
			if(connection != null){ // 回收连接对象
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
		return result;
	}
}

五、ActiveMQ安全认证　

　　上面简单的写了个生产者、消费者的案例，其中在创建连接工厂ConnectionFactory的时候，前两个参数都没有去设置值，程序运行也没有任何的问题。这是因为安装的ActiveMQ尚未启用安全认证插件，这种情况下，只要知道mq的地址信息，均可以连接上去进行消息的收发。

　　在一些特定的需求中，需要对ActiveMQ的连接进行认证，下面介绍ActiveMQ的安全认证配置。

　　ActiveMQ的安全认证配置，是基于用户名密码校验的。启用安全认证，需要对activemq.xml进行修改，具体修改办法：

　　　　在broker标签中，增加安全认证插件：

<plugins>
            <!--  use JAAS to authenticate using the login.config file on the classpath to configure JAAS -->
			<!--  添加jaas认证插件activemq在login.config里面定义,详细见login.config-->
            <jaasAuthenticationPlugin configuration="activemq" />
            <!--  lets configure a destination based authorization mechanism -->
            <authorizationPlugin>
                <map>
                    <authorizationMap>
                        <authorizationEntries>
                            <authorizationEntry topic=">" read="admins" write="admins" admin="admins" />
                            <authorizationEntry queue=">" read="admins" write="admins" admin="admins" />
                            <authorizationEntry topic="ActiveMQ.Advisory.>" read="admins" write="admins" admin="admins"/>
                            <authorizationEntry queue="ActiveMQ.Advisory.>" read="admins" write="admins" admin="admins"/>
                        </authorizationEntries>
                    </authorizationMap>
                </map>
            </authorizationPlugin>
        </plugins>

　　/conf/login.config配置内容：

activemq {
    org.apache.activemq.jaas.PropertiesLoginModule required
        org.apache.activemq.jaas.properties.user="users.properties"
        org.apache.activemq.jaas.properties.group="groups.properties";
};

　　其中 user 代表的是用户的配置文件信息，group 代表的是用户组信息配置文件。

　　/conf/user.properties，配置文件中格式为：用户名=密码

admin=admin

　　/conf/group.properties，配置文件中格式为：用户组名=用户名,用户名,用户名

admins=admin

　　确认添加完毕后，需要重启ActiveMQ。之后Java应用程序创建访问ConnectionFactory的时候，必须指定上述配置文件中正确的用户名密码，否则会报错如下：