ActiveMQ消息中间件之JMS概念及用法(二)

末蓝、 2022-04-13 15:17 198阅读 0赞

JMS（Java Messageing Service）是Java平台面向消息中间件的技术规范，实际上是一套API，便于消息系统中Java应用程序进行消息交互，并且通过提供标准的产生，发送、接收消息的接口简化企业开发，ActiveMQ是这规范的一个具体实现。

JMS规范模型：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTc0Mzg5_size_16_color_FFFFFF_t_70][]

1）、连接工厂：负责创建一个JMS连接

2）、JMS连接：表示JMS客户端与服务器端之间的一个活动的连接，是由客户端通过调用连接工厂方法建立连接

3）、JMS会话：表示JMS客户端与JMS服务端的会话状态。JMS会话建立在JMS连接上，表示客户端与服务器的一个会话线程。

4）、JMS目的/ Broker：客户用来指定它生产的目标和它消费的消息来源对象 ，一个消息中间件的实例。

5）、JMS生产者和消费者：生产者和消费者是有对象创建的session 队列创建，用于发送、接收消息。

消息的消费可以采取以下两种方式：

同步方式：通过调用消费者的receive(接收)方法从目的地中提取消息，receive(接收)方法可以一直堵塞到消息到达

异步方式：客户可以为消费者注册一个消息监听器，以定义在消息到达时采取的动作

JMS规范中的消息：

消息头：每个消息头字段都有对应的getter和setter方法。

消息属性：如果需要除消息头的字段的话，那么久可以使用消息属性。

消息体：JMS定义的消息类型有TextMassage、MapMassage、BytesMassage、StreamMessage和ObjectMassage。ActiveMQ也有对应的实现。

## JMS消息模型： ##

###    Point-to-Point（P2P）/ 点对点 ###

消息通过队里的一个虚拟通道来进行交换。队列是生产者发送消息的目的地和接受者消费消息的消息源。

每个消息仅会传送给一个接受者。可能会有多个接受者在一队列中侦听，但是每个队列中的消息只能被队列中的一个接受者消费

消费存在先后顺序，一个队列会按照消息服务器将消息放入队列中的顺序，把它们传送给消费者，当消费者将消息消费时，将会从队列中删除。（每个消息只有一个消费者，一旦被消费，消息就不会出现在消息队列中）

发送者发送了消息之后，不管接受者有没有正常运行，都不会一影响消息被发送到队列，接受者在成功接收消息之后需向队列返回是否成功（类似于Java中的事物管理，后面咱在细说），如果希望发送的每个消息都应该被成功处理的话，就是用这个P2P模式。

### Topic / 主题（发布订阅（pub / Sub ）） ###

1、消息生产者（发布）将消息发送到Topic中，同时又多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic的消息会被所有订阅者消费。

2、如果希望发送的消息可以不被做任何处理、或者被一个消息者处理、或者可以被多个消费者处理的话，那么就可以采用topic模型

## 使用ActiveMQ： ##

###       原生API写法（中间包含队列模式，topic模式） ###

先引入Jar包

<dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-all</artifactId>
        <version>5.8.0</version>
    </dependency>

0、先启动ActiveMQ（我是直接用本地Win启动的）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTc0Mzg5_size_16_color_FFFFFF_t_70 1][]

1、创建JmsProducer生产者端

import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * 类说明：生产者端
     */
    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 = 3;
    
        public static void main(String[] args) {
            /* 连接工厂*/
            ConnectionFactory connectionFactory;
            /* 连接*/
            Connection connection = null;
            /* 会话*/
            Session session;
            /* 消息的目的地*/
            Destination destination;
            /* 消息的生产者*/
            MessageProducer messageProducer;
    
            /* 实例化连接工厂*/
            connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
            try {
                /* 通过连接工厂获取连接*/
                connection = connectionFactory.createConnection();
                /* 启动连接*/
                connection.start();
                /* 创建session
                 * 第一个参数表示是否使用事务，第二次参数表示是否自动确认*/
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                /* 创建一个名为HelloWorld消息队列*/
                destination = session.createQueue("HelloActiveMqQueue");
    //          destination = session.createTopic("HelloActiveMq"); //topic模式
               
                /* 创建消息生产者*/
                messageProducer = session.createProducer(destination);
                /* 循环发送消息*/
                for (int i = 0; i < SENDNUM; i++) {
                    String msg = "发送消息" + i + " " + System.currentTimeMillis();
                    TextMessage textMessage = session.createTextMessage(msg);
                    System.out.println("标准用法:" + msg);
                    messageProducer.send(textMessage);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
    
            }
        }
    
    
    }

2、创建JmsConsumer消费者端

import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * 类说明：消费者端
     */
    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(USERNAME, PASSWORD, BROKEURL);
    
            try {
                /* 通过连接工厂获取连接*/
                connection = connectionFactory.createConnection();
                /* 启动连接*/
                connection.start();
                /* 创建session*/
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                /* 创建一个名为HelloWorld消息队列*/
                destination = session.createQueue("HelloActiveMqQueue");
    //            destination = session.createTopic("HelloActiveMq");    //topic模式
                /* 创建消息消费者*/
                messageConsumer = session.createConsumer(destination);
                Message message;
                while ((message = messageConsumer.receive()) != null) {
                    System.out.println(((TextMessage) message).getText());
                }
    
    
                //Tibco EMS zeroQ,
    
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }
    }

3、启动后就可以再控制台看到交互的数据了，另外打开[http://localhost:8161/admin/][http_localhost_8161_admin]管理控制台可以看到ActiceMQ的运行情况及队列情况，如图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTc0Mzg5_size_16_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTc0Mzg5_size_16_color_FFFFFF_t_70]: /images/20220413/b563a56744494403bc2daa666a0ad391.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTc0Mzg5_size_16_color_FFFFFF_t_70 1]: /images/20220413/8ce3abc348234e2f8322bbfdaf2edec6.png
[http_localhost_8161_admin]: http://localhost:8161/admin/queues.jsp
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NTc0Mzg5_size_16_color_FFFFFF_t_70 2]: /images/20220413/e0e933d80ff84fdeac56f6e0551d68dd.png