1、JMS介绍
JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。
JMS消息通常有两种类型介绍/实现。ActiveMQ的安全机制。
2、JMS消息通常有两种类型：
① 点对点（Point-to-Point）。在点对点的消息系统中，消息分发给一个单独的使用者。点对点消息往往与队列（javax.jms.Queue）相关联。 ② 发布/订阅（Publish/Subscribe）。发布/订阅消息系统支持一个事件驱动模型，消息生产者和消费者都参与消息的传递。生产者发布事件，而使用者订阅感兴趣的事件，并使用事件。
3、实现两种类型
添加依赖
<!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.14.2</version>
</dependency>
3.1、点对点的实现代码
消息生产者
package com.czy.activemq.test1;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @auther 陈郑游
* @create 2017/4/4 0004
* @功能 消息生产者
* @问题
* @说明
* @URL地址
* @进度描述
*/
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; // 默认的连接地址
public static void main(String[] args) {
// 1、创建连接工厂ConnectionFactory对象、需要填写用户名密码、url
ConnectionFactory connectionFactory;
// 2、通过ConnectionFactory对象给我们的连接对象、并且启动连接。connection默认是关闭的。
Connection connection = null;
// 3、通过Connection对象创建session会话、接受或者发送消息的线程。
Session session;
// 4、通过Session对象创建Destination对象创建连接的消息队列。
Destination destination;
// 5、通过Session对象创建MessageConsumer对象创建消息消费者
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.createQueue("First-Queue"); // 创建消息队列
messageProducer=session.createProducer(destination); // 创建消息生产者
sendMessage(session, messageProducer); // 发送消息
session.commit();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
if (connection != null) {
try {
connection.close();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
/**
* 发送消息
* @param session
* @param messageProducer
* @throws Exception
*/
public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{
int SENDNUM=10; // 发送的消息数量
for(int i=0;i<SENDNUM;i++){
TextMessage message=session.createTextMessage("ActiveMQ 发送的消息"+i);
System.out.println("发送消息："+"ActiveMQ 发送的消息"+i);
messageProducer.send(message);
}
}
}
消费者1（无监听）
package com.czy.activemq.test1;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @auther 陈郑游
* @create 2017/4/4 0004
* @功能 消息消费者
* @问题
* @说明
* @URL地址
* @进度描述
*/
public class JMSConsumer {
//在active.xml文件里头111行
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) {
// 1、创建连接工厂ConnectionFactory对象、需要填写用户名密码、url
ConnectionFactory connectionFactory;
// 2、通过ConnectionFactory对象给我们的连接对象、并且启动连接。connection默认是关闭的。
Connection connection = null;
// 3、通过Connection对象创建session会话、接受或者发送消息的线程。
Session session;
// 4、通过Session对象创建Destination对象创建连接的消息队列。
Destination destination;
// 5、通过Session对象创建MessageConsumer对象创建消息消费者
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.createQueue("First-Queue"); // 创建连接的消息队列
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) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
实现效果
消费者2（使用Listener监听）
package com.czy.activemq.test1;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @auther 陈郑游
* @create 2017/4/4 0004
* @功能 消息消费者\使用 Listener 监听方式
* @问题
* @说明
* @URL地址
* @进度描述
*/
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) {
// 1、创建连接工厂ConnectionFactory对象、需要填写用户名密码、url
ConnectionFactory connectionFactory;
// 2、通过ConnectionFactory对象给我们的连接对象、并且启动连接。connection默认是关闭的。
Connection connection = null;
// 3、通过Connection对象创建session会话、接受或者发送消息的线程。
Session session;
// 4、通过Session对象创建Destination对象创建连接的消息队列。
Destination destination;
// 5、通过Session对象创建MessageConsumer对象创建消息消费者
MessageConsumer messageConsumer;
// 实例化连接工厂
connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);
try {
// 通过连接工厂获取连接
connection=connectionFactory.createConnection();
// 启动连接
connection.start();
// 创建Session
session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 创建连接的消息队列
destination=session.createQueue("First-Queue");
// 创建消息消费者
messageConsumer=session.createConsumer(destination);
// 注册消息监听
messageConsumer.setMessageListener(new Listener());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
消息监听
package com.czy.activemq.test1;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
/**
* @auther 陈郑游
* @create 2017/4/4 0004
* @功能 消息监听
* @问题
* @说明
* @URL地址
* @进度描述
*/
public class Listener implements MessageListener{
@Override
public void onMessage(Message message) {
// TODO Auto-generated method stub
try {
System.out.println("收到的消息："+((TextMessage)message).getText());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
实现效果
3.2、订阅/发布模式的实现代码
消息提供者
package com.czy.activemq.test2;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @auther 陈郑游
* @create 2017/4/4 0004
* 发布-订阅消息模式实现
* @功能 消息生产者-消息发布者
* @问题
* @说明
* @URL地址
* @进度描述
*/
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) {
// 1、创建连接工厂ConnectionFactory对象、需要填写用户名密码、url
ConnectionFactory connectionFactory;
// 2、通过ConnectionFactory对象给我们的连接对象、并且启动连接。connection默认是关闭的。
Connection connection = null;
// 3、通过Connection对象创建session会话、接受或者发送消息的线程。
Session session;
// 4、通过Session对象创建Destination对象创建连接的消息队列。
Destination destination;
// 5、通过Session对象创建MessageConsumer对象创建消息消费者
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("First-Topic");
messageProducer=session.createProducer(destination); // 创建消息生产者
sendMessage(session, messageProducer); // 发送消息
session.commit();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally{
if(connection!=null){
try {
connection.close();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
/**
* 发送消息
* @param session
* @param messageProducer
* @throws Exception
*/
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);
}
}
}
消费者（listener监听）
package com.czy.activemq.test2;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @auther 陈郑游
* @create 2017/4/4 0004
* 发布-订阅消息模式实现
* @功能 消息消费者-消息订阅者一
* @问题
* @说明
* @URL地址
* @进度描述
*/
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) {
// 1、创建连接工厂ConnectionFactory对象、需要填写用户名密码、url
ConnectionFactory connectionFactory;
// 2、通过ConnectionFactory对象给我们的连接对象、并且启动连接。connection默认是关闭的。
Connection connection = null;
// 3、通过Connection对象创建session会话、接受或者发送消息的线程。
Session session;
// 4、通过Session对象创建Destination对象创建连接的消息队列。
Destination destination;
// 5、通过Session对象创建MessageConsumer对象创建消息消费者
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("First-Topic");
messageConsumer=session.createConsumer(destination); // 创建消息消费者
messageConsumer.setMessageListener(new Listener()); // 注册消息监听
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
监听
package com.czy.activemq.test2;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
/**
* 消息监听-订阅者一
* @author Administrator
*
*/
public class Listener implements MessageListener{
@Override
public void onMessage(Message message) {
// TODO Auto-generated method stub
try {
System.out.println("订阅者一收到的消息："+((TextMessage)message).getText());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
4、ActiveMQ的安全机制
第一打开ip:8161/admin/ 就是activemq的管理控制台，它的默认账号和密码都是admin,在生产环境肯定需要更改密码的，这要怎么做呢？
apache-activemq-5.11.1\conf\activemq.xml中找到
上面的代码就得修改了、把下面修改成自己的用户密码。
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; // 默认的连接地址
转载请注明原文地址: https://ju.6miu.com/read-664489.html
