一：ActiveMQ知识整理

Posted 2021-02-16 itzlg

一：JMS概念

JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。JMS是一种与厂商无关的 API，用来访问收发系统消息，它类似于JDBC(Java Database Connectivity)。

JMS百度百科: https://baike.baidu.com/item/JMS/2836691?fr=aladdin

1.体系架构

JMS由以下元素组成：

JMS提供者：连接面向消息中间件的，JMS接口的一个实现。提供者可以是Java平台的JMS实现，也可以是非Java平台的面向消息中间件的适配器。

JMS客户：生产或消费基于消息的Java的应用程序或对象。

JMS生产者：创建并发送消息的JMS客户。

JMS消费者：接收消息的JMS客户。

JMS消息：包括可以在JMS客户之间传递的数据的对象。

JMS队列：一个容纳那些被发送的等待阅读的消息的区域。与队列名字所暗示的意思不同，消息的接受顺序并不一定要与消息的发送顺序相同。一旦一个消息被阅读，该消息将被从队列中移走。

JMS主题：一种支持发送消息给多个订阅者的机制。

2.对象模型

JMS对象模型包含如下几个要素：

1）连接工厂。连接工厂（ConnectionFactory）是由管理员创建，并绑定到JNDI树中。客户端使用JNDI查找连接工厂，然后利用连接工厂创建一个JMS连接。

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

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

4）JMS目的。JMS目的（Destination），又称为消息队列，是实际的消息源。

5）JMS生产者和消费者。生产者（Message Producer）和消费者（Message Consumer）对象由Session对象创建，用于发送和接收消息。

6）JMS消息通常有两种类型：

① 点对点（Point-to-Point）。在点对点的消息系统中，消息分发给一个单独的使用者。点对点消息往往与队列（javax.jms.Queue）相关联。

② 发布/订阅（Publish/Subscribe）。发布/订阅消息系统支持一个事件驱动模型，消息生产者和消费者都参与消息的传递。生产者发布事件，而使用者订阅感兴趣的事件，并使用事件。该类型消息一般与特定的主题（javax.jms.Topic）关联。

3.模型

Java消息服务应用程序结构支持两种模型：

点对点或队列模型 Point-to-Point(P2P)

发布者/订阅者模型 Publish/Subscribe(Pub/Sub)

1.P2P模式图：

2.涉及到的概念

1.  消息队列（Queue）
2.  发送者(Sender)
3.  接收者(Receiver)
4.  每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。

3.P2P的特点

1.每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)
2.发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
3.接收者在成功接收消息之后需向队列应答成功

---

1.Pub/Sub模式图

2.涉及到的概念

主题（Topic）
发布者（Publisher）
订阅者（Subscriber） 
客户端将消息发送到主题。多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。

3.Pub/Sub的特点

每个消息可以有多个消费者。
发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息，而且为了消费消息，订阅者必须保持运行的状态。
为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。

4.传递方式

JMS有两种传递消息的方式。标记为NON_PERSISTENT的消息最多投递一次，而标记为PERSISTENT的消息将使用暂存后再转送的机理投递。如果一个JMS服务离线，那么持久性消息不会丢失但是得等到这个服务恢复联机时才会被传递。所以默认的消息传递方式是非持久性的。即使使用非持久性消息可能降低内务和需要的存储器，并且这种传递方式只有当你不需要接收所有的消息时才使用。

JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。

StreamMessage -- Java原始值的数据流

MapMessage--一套名称-值对

TextMessage--一个字符串对象

ObjectMessage--一个序列化的 Java对象

BytesMessage--一个未解释字节的数据流

二：消息中间件概述

1.MOM概述

MOM（Message Oriented Middleware）是面向消息的中间件，使用消息传送提供者来协调消息传送操作。MOM 需要提供 API 和管理工具。客户端使用api调用，把消息发送到由提供者管理的目的地。在发送消息之后，客户端会继续执行其他工作，并且在接收方收到这个消息确认之前，提供者一直保留该消息。

2.消息中间件产生的背景

在客户端与服务器进行通讯时.客户端调用后，必须等待服务对象完成处理返回结果才能继续执行。

客户与服务器对象的生命周期紧密耦合,客户进程和服务对象进程都都必须正常运行;如果由于服务对象崩溃或者网络故障导致用户的请求不可达,客户会受到异常。

点对点通信:客户的一次调用只发送给某个单独的目标对象。

3.什么是消息中间件

面向消息的中间件(MessageOrlented MiddlewareMOM)较好的解决了以上问题。发送者将消息发送给消息服务器，消息服务器将消息存放在若千队列中，在合适的时候再将消息转发给接收者。

这种模式下，发送和接收是异步的，发送者无需等待; 二者的生命周期未必相同: 发送消息的时候接收者不一定运行，接收消息的时候发送者也不一定运行;一对多通信: 对于一个消息可以有多个接收者。

三：springboot整合activemq

springboot2.0整合activemq： https://www.cnblogs.com/toov5/p/9937989.html

1.引入依赖

<!-- 引入web依赖 -->
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- activemq的springboot启动器 -->
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

2.配置yml文件

spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616
    user: admin
    password: admin
my_queue: springboot-queue
server:
  port: 8080

3.创建QueueConfig类

@Configuration
public class QueueConfig {
    @Value("${my_queue}")
    private String myQueue;
    
    //首先将队列注入到SpringBoot容器中去
    @Bean
    public Queue logQueue() {
        return new ActiveMQQueue(myQueue);
    }
}

4.创建Producer类

@Component
public class Producer {

    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;
    
    //把队列注入进来
    @Autowired  //此注解默认是以类型找  在配置文件中 已经注入的  @Bean
    private Queue myqueue;
    
    //每隔5s时间向队列发送消息
    @Scheduled(fixedDelay=5000)
    public void send() {
        String msg = "测试消息队列"+System.currentTimeMillis();
        System.out.println("msg:"+msg);
        this.jmsMessagingTemplate.convertAndSend(myqueue, msg);
    }
}

5.创建Consumer类

另外新建一个springboot工程，pom文件相同。yaml文件中服务端口号与第一个不同。

spring:
  activemq:
    broker-url: tcp://127.0.0.1:61616
    user: admin
    password: admin
my_queue: springboot-queue
server:
  port: 8081

@Component
public class Consumer {
    //用这个注解去监听 监听的队列
    @JmsListener(destination="${my_queue}")
    public void receive(String msg) {
        System.out.println("消费者接受消息："+msg);
    }   
}