03、ActiveMQ 实战 - Java编码实现ActiveMQ通讯（Queue）
前言
这篇文章是大家最喜欢看到的文章~~怎么用Java去操作ActiveMQ\color{red}怎么用Java去操作ActiveMQ怎么用Java去操作ActiveMQ 值得一提，在实际项目中，不会使用这种方式去操作ActiveMQ，这就好比我们在学习MySql时，一开始是通过JDBC去操作数据库，在项目中往往写的并不是JDBC，而是使用Mybatis操作数据库 作为学习者来说，我们是有必要了解Java底层是如何操作数据库，万变不离其宗，以一个activemq突破口
回顾JDBC操作数据库
Java对数据库的操作有很多种技术。例如说jdbc，dbutil +C3p0，hibernate,jdbcTemplate等等，到后面的话使用一些高级的框架去操作数据库，比如hibernate。但是底层操作数据库是重要的，高级框架也是也底层为基础搭建的
通过几种操作MySql方式演变，最后我们使用框架，只需要关注与业务操作相关即可，无需担忧连接配置\color{red}通过几种操作MySql方式演变，最后我们使用框架，只需要关注与业务操作相关即可，无需担忧连接配置通过几种操作MySql方式演变，最后我们使用框架，只需要关注与业务操作相关即可，无需担忧连接配置 ActiveMQ亦是如此，不过不急，请耐心看完
回顾JDBC操作六个步骤
JDBC的六个固定步骤
1、注册数据库驱动[利用反射]
2、取得数据库连接对象Connection
3、创建SQL对象
4、执行SQL命令，并返回结果集
5、处理结果集
6、依次关闭结果集
//1，注册数据库驱动有两种方式
第一种是：直接注册数据库驱动
DriverManager.registerDriver(new Driver());
第二种是：利用反射机制间接加载数据库驱劝，推荐用第二种
Class.forName("com.mysql.jdbc.Driver");
//2. 取得数据库连接对象Connection
//取得与MySQL数据库连接的桥梁,参数分别是：连接数据库 用户名 密码Connection conn = DriverManager.getConnection( "jdbc:mysql:///zz2017","root","xiaozheng");
//前两步骤需要记住以下。下面的
//4:执行sql语句
insert/update/delete----PreparedStatement .executeUpdate(sql)：返回值表示影响记录的行数
select------------------PreparedStatement .exeucteQuery():返回值表示符合条件的记录
正文
JMS编码总体规范
ConnectionFactory加载MQ连接驱动，并创建Connnection连接对象（类似连接数据库时加载JDBC驱动，生成java.sql.Connection对象），由Connection对象创建Session会话对象，Session可以创建Message（消息），Message Producer（消息生产者）和Message Consumer（消息消费者）。Message Producer（消息生产者）将Message（消息）发送到Destination（目的地），Message Consumer（消息消费者）从Destination（目的地）中接受消息。
JMS开发基本步骤
1、创建一个connection factory
2、通过connection factory来创建JMS connection
3、启动JMS connection
4、通过JMS connection创建JMS session
5、通过JMS session创建JMS destination（目的地 队列/主题）
6、通过JMS session创建JMS producer或者创建JMS consume并设置destination
7、JMS consumer同步接受或者注册一个JMS message listener异步接受JMS message
8、发送(send)或者接收(receive)JMS message
9、关闭所有JMS资
Destination简介
Destination即消息队列（Queue）和主题（Topic），存在一下两者模式：
1、 队列1对1模式：；
消息发送者（Sender）将消息发送到队列（Queue），接受者（Receiver）从队列中接受消息，一个发送者发送给一个接收者。
2、 主题1对n模式：；
消息发布者（Publisher）将消息发布到指定的主题（Topic），订阅该主题的所有订阅者（Subscriber）从该主题中接受消息，一个发布者的消息可以被多个订阅者接受。
队列的特点（这一篇先讲主题）
每个消息只能有一个消费者，类似于1对1的关系。好比个人快递自己领自己的。
消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消息的时候是否处于运行状态，消费者都可以提取消息。好比我们的发送短信，发送者发送后不见得接收者会即收即看。
消息被消费后队列中不会再存储，所以消费者不会消费到已经被消费掉的消息。
环境准备
JDK+ ActiveMQ服务 + Maven
构建ActiveMQDemo的Maven工程，Maven依赖如下：
<!-- activemq 所需要的jar 包-->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.9</version>
</dependency>
<!-- activemq 和 spring 整合的基础包 -->
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>3.16</version>
</dependency>
生产者生产消息
package com.at.activemq.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class JmsProduce {
// linux 上部署的activemq 的 IP 地址 + activemq 的端口号，如果用自己的需要改动
public static final String ACTIVEMQ_URL = "tcp://47.98.163.118:61616";
// public static final String ACTIVEMQ_URL = "nio://47.98.163.118:61608";
public static final String QUEUE_NAME = "jdbc01";
public static void main(String[] args) throws Exception{
// 1 按照给定的url创建连接工程，这个构造器采用默认的用户名密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
// 设置允许有数据丢失
// activeMQConnectionFactory.setUseAsyncSend(true);
// 2 通过连接工厂连接 connection 和 启动
javax.jms.Connection connection = activeMQConnectionFactory.createConnection();
// 启动
connection.start();
// 3 创建回话 session
// 两个参数，第一个事务， 第二个签收
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
// 4 创建目的地 （两种 ： 队列/主题 这里用队列）
Queue queue = session.createQueue(QUEUE_NAME);
// 5 创建消息的生产者
MessageProducer messageProducer = session.createProducer(queue);
// 非持久化消息 和持久化消息演示
messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT); // 持久化 如果开启
// 就会存入文件或数据库中
// 6 通过messageProducer 生产 3 条 消息发送到消息队列中
for (int i = 1; i < 4 ; i++) {
// 7 创建字消息
TextMessage textMessage = session.createTextMessage("msg--" + i);
// 8 通过messageProducer发布消息
messageProducer.send(textMessage);
}
// 9 关闭资源
messageProducer.close();
session.close();
connection.close();
// session.commit();
System.out.println(" **** 消息发送到MQ完成 ****");
}
}
点击运行，无异常打印，并且控制台成功打印finish，则表示程序运行成功。
查看页面发现此时如果存在队列名称为jdbc01有3条未读消息，则表示消息成功发送到了ActiveMQ。
队列表头说明：
由类的结构图可知，队列（Queue）和主题（Topic）拥有共同的父接口（Destination）
消费者消费消息（同步阻塞）
package com.at.activemq.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
// 消息的消费者 也就是回答消息的系统
public class JmsConsumer {
// linux 上部署的activemq 的 IP 地址 + activemq 的端口号，如果用自己的需要改动
public static final String ACTIVEMQ_URL = "tcp://47.98.163.118:61616";
// public static final String ACTIVEMQ_URL = "nio://47.98.163.118:61608";
public static final String QUEUE_NAME = "jdbc01";
public static void main(String[] args) throws Exception{
System.out.println(" 这里是 1 号 消费者 ");
// 1 按照给定的url创建连接工程，这个构造器采用默认的用户名密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
// 2 通过连接工厂连接 connection 和 启动
javax.jms.Connection connection = activeMQConnectionFactory.createConnection();
// 启动
connection.start();
// 3 创建回话 session
// 两个参数，第一个事务， 第二个签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4 创建目的地 （两种 ： 队列/主题 这里用队列）
Queue queue = session.createQueue(QUEUE_NAME);
// 5 创建消息的消费者
MessageConsumer messageConsumer = session.createConsumer(queue);
/* 同步阻塞方式reveive() 空参数的receive方法是阻塞，有参数的为等待时间
订阅者或消费者使用MessageConsumer 的receive() 方法接收消息，receive 在接收之前一直阻塞 */
while(true){
// 这里是 TextMessage 是因为消息发送者是 TextMessage ， 接受处理的
// 也应该是这个类型的消息
TextMessage message = (TextMessage)messageConsumer.receive();
System.out.println("****消费者的消息："+message.getText());
}
}
}
此时消息队列中情况如下：
未出队数量（未读）：0
消费者数量：1 （Java程序仍然连接着ActiveMQ）
累计入队数量：3
累计出队数量：3
Java操作ActiveMQ的API中存在很多重载方法，难度不高，感兴趣的同学们可自行研究。
receive增加入参
消费者核心代码
/* 同步阻塞方式reveive() 空参数的receive方法是阻塞，有参数的为等待时间
订阅者或消费者使用MessageConsumer 的receive() 方法接收消息，receive 在接收之前一直阻塞*/
while(true){
// 这里是 TextMessage 是因为消息发送者是 TextMessage ， 接受处理的
// 也应该是这个类型的消息
TextMessage message = (TextMessage)messageConsumer.receive(4000L); // 4秒
if (null != message){
System.out.println("****消费者的消息："+message.getText());
}else {
break;
}
}
// 通过监听的方式来消费消息
// 通过异步非阻塞的方式消费消息
// 通过messageConsumer 的setMessageListener 注册一个监听器，
// 当有消息发送来时，系统自动调用MessageListener 的 onMessage 方法处理消息
/*messageConsumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
if (null != message && message instanceof TextMessage){
TextMessage textMessage = (TextMessage)message;
try {
System.out.println("****消费者的消息："+textMessage.getText());
}catch (JMSException e) {
e.printStackTrace();
}
}
}
});
*/
// 保证控制台不灭 不然activemq 还没连上就关掉了连接
//System.in.read();
messageConsumer.close();
session.close();
connection.close();
说明 第一种方式，receive()方法没有指定时间会一直阻塞在这里，整个程序不会结束，而第二种方式设置阻塞时间，4秒后自动往下执行\color{red}第一种方式，receive()方法没有指定时间会一直阻塞在这里，整个程序不会结束，而第二种方式设置阻塞时间，4秒后自动往下执行第一种方式，receive()方法没有指定时间会一直阻塞在这里，整个程序不会结束，而第二种方式设置阻塞时间，4秒后自动往下执行
消费者消费消息之消息监听器（异步非阻塞）
package com.at.activemq.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
// 消息的消费者 也就是回答消息的系统
public class JmsConsumer {
// linux 上部署的activemq 的 IP 地址 + activemq 的端口号，如果用自己的需要改动
public static final String ACTIVEMQ_URL = "tcp://47.98.163.118:61616";
// public static final String ACTIVEMQ_URL = "nio://47.98.163.118:61608";
public static final String QUEUE_NAME = "jdbc01";
public static void main(String[] args) throws Exception{
System.out.println(" 这里是 1 号 消费者 ");
// 1 按照给定的url创建连接工程，这个构造器采用默认的用户名密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
// 2 通过连接工厂连接 connection 和 启动
javax.jms.Connection connection = activeMQConnectionFactory.createConnection();
// 启动
connection.start();
// 3 创建回话 session
// 两个参数，第一个事务， 第二个签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 4 创建目的地 （两种 ： 队列/主题 这里用队列）
Queue queue = session.createQueue(QUEUE_NAME);
// 5 创建消息的消费者
MessageConsumer messageConsumer = session.createConsumer(queue);
/* 同步阻塞方式reveive() 空参数的receive方法是阻塞，有参数的为等待时间
订阅者或消费者使用MessageConsumer 的receive() 方法接收消息，receive 在接收之前一直阻塞*/
// 通过监听的方式来消费消息
// 通过异步非阻塞的方式消费消息
// 通过messageConsumer 的setMessageListener 注册一个监听器，
// 当有消息发送来时，系统自动调用MessageListener 的 onMessage 方法处理消息
messageConsumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
if (null != message && message instanceof TextMessage){
TextMessage textMessage = (TextMessage)message;
try {
System.out.println("****消费者的消息："+textMessage.getText());
}catch (JMSException e) {
e.printStackTrace();
}
}
}
});
// 保证控制台不灭 不然activemq 还没连上就关掉了连接
System.in.read();
messageConsumer.close();
session.close();
connection.close();
}
}
小总结两种消费方式
1：同步阻塞方式(receive)
订阅者或接收者抵用MessageConsumer的receive()方法来接收消息，receive方法在能接收到消息之前（或超时之前）将一直阻塞
2：异步非阻塞方式（监听器onMessage()）
订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器，
当消息到达之后，系统会自动调用监听器MessageListener的onMessage(Message message)方法。
案例
1、 先生产，只启动1号消费者，1号消费者能消费者？
运行之前编写生产者JMSProducer生产3条消息消息。
再运行其中一个消费者消费消息。
答案：1号消费者能消费。
2、 先生产，先启动1号消费者，再启动2号消费者，2号消费者还能消费消息吗？；
验证此问题，需要对同一个消费者类启动两次main方法，IDEA默认只能启动一次。解决此问题需要勾选Allow parallel run。
和以上相同方式启动生产者JMSProducer生产3条消息。
第一次启动JMSConsumerForListener消费消息。
第二次启动JMSConsumerForListener消费消息。
答案：队列里消息已被先启动的1号消费者消费掉了，2号消费者再启动无消息消费。
1、 先启动2个消费者，再生产6条消息，消费情况如何？
选项：A. 只1号消费者消费 B.只2号消费者消费 C.都能消费 D.都不能消费
启动2次JMSConsumerForListener消费者等待消息消费。
2个消费者消费情况如下：
答案：C（负载均衡）
扩展阅读
回顾9个步骤
1：创建一个connection factory
2：通过connection factory来创建JMS connection
3：启动JMS connection
4：通过JMS connection创建JMS session
5：通过JMS session创建JMS destination（目的地 队列/主题）
6：通过JMS session创建JMS producer或者创建JMS consume并设置destination
7：JMS consumer同步接受或者注册一个JMS message listener异步接受JMS message
8：发送(send)或者接收(receive)JMS message
9：关闭所有JMS资
// 1 按照给定的url创建连接工程，这个构造器采用默认的用户名密码
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
// 2 通过连接工厂连接 connection 和 启动
javax.jms.Connection connection = activeMQConnectionFactory.createConnection();
// 启动
connection.start();
以ActiveMQConnectionFactory为例看看源码
看看如何创建createConnection
有空的话可以看看源码，对你们有帮助的