J2EE JMS规范概述与ActiveMQ使用案例代码详解

逃离我推掉我的手 2022-03-01 08:47 156阅读 0赞

## 1.J2EE JMS规范 ##

**1.1 JMS( Java Message Service)基本概述**  
作用:用于发送异步消息,在不同模块和不同系统之间实现集群

好处:1降低模块与模块或系统与系统之间耦合度  
2.实现异步通信  
消息机制模型:  
Point to Point : 表现为队列的形式:发送的消息只能被一个人取走  
Topic:可以被多个订阅者订阅,类似于群发  
**1.2 核心接口与类**  
思维导图:  
联系图![在这里插入图片描述][20190322100337479.png] 1.2.1 ConnectionFactory  
用于jms client获取与jms provider连接  
1.2.2 Connection  
表示jms client与jms provider连接,由ConnectionFactory产生  
1.2.3 Session:  
由Connection产生:表示一个会话  
1.2.4  
Message:表示传输  
属性

1.  properties
2.  body(必选)
3.  head

1.2.5 Destination  
消息源,子接口:Queue Topic 用于对应两种消息机制模型  
1.2.6 MessageProducer  
消息发送者

//创建一个Queue，名称为SongLiGuo_FirstQueue
    
    destination = session.createQueue("SongLiGuo_FirstQueue");
    
    //得到消息生产者【发送者】
    
    messageProducer = session.createProducer(destination);

1.2.7 MessageConsumer  
消息接收者  
与MessageProducer相反的组件

## 2 概括 ##

**2.1 ActiveMQ简介**  
能力强劲的开源消息总线,ActiveMQ是Apache的一个开源项目，它是一个能力强劲的开源消息总线，也是一个中间件产品。它是JMS的一个实现。

**2.2 Model**

**Sender.java**

import javax.jms.Connection;
    
    import javax.jms.ConnectionFactory;
    
    import javax.jms.DeliveryMode;
    
    import javax.jms.Destination;
    
    import javax.jms.MessageProducer;
    
    import javax.jms.Session;
    
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Sender {
    
    private static final int SEND_NUMBER = 10;
    
    public static void main(String[] args) {
    
    //ConnectionFactory是连接工厂，JMS用它创建连接
    
    ConnectionFactory connectionFactory;
    
    //Connection JMS客户端到JMS provider的连接
    
    Connection connection = null;
    
    //Session 一个发送或者接收消息的线程
    
    Session session;
    
    //Destination 消息发送目的地，消息发送给谁接收
    
    Destination destination;
    
    //MessageProducer 消息发送者
    
    MessageProducer messageProducer;
    
    //构造ConnectionFactory 实例对象，此处采用ActiveMQ的实现jar
    
    connectionFactory = new ActiveMQConnectionFactory(
    
    ActiveMQConnection.DEFAULT_USER,
    
    ActiveMQConnection.DEFAULT_PASSWORD,
    
    "tcp://localhost:61616");
    
    try {
    
    //构造工厂得到连接对象
    
    connection = connectionFactory.createConnection();
    
    //启动
    
    connection.start();
    
    //获取操作连接
    
    session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    
    //创建一个Queue，名称为SongLiGuo_FirstQueue
    
    destination = session.createQueue("SongLiGuo_FirstQueue");
    
    //得到消息生产者【发送者】
    
    messageProducer = session.createProducer(destination);
    
    //设置不持久化，根据实际情况而定
    
    messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    
    //构造消息，此处写死，项目就是参数或者方法获取
    
    sendMessage(session, messageProducer);
    
    session.commit();
    
    } catch (Exception e) {
    
    e.printStackTrace();
    
    }finally {
    
    try {
    
    if(null != connection){
    
    connection.close();
    
    }
    
    } catch (Throwable ignore) {
    
    }
    
    }
    
    }
    
    public static void sendMessage(Session session, MessageProducer producer)throws Exception {
    
    for (int i = 1; i <= SEND_NUMBER; i++) {
    
    TextMessage message = session.createTextMessage("ActiveMq 发送的消息" + i);
    
    // 发送消息到目的地方
    
    System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
    
    producer.send(message);
    
    }
    
    }
    
    }

**Recevier.java**

import javax.jms.Connection;
    
    import javax.jms.ConnectionFactory;
    
    import javax.jms.Destination;
    
    import javax.jms.MessageConsumer;
    
    import javax.jms.Session;
    
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Receiver {
    
    public static void main(String[] args) {
    
    //connectionFactory 连接工厂，JMS用它创建连接
    
    ConnectionFactory connectionFactory;
    
    //connection JMS客户端到JMS provider 的连接
    
    Connection connection = null;
    
    //session一个发送或者接收的线程
    
    Session session;
    
    //destination 消息目的地，发送给谁接收
    
    Destination destination;
    
    //消费者消息接收者
    
    MessageConsumer consumer;
    
    connectionFactory = new ActiveMQConnectionFactory(
    
    ActiveMQConnection.DEFAULT_USER,
    
    ActiveMQConnection.DEFAULT_PASSWORD,
    
    "tcp://localhost:61616");
    
    try {
    
    //构造工厂得到连接对象
    
    connection = connectionFactory.createConnection();
    
    //启动
    
    connection.start();
    
    //获取操作连接
    
    session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    
    destination = session.createQueue("SongLiGuo_FirstQueue");
    
    consumer = session.createConsumer(destination);
    
    while(true){
    
    //设置接收者收消息的时间，为了方便测试，这里暂定设置为100s
    
    TextMessage message = (TextMessage)consumer.receive(100);
    
    if(null != message){
    
    System.out.println("收到消息==="+message.getText());
    
    }else{
    
    break;
    
    }
    
    }
    
    } catch (Exception e) {
    
    e.printStackTrace();
    
    }finally{
    
    try {
    
    if(null != connection){
    
    connection.close();
    
    }
    
    } catch (Throwable ignore) {
    
    }
    
    }
    
    }
    
    }

注意事项  
项目所引用的jar最后在ActiveMQ下的lib中找

[20190322100337479.png]: /images/20220301/af5e68e1d19a43f1b9d0fb3597a6b828.png