ActiveMQ-消息中间件

淡淡的烟草味﹌ 2023-10-09 00:58 14阅读 0赞

### 文章目录 ###

*  编程实现\[主题\]
 *   *  生产者
     *  消费者
     *  总结
 *  JMS
 *   *  什么是JMS
     *  JMS的四个组成元素
     *   *  JMS provider
         *  JMS producer
         *  JMS consumer
         *  JMS message
         *   *  消息头
             *  消息体
             *  消息属性
             *  消息的可靠性

# 编程实现\[主题\] #

## 生产者 ##

package cn.oceanstar;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    /**
     * Hello world!
     *
     */
    public class JmsProducer_Topic
    {
        
    
        public static final String USERNAME = ActiveMQConnection.DEFAULT_USER;    //默认连接用户名
        public static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; //默认连接密码
        public static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;     //默认连接地址
    
        public static final String TOPIC_NAME = "topic01";
    
        public static void main(String[] args) throws Exception{
        
            //1、使用默认用户名和密码创建连接工厂
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(JmsProducer_Topic.USERNAME, JmsProducer_Topic.PASSWORD, JmsProducer_Topic.BROKEURL);;
    
            //2、获取connection, 并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
    
            //3、创建会话： 两个参数：事务, 签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //4、创建目的地：主题/队列
            Topic topic = session.createTopic(TOPIC_NAME);
    
            //5、消息生产者
            MessageProducer messageProducer = session.createProducer(topic);
    
            //6、消息生产者生产消息并发送到mq的队列里面
            for (int i=1;i <=3; i++){
        
                //7、创建消息
                TextMessage textMessage = session.createTextMessage("topic:" + i);
    
                //8、发送消息
                messageProducer.send(textMessage);
            }
    
            //9、关闭资源
            messageProducer.close();
            session.close();
            connection.close();
    
            System.out.printf("消息发送完成了。。。。");
        }
    }

## 消费者 ##

package cn.oceanstar;
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    public class JmsConsumer_Topic {
        
        public static final String TOPIC_NAME = "topic01";
    
        public static void main(String[] args) throws Exception {
        
            //1、使用默认用户名和密码创建连接工厂
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(JmsProducer_Topic.USERNAME, JmsProducer_Topic.PASSWORD, JmsProducer_Topic.BROKEURL);;
    
            //2、获取connection, 并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
    
            //3、创建会话： 两个参数：事务, 签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    
            //4、创建目的地：主题/队列
            Topic topic = session.createTopic(TOPIC_NAME);
    
            //5、创建消费者
            MessageConsumer messageConsumer = session.createConsumer(topic);
    
            //------------------------------------------
            //通过监听的方式来消费消息
            messageConsumer.setMessageListener(new MessageListener() {
          //注册监听器
                @Override  //消息到达之后，系统自动调用监听器MessageListener的onMessage
                public void onMessage(Message message) {
        
                    if (null != message && message instanceof TextMessage){
        
                        TextMessage textMessage = (TextMessage)message;
                        try {
        
                            System.out.println("消费者收到消息：" + textMessage.getText());
                        } catch (JMSException e) {
        
                            e.printStackTrace();
                        }
                    }
                }
            });
    
    
            System.in.read();
            //--------------------
    
    
            messageConsumer.close();
            session.close();
            connection.close();
        }
    
    }

## 总结 ##

*  生产者将消息发布到topic中，每个消息可以有多个消费者，以实一对多
 *  生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息。
 *  生产者生成时，topic不保存消息，它是无状态的不落地，如果无人订阅就去生成，就是一天废消息，所以，一般先启动消费者在启动生产者。

JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。

![在这里插入图片描述][20190925140416907.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]

# JMS #

## 什么是JMS ##

jms是JaveEE中的一个子规范，又叫java消息服务。  
java消息服务指的是两个应用程序之间进行异步通信的API，它为标准消息协议和消息服务提供了一组通用接口，包括创建，发送，读取消息等，用于支持Java应用程序开发。在JavaEE中，当两个应用程序使用JMS进行通信时，它们之间不是直接相连的，而是通过一个共同的消息收发服务器组件关联起来以达到解耦/异步削峰的效果。

备注：什么是JavaEE  
JavaEE是一套使用java进行企业级应用开发的大家一致遵循的13个核心规范工业标准。JavaEE平台提供了一个基于组件来加快设计，开发，装配以及部署企业应用程序。包括JDBC(数据库连接)，JNDI，EJB，RMI，JSP，Servlet，XML，JMS(Java消息服务)，JTA(Java事务API)，JavaMail，JAF这13个标准

![在这里插入图片描述][20190925160537692.png]

## JMS的四个组成元素 ##

### JMS provider ###

实现jms接口和规范的消息中间件，即MQ服务器

### JMS producer ###

消息生产者，创建和发送JMS消息的客户端应用

### JMS consumer ###

消息消费者，接受和处理JMS消息的客户端应用

### JMS message ###

#### 消息头 ####

*  JMSDestination：消息发送的目的地：Queue/Topic
 *  JMSDeliveryMode：
    
     *  持久模式：如果JMS提供者出现故障，该消息并不会丢失，他会在服务器恢复之后再次传递
     *  非持久模式：最大传送一次，这意味着如果服务器出现故障，该消息将永久丢失
 *  JMSExpiration：可以设置消息在一定时间后过期，默认是永不过期。
    
     *  消息过期时间，等于Destination中的send方法中的timeToLive值加上1发送时刻的GMT时间值
     *  如果timeToLive等于0，则JMSExpiration被设为0，表示该消息永不过期
     *  如果发送后，在消息过期时间之后消息还没有发被发送到目的地，则该消息被清除。
 *  JMSPriority：消息优先级，从0到9十个级别，0到4是普通消息，5到9是加急消息。JMS不要求MQ严格按照这十个优先级发送消息，但是必须保证加急消息要先于普通消息到达。默认是4级。
 *  JMSMessageID：唯一识别每个消息的id，由MQ产生

#### 消息体 ####

用来封装具体的消息数据，一共有5种消息体格式，其发送和接受的消息体类型必须一致对应。

*  TextMessage：普通的字符串消息，包含一个string
 *  MapMessage：一个Map类型的消息，key为string类型，而值为java的基本类型
 *  BytesMessage：二进制数组消息，包含一个Byte\[\]
 *  StreamMessage：Java数据流消息，用标准流操作来顺序的填充和读取
 *  ObjectMessage：对象消息，包含一个可序列化的Java对象。

#### 消息属性 ####

用于识别，去重，重点标注等。比如

textMessage.setStringProperty("c01", "vip");

textMessage.getStringProperty("c01");

#### 消息的可靠性 ####

[20190925140416907.png]: https://img-blog.csdnimg.cn/20190925140416907.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190925143820182.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg==,size_16,color_FFFFFF,t_70
[20190925160537692.png]: https://img-blog.csdnimg.cn/20190925160537692.png