【ActiveMQ笔记二】JMS消息模型

我就是我 2023-06-19 02:30 32阅读 0赞

# JMS和ActiveMQ #

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

## JMS规范 ##

### JMS对象模型 ###

1.  连接工厂。连接工厂负责创建一个JMS连接。
2.  JMS连接。JMS连接（Connection）表示JMS客户端和服务器端之间的一个活动的连接，是由客户端通过调用连接工厂的方法建立的。
3.  JMS会话。JMS会话（Session）表示JMS客户与JMS服务器之间的会话状态。JMS会话建立在JMS连接上，表示客户与服务器之间的一个会话线程。
4.  JMS目的/ Broker。客户用来指定它生产的消息的目标和它消费的消息的来源的对象，一个消息中间件的实例。
5.  JMS生产者和消费者。生产者（Message Producer）和消费者（Message Consumer）对象由Session对象创建，用于发送和接收消息。

#### 消息的消费可以采用以下两种方法之一： ####

*  同步消费。通过调用消费者的receive 方法从目的地中显式提取消息。receive方法可以一直阻塞到消息到达。
 *  异步消费。客户可以为消费者注册一个消息监听器，以定义在消息到达时所采取的动作。  
    **springboot中配置：**

@Bean
        public ConnectionFactory connectionFactory(){ 
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
            connectionFactory.setBrokerURL(brokerUrl);
            // 配置消费的同步或者异步,默认是异步的
            // Enables or disables the default setting of whether or not 
            // consumers have their messages synchronously or asynchronously by the broker
            connectionFactory.setDispatchAsync(true);
            return connectionFactory;
        }

### JMS规范中的消息 ###

JMS 消息由以下三部分组成：

*  消息头。每个消息头字段都有相应的getter 和setter 方法。
 *  消息属性。如果需要除消息头字段以外的值，那么可以使用消息属性。
 *  消息体。JMS 定义的消息类型有TextMessage、MapMessage、BytesMessage、StreamMessage 和 ObjectMessage。ActiveMQ也有对应的实现。具体实现参考`ActiveMQSession`

jmsTemplate.send(destination, session -> session.createTextMessage(time + ""));

### JMS消息模型 ###

#### Point-to-Point(P2P) / 点对点 ####

消息通过称为队列的一个虚拟通道来进行交换。队列是生产者发送消息的目的地和接受者消费消息的消息源。  
每条消息通仅会传送给一个接受者。可能会有多个接受者在一个队列中侦听，但是每个队列中的消息只能被队列中的一个接受者消费。  
 消息存在先后顺序。一个队列会按照消息服务器将消息放入队列中的顺序，把它们传送给消费者当消息已被消费时，就会从队列头部将它们删除。  
 每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)  
 发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列  
 接收者在成功接收消息之后需向队列应答成功  
 如果希望发送的每个消息都应该被成功处理的话，使用这个P2P模式。

#### Topic/ 主题（发布订阅(Pub/Sub) ） ####

消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic的消息会被所有订阅者消费。  
 如果你希望发送的消息可以不被做任何处理、或者被一个消息者处理、或者可以被多个消费者处理的话，那么可以采用topic模型

#### spring-boot配置消息模型 ####

1.  `spring-boot-starter-activemq`通过`spring.jms.pub-sub-domain`来控制的。 此控制是全局的。
2.  通过javaConfig方式，这种方式，可以针对某一个destination单独定制

// 定义一个 JmsListenerContainerFactory
        @Bean("jmsTopicListenerContainerFactory")
        public JmsListenerContainerFactory jmsTopicListenerContainerFactory(
                ConnectionFactory connectionFactory
        ) { 
            DefaultJmsListenerContainerFactory factory
                    = new DefaultJmsListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setPubSubDomain(true);
            return factory;
        }

**消费者**  
通过`@JmsListener`添加消费者

@JmsListener(destination = "queueA")
        public void queue1(String time) { 
            System.out.println("recive1 message from queueA [" + time + "]");
        }
    
        @JmsListener(destination = "queueB",containerFactory = "jmsTopicListenerContainerFactory")
        public void queue2(String time) { 
            System.out.println("recive2 message from queueB [" + time + "]");
        }
    
        @JmsListener(destination = "queueB",containerFactory = "jmsTopicListenerContainerFactory")
        public void queue3(String time) { 
            System.out.println("recive3 message from queueB [" + time + "]");
        }
    
        @JmsListener(destination = "queueB")
        public void queue4(String time) { 
            System.out.println("recive4 message from queueB [" + time + "]");
        }

**生产者**

@Test
        public void testA(){ 
            for (int i = 0; i < 2; i++) { 
                long time = System.currentTimeMillis();
                System.out.println("send --> " + time);
                // 如果不指定Destination是Queue还是Topic，
                // 则根据系统的spring.jms.pub-sub-domain配置
                jmsTemplate.convertAndSend("queueA", time);
    
                // 指定Queue
                Destination queueB = new ActiveMQQueue("queueB");
                jmsTemplate.convertAndSend(queueB, time);
    
                // 指定Topic
                Destination destination = new ActiveMQTopic("queueB");
                jmsTemplate.convertAndSend(destination, time);
                
    			// 执行message类型
                jmsTemplate.send(destination, session -> session.createTextMessage(time + ""));
    
            }
        }

关于 `jmsTemplate.convertAndSend("queueA", time);`发送消息模型的源码解释，确认发送那种模型

@Override
    	public Destination resolveDestinationName(
    					@Nullable Session session, 
    					String destinationName, 
    					boolean pubSubDomain)
    			throws JMSException { 
    
    		Assert.notNull(session, "Session must not be null");
    		Assert.notNull(destinationName, "Destination name must not be null");
    		if (pubSubDomain) { 
    			return resolveTopic(session, destinationName);
    		}
    		else { 
    			return resolveQueue(session, destinationName);
    		}
    	}