JMS之ActiveMQ基本应用

一、JMS：

       概念：

       JMS即Java Message Service（Java消息服务）应用程序接口，主要用于两个应用程序或分布式系统中的异步通信，JMS就像JDBC一样是一个Java平台的技术规范，JDBC就是用来访问不同关系数据库的API，而JMS就是用来进行消息收发的API。

       体系架构：

       1、JMS提供者：实现JMS规范的产品，下面讲的ActiveMQ就是其中一个，当然除了ActiveMQ外还有很多其他的提供者。

       2、JMS客户：生产或消费消息的Java应用程序或对象。

       3、JMS生产者：创建并发送消息的JMS客户。

       4、JMS消费者：接收消息的JMS客户。

       5、JMS消息：JMS客户之间传递的数据对象，消息类型可以为多种类型。

       6、JMS队列：由于消息被生产者发送之后不会被消费者立即消费掉，所以需要先将消息放在JMS队列里等待消费者收取，消息被消费者消费之后就会从队列中移除。

       7、JMS主题：支持发送消息给多个订阅者的机制，简单说就是生产者将消息发送到主题之后，主题可以将每个消息发给每个消费者。

       对象模型：

       1、连接工厂（ConnectionFactory）：用来创建JMS连接。

       2、JMS连接（Connection）：JMS客户端与消息服务器之间的连接，即JMS客户端和消息服务器之间通信的高速公路。

       3、JMS会话（Session）：即JMS客户端和消息服务器之间的一个线程，有了高速公路还需要有一辆卡车将货物运送过去啊。需要注意在实际的应用程序中会话可以支持事务，当选择支持事务的情况下，消息会先存放在会话中，当事务提交时会将消息一并发送，提交之前可以进行回滚操作。

       4、JMS目的（Destination）：消息发布和接收的地点，可以是队列或主题，至于是队列还是主题要看消息通信的模型了（点对点模型与发布订阅模型），下边说明。

       5、JMS生产者和消费者（MessageProducter，MessageConsumer）：消息发送和接收的客户端对象。

       JMS消息通信模型：

       1、点对点：

       一个生产者向特定队列发送消息，一个消费者从该队列收取消息。该模式下生产者无需在消费者消费消息的情况下处于运行状态，消费者也无需在生产者发送消息的情况下处于运行状态。

       2、发布订阅：

       一个生产者向特定主题发送消息，0个或多个消费者从该主题获取消息，多个消费者之间的消费互不影响。该模式下发布和订阅者之间存在时间依赖性，订阅者需要先订阅对应主题的消息，否则将无法获取消息，也就是说你今天订阅人民日报的报纸，那么之前的报纸消息你是无法获取到的。

 

二、ActiveMQ

       上边说了ActiveMQ是JMS规范的一个实现，那么也自然符合上边的相关体系和模型。下边通过代码看ActiveMQ的实现过程。

       点对点模型

       1、消息生产者：

/**
* 消息生产者
*/
public class JMSProducer {

private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
private static final String BROKEURL=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; // 消息生产者
// 实例化连接工厂
connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
try {
connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
connection.start(); // 启动连接
session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建Session，参数为true表示开启事务
destination=session.createQueue("Queue1"); // 创建消息队列
messageProducer=session.createProducer(destination); // 创建消息生产者
sendMessage(session, messageProducer); // 发送消息
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally{
if(connection!=null){
try {
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}

/**
* 发送消息
*/
public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{
for(int i=0;i<JMSProducer.SENDNUM;i++){
TextMessage message=session.createTextMessage("ActiveMQ 发送的消息"+i);
messageProducer.send(message); //消息生产者发送消息
}
}
}

       2、消费者基本用法：

 /**
* 消息消费者
*/
public class JMSConsumer {

private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址

public static void main(String[] args) {
ConnectionFactory connectionFactory; // 连接工厂
Connection connection = null; // 连接
Session session; // 会话 接受或者发送消息的线程
Destination destination; // 消息的目的地
MessageConsumer messageConsumer; // 消息的消费者

// 实例化连接工厂
connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
try {
connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
connection.start(); // 启动连接
session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session，参数为false表示不开启事务
destination=session.createQueue("Queue1"); // 创建连接的消息队列
messageConsumer=session.createConsumer(destination); // 创建消息消费者
//定时从队列中收取消息
while(true){
TextMessage textMessage=(TextMessage)messageConsumer.receive(100000);
if(textMessage!=null){
System.out.println("收到的消息为："+textMessage.getText());
}else{
break;
}
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
       3、消费者常用用法：

       在上边的实现中消费者不断的从消息队列中收取消息，这样太耗费资源，通过我们会给消费者注册一个监听器，当监听到有新的消息时就通知消费者消费。

 /**
* 消息消费者
*/
public class JMSConsumer2 {

private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址

public static void main(String[] args) {
ConnectionFactory connectionFactory; // 连接工厂
Connection connection = null; // 连接
Session session; // 会话 接受或者发送消息的线程
Destination destination; // 消息的目的地
MessageConsumer messageConsumer; // 消息的消费者

// 实例化连接工厂
connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);
try {
connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
connection.start(); // 启动连接
session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
destination=session.createQueue("Queue1"); // 创建连接的消息队列
messageConsumer=session.createConsumer(destination); // 创建消息消费者
messageConsumer.setMessageListener(new Listener()); // 注册消息监听
} catch (JMSException e) {
e.printStackTrace();
}
}
}

/**
* 消息监听
*/
public class Listener implements MessageListener{

@Override
public void onMessage(Message message) {
try {
System.out.println("收到的消息："+((TextMessage)message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}

       订阅模式

       1、消息发布者

/**
* 消息生产者-消息发布者
*/
public class JMSProducer {

private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
private static final String BROKEURL=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; // 消息生产者

// 实例化连接工厂
connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
try {
connection=connectionFactory.createConnection(); //通过连接工厂获取连接
connection.start(); //启动连接
session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); //创建Session
destination=session.createTopic("Topic1"); //这里创建的为消息主题
messageProducer=session.createProducer(destination); //创建消息生产者
sendMessage(session, messageProducer); //发送消息
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally{
if(connection!=null){
try {
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}

/**
* 发送消息
*/
public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{
for(int i=0;i<JMSProducer.SENDNUM;i++){
TextMessage message=session.createTextMessage("ActiveMQ 发送的消息"+i);
System.out.println("发送的消息："+"ActiveMQ 发布的消息"+i);
messageProducer.send(message); //发送消息
}
}
}
       2、消息订阅者一

/**
* 消息消费者-消息订阅者一
*/
public class JMSConsumer {

private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址

public static void main(String[] args) {
ConnectionFactory connectionFactory; // 连接工厂
Connection connection = null; // 连接
Session session; // 会话 接受或者发送消息的线程
Destination destination; // 消息的目的地
MessageConsumer messageConsumer; // 消息的消费者

// 实例化连接工厂
connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
try {
connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
connection.start(); // 启动连接
session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
destination=session.createTopic("Topic1"); //在这里创建的为消息主题
messageConsumer=session.createConsumer(destination); // 创建消息消费者
messageConsumer.setMessageListener(new Listener()); // 注册消息监听
} catch (JMSException e) {
e.printStackTrace();
}
}
}

/**
* 消息监听-订阅者一
*/
public class Listener implements MessageListener{

@Override
public void onMessage(Message message) {
try {
System.out.println("订阅者一收到的消息："+((TextMessage)message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}

       3、消息订阅者二

/**
* 消息消费者-消息订阅者二
*/
public class JMSConsumer2 {

private static final String USERNAME=ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
private static final String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址

public static void main(String[] args) {
ConnectionFactory connectionFactory; // 连接工厂
Connection connection = null; // 连接
Session session; // 会话 接受或者发送消息的线程
Destination destination; // 消息的目的地
MessageConsumer messageConsumer; // 消息的消费者

// 实例化连接工厂
connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);
try {
connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
connection.start(); // 启动连接
session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
destination=session.createTopic("Topic1");
messageConsumer=session.createConsumer(destination); // 创建消息消费者
messageConsumer.setMessageListener(new Listener2()); // 注册消息监听
} catch (JMSException e) {
e.printStackTrace();
}
}
}

/**
* 消息监听-订阅者二
*/
public class Listener2 implements MessageListener{

@Override
public void onMessage(Message message) {
try {
System.out.println("订阅者二收到的消息："+((TextMessage)message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}