一、消息队列(MQ)概述
消息队列(Message Queue),是分布式系统中重要的组件,其通用的使用场景可以简单地描述为:
当不需要立即获得结果,但是并发量又需要进行控制的时候,差不多就是需要使用消息队列的时候。
消息队列主要解决了应用耦合、异步处理、流量削锋等问题。
当前使用较多的消息队列有RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等,而部分数据库如Redis、Mysql以及phxsql也可实现消息队列的功能。
二 、ActiveMQ
有那么多得消息队列,各有各的有优缺点,今天主要来聊聊ActiveMQ
ActiveMQ是由Apache出品,ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。它非常快速,支持多种语言的客户端和协议,而且可以非常容易的嵌入到企业的应用环境中,并有许多高级功能。
主要特性:
服从 JMS 规范:JMS 规范提供了良好的标准和保证,包括:同步或异步的消息分发,一次和仅一次的消息分发,消息接收和订阅等等。遵从 JMS 规范的好处在于,不论使用什么 JMS 实现提供者,这些基础特性都是可用的;
连接性:ActiveMQ 提供了广泛的连接选项,支持的协议有:HTTP/S,IP 多播,SSL,STOMP,TCP,UDP,XMPP等等。对众多协议的支持让 ActiveMQ 拥有了很好的灵活性。
支持的协议种类多:OpenWire、STOMP、REST、XMPP、AMQP ;
持久化插件和安全插件:ActiveMQ 提供了多种持久化选择。而且,ActiveMQ 的安全性也可以完全依据用户需求进行自定义鉴权和授权;
支持的客户端语言种类多:除了 Java 之外,还有:C/C++,.NET,Perl,PHP,Python,Ruby;
代理集群:多个 ActiveMQ 代理可以组成一个集群来提供服务;
异常简单的管理:ActiveMQ 是以开发者思维被设计的。所以,它并不需要专门的管理员,因为它提供了简单又使用的管理特性。有很多中方法可以监控 ActiveMQ 不同层面的数据,包括使用在 JConsole 或者 ActiveMQ 的Web Console 中使用 JMX,通过处理 JMX 的告警消息,通过使用命令行脚本,甚至可以通过监控各种类型的日志。
使用ActiveMQ需要:
Java JDK
ActiveMQ安装包
ActiveMQ可以运行在Java语言所支持的平台之上。
优点:
跨平台(JAVA编写与平台无关有,ActiveMQ几乎可以运行在任何的JVM上)
可以用JDBC:可以将数据持久化到数据库。虽然使用JDBC会降低ActiveMQ的性能,但是数据库一直都是开发人员最熟悉的存储介质。将消息存到数据库,看得见摸得着。而且公司有专门的DBA去对数据库进行调优,主从分离;
支持JMS :支持JMS的统一接口;
支持自动重连;
有安全机制:支持基于shiro,jaas等多种安全配置机制,可以对Queue/Topic进行认证和授权。
监控完善:拥有完善的监控,包括Web Console,JMX,Shell命令行,Jolokia的REST API;
界面友善:提供的Web Console可以满足大部分情况,还有很多第三方的组件可以使用,如hawtio;
缺点:
社区活跃度不及RabbitMQ高;
根据其他用户反馈,会出莫名其妙的问题,会丢失消息;
目前重心放到activemq6.0产品-apollo,对5.x的维护较少;
不适合用于上千个队列的应用场景;
三、具体实例
首先我们去官网下载MQ,官网地址:http://activemq.apache.org/,我目前下载的是5.8.0版本的,下载好后直接双击activemq.bat启动
启动成功后,输入默认的访问地址:http://localhost:8161/admin/index.jsp
用户名:admin 密码:admin
用户名密码设置具体配置文件在conf下jetty-realm.properties里面,jetty.xml文件修改端口号。
然后点击队列(Queues),输入队列名称(Queue Name)FirstQueue,然后点创建(Create)
下一步我们创建一个maven 项目,在pom.xml引入jar
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.8.0</version>
</dependency>
新建文件 Procuder
package Mq.ActivityMq;
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.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
*
* @author 生产者
*
*/
public class Procuder {
//默认连接用户名
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
//默认连接密码
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
//默认连接地址
private static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
public static void main(String[] args) {
//连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
try {
//连接
Connection connection = connectionFactory.createConnection();
//启动连接
connection.start();
//创建session
Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
//消息目的地
Destination destination = session.createQueue("FirstQueue");
//消息生产者
MessageProducer producer = session.createProducer(destination);
//设置不持久化,此处学习,实际根据项目决定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//发送消息
for (int i = 0; i < 5; i++) {
//创建一条文本消息
TextMessage message = session.createTextMessage("你好ActiveMQ: 这是第 " + i + " 条消息");
//生产者发送消息
producer.send(message);
}
session.commit();
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
新建文件Consumer
package Mq.ActivityMq;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
*
* @author 消费者
*
*/
public class Consumer {
//默认连接用户名
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
//默认连接密码
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
//默认连接地址
private static final String BROKER_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
public static void main(String[] args) {
//连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
try {
//连接
Connection connection = connectionFactory.createConnection();
//启动连接
connection.start();
//创建session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//消息目的地
Destination destination = session.createQueue("FirstQueue");
//消息消费者
MessageConsumer consumer = session.createConsumer(destination);
while (true) {
TextMessage message = (TextMessage) consumer.receive();
if (message != null) {
System.out.println("接收到消息: " + message.getText());
} else {
break;
}
}
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
测试结果
先运行Producer(生产者),再运行Consumer(消费者)
这是我们查看MQ视图
我们看到有0条等待消费的消息,有1条消费者,生产者发布了5条消息,消费者消费了5条消息。名词解释如下:
Messages Enqueued:表示生产了多少条消息,记做P
Messages Dequeued:表示消费了多少条消息,记做C
Number Of Consumers:表示在该队列上还有多少消费者在等待接受消息
Number Of Pending Messages:表示还有多少条消息没有被消费,实际上是表示消息的积压程度,就是P-C
四、总结
如上所示只是ActiveMQ简单实例,实际的业务场景需要配置更多的参数,比如消费者消费完毕后要通知生成者对消息进行销毁.....等等,掌握了基本的原理后,其他都是结合具体业务场景水到渠成的事情。