一.简介
1.1 JMS简介
JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。
JMS是一种与厂商无关的 API，用来访问收发系统消息，它类似于JDBC(Java Database Connectivity)。这里，JDBC 是可以用来访问许多不同关系数据库的 API，而 JMS 则提供同样与厂商无关的访问方法，以访问消息收发服务。许多厂商都支持 JMS，包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ。 JMS 使您能够通过消息收发服务（有时称为消息中介程序或路由器）从一个 JMS 客户机向另一个 JMS客户机发送消息。消息是 JMS 中的一种类型对象，由两部分组成：报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。根据有效负载的类型来划分，可以将消息分为几种类型，它们分别携带：简单文本(TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。
1.2 ActiveMQ简介
Apache ActiveMQ是Apache软件基金会所研发的开放源代码消息中间件；由于ActiveMQ是一个纯Java程序，因此只需要操作系统支持Java虚拟机，ActiveMQ便可执行。
1.3 ActiveMQ启动和运行
主页：http://activemq.apache.org/
目前最新版本：5.15.12 (March 9, 2020)
开发包及源码下载地址：http://activemq.apache.orgactivemq-5111-release.htmActiveMQ
服务启动地址：http://127.0.0.1:8161/admin/
用户名/密码：admin/admin
二.ActiveMQ点对点消息实现
2.1 直接 Receive 方式
Session.AUTO_ACKNOWLEDGE。当客户成功的从receive方法返回的时候，或者从MessageListener.onMessage方法成功返回的时候，会话自动确认客户收到的消息。
Session.CLIENT_ACKNOWLEDGE。 客户通过消息的 acknowledge 方法确认消息。需要注意的是，在这种模式中，确认是在会话层上进行：确认一个被消费的消息将自动确认所有已被会话消 费的消息。例如，如果一个消息消费者消费了 10 个消息，然后确认第 5 个消息，那么所有 10 个消息都被确认。
Session.DUPS_ACKNOWLEDGE。 该选择只是会话迟钝第确认消息的提交。如果 JMS provider 失败，那么可能会导致一些重复的消息。如果是重复的消息，那么 JMS provider 必须把消息头的 JMSRedelivered 字段设置为 true。
生产者：
package main.activeMq;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
/**
* 消息生产者
*
* @author wangjian
* @date 2020/5/8 22:26
*/
public class Producer {
// 默认的用户名，密码，连接地址
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
// 发送消息数量
private static final int SENDNUM = 10;
public static void main(String[] args) {
ConnectionFactory connectionFactory; // 连接工厂
Connection connection = null; // 连接
Session session; // 会话：接收或者发送消息的线程
Destination destination; // 消息的目的地
MessageProducer messageProducer; // 消息生产者
// 实例化连接工厂(URL:failover://tcp://localhost:61616)
connectionFactory = new ActiveMQConnectionFactory(Producer.USERNAME, Producer.PASSWORD, Producer.URL);
try {
// 通过连接工厂获取连接
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 创建session,true:开启事务
session = connection.createSession(true, SENDNUM);
// 创建消息队列
destination = session.createQueue("firstQueue"); // 点对点
// destination = session.createTopic("firstTopic"); // 发布订阅
// 创建消息生产者
messageProducer = session.createProducer(destination);
sendMessage(session, messageProducer);
session.commit();
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
/**
* 发送消息
*
* @param session
* @param producer
*/
public static void sendMessage(Session session, MessageProducer producer) {
for (int i = 0; i < SENDNUM; i++) {
try {
TextMessage textMessage = session.createTextMessage("ActiveMQ 发送的消息：" + i);
System.out.println("发送消息：" + "ActiveMQ 发送的消息" + i);
producer.send(textMessage);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
消费者：
package main.activeMq;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
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;
/**
* 消费者
*
* @author wangjian
* @date 2020/5/8 22:53
*/
public class Consumer {
// 默认的用户名，密码，连接地址
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
public static void main(String[] args) {
ConnectionFactory connectionFactory; // 连接工厂
Connection connection = null; // 连接
Session session; // 会话：接收或者发送消息的线程
Destination destination; // 消息的目的地
MessageConsumer messageConsumer; // 消息消费者
// 实例化连接工厂
connectionFactory = new ActiveMQConnectionFactory(Consumer.USERNAME, Consumer.PASSWORD, Consumer.URL);
try {
// 创建连接
connection = connectionFactory.createConnection();
// 启动连接
connection.start();
// 创建session
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
destination = session.createQueue("firstQueue");
MessageConsumer sessionConsumer = session.createConsumer(destination);
while (true) {
TextMessage textMessage = (TextMessage) sessionConsumer.receive();
if (textMessage != null) {
System.out.println("收到的消息:" + textMessage.getText());
} else {
break;
}
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}2.2 使用 Listener 监听方式
Listener监听：
package main.activeMq;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
/**
* 消息监听
*
* @author wangjian
* @date 2020/5/8 23:01
*/
public class Listener implements MessageListener {
@Override
public void onMessage(Message message) {
try {
System.out.println("收到的消息：" + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}消费者：
package main.activeMq;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
/**
* 消费者：监听方式
* @author wangjian
* @date 2020/5/8 23:03
*/
public class Consumer2 {
// 默认的用户名，密码，连接地址
private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
private static final String URL = ActiveMQConnection.DEFAULT_BROKER_URL;
public static void main(String[] args) {
ConnectionFactory connectionFactory; // 连接工厂
Connection connection = null; // 连接
Session session; // 会话：接收或者发送消息的线程
Destination destination; // 消息的目的地
MessageConsumer messageConsumer; // 消息消费者
// 实例化连接工厂
connectionFactory = new ActiveMQConnectionFactory(Consumer2.USERNAME, Consumer2.PASSWORD, Consumer2.URL);
try {
// 创建连接
connection = connectionFactory.createConnection();
// 启动连接
connection.start();
// 创建session
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建消息的连接队列
destination = session.createQueue("firstQueue");
// 创建消息消费者
MessageConsumer sessionConsumer = session.createConsumer(destination);
// 注册消息监听
sessionConsumer.setMessageListener(new Listener());
} catch (JMSException e) {
e.printStackTrace();
}
}
}三.ActiveMQ 发布-订阅消息模式实现
必须先订阅，后发布
destination = session.createTopic("firstTopic"); // 发布订阅