MQ 系列之 ActiveMQ 传输协议

冷不防 2022-12-27 08:47 224阅读 0赞

## 1.1 简介 ##

### 1.1.1 概述 ###

ActiveMQ 支持的 client-broker 通讯协议有：TCP、NIO、UDP、SSL、Http(s)、VM。其中配置 Transport Connector 的文件在 activeMQ 安装目录的 conf/activemq.xml 中的 <transportConnectors> 标签之内。从下图中我们可以发现一个奇怪的东西，描述 amqp 协议的监听端口时，采用的 URI 描述格式为 `amqp://···`，描述 Stomp 协议的监听端口时，采用的 URI 描述格式为 `stomp://···`，唯独在进行 openwire 协议描述时，URI 头却采用的 `tcp://···`。  
  Transmission Control Protocol(TCP)，这是默认的 Broker 配置，TCP 的 Client 监听端口61616。在网络传输数据前，必须要序列化数据，消息是通过一个叫 wire protocol 的来序列化成字节流。默认情况下 ActiveMQ 把 wire protocol 叫做 OpenWire，它的目的是促使网络上的效率和数据快速交互。TCP 连接的 URI 形式为 `tcp://hostname:port?key=value&key=value`，后面的参数是可选。  
![在这里插入图片描述][20201209182202276.png]

### 1.1.2 协议对比 ###

<table> 
 <thead> 
  <tr> 
   <th align="left">协议</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">TCP【常用】</td> 
   <td align="left">默认的协议，性能相对可以</td> 
  </tr> 
  <tr> 
   <td align="left">NIO【常用】</td> 
   <td align="left">基于 TCP 协议之上的，进行了扩展和优化，具有更好的扩展性</td> 
  </tr> 
  <tr> 
   <td align="left">UDP</td> 
   <td align="left">性能比 TCP 更好，但是不具有可靠性</td> 
  </tr> 
  <tr> 
   <td align="left">SSL</td> 
   <td align="left">安全链接</td> 
  </tr> 
  <tr> 
   <td align="left">HTTP(S)</td> 
   <td align="left">基于 HTTP 或者 HTTPS</td> 
  </tr> 
  <tr> 
   <td align="left">VM</td> 
   <td align="left">VM 本身不是协议，当客户端和代理在同一个 Java 虚拟机(VM)中运行时<br>他们之间需要通信，但不想占用网络通道，而是直接通信，可以使用该方式</td> 
  </tr> 
 </tbody> 
</table>

## 1.2 ActiveMQ 优化 ##

### 1.2.1 NIO ###

NIO 协议和 TCP 协议类似，但是 NIO 更侧重于底层的访问操作。它允许开发人员对同一资源可有更多的 client 调用和服务器端有更多的负载。NIO 协议适用于以下场景：  
  ♞ 可能对于 Broker 有一个很迟钝的网络传输，NIO 比 TCP 提供更好的性能  
  ♞ 可能有大量的 Client 去连接到 Broker 上，一般情况下，大量的 Client 去连接 Broker 是被操作系统的线程所限制的。因此，NIO 的实现比 TCP 需要更少的线程去运行，所以建议使用 NIO 协议  
  如果不特别指定 ActiveMQ 的网络监听端口，那么这些端口都将使用 BIO 网络 IO 模型，即：OpenWire、STOMP、AMQP ···，所以为了首先提高单节点的网络吞吐性能，我们需要明确指定 Active 的网络 IO 模型，如下所示：NIO 连接的 URI 形式： `nio//hostname:port?key=value`，表示这个端口使用以 TCP 协议为基础的 NIO 网络 IO 模型。

### 1.2.2 使用 NIO 协议 ###

#### ☞ 修改配置文件，增加 nio ####

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

#### ☞ 编码 ####

/** * Created with IntelliJ IDEA. * * @author Demo_Null * @date 2020/12/9 * @description NIO */
    @SpringJUnitConfig(locations = "classpath:application.properties")
    public class ActiveMQSend { 
        private String URL = "nio://127.0.0.1:61618";
    
        @Test
        public void send() throws JMSException { 
            // 创建工厂对象
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(URL);
            // 由工厂对象创建连接对象
            Connection connection = factory.createConnection();
            // 开启连接
            connection.start();
    
            // 创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 创建消息
            TextMessage textMessage = session.createTextMessage();
            textMessage.setText("我是消息生产者");
    
            // 创建生产者
            Destination destination = session.createQueue("myText");
            MessageProducer producer = session.createProducer(destination);
    
            // 发送消息
            producer.send(textMessage);
    
            // 关闭资源
            producer.close();
            session.close();
            connection.close();
        }
    }

### 1.2.3 AUTO + NIO ###

#### ☞ 概述 ####

URI 格式头以 `nio` 开头，表示这个端口使用以 TCP 协议为基础的 NIO 网络 IO 模型。但是这样的设置方式，只能使这个端口支持 openwire 协议。那么我们怎么既让这个端口支持 NIO 网络 IO 模型，又让它支持多个协议呢？从 5.13.0 版本开始，ActiveMQ 支持 wire format 协议检测，可以自动检测 OpenWire，STOMP，AMQP 和 MQTT，允许为这 4种类型的客户端共享一个传输。要通过 NIO TCP 连接配置 ActiveMQ 自动 wire format 检测，使用 auto+nio 传输前缀。

#### ☞ 配置方式 ####

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

#### ☞ 编码 ####

/** * Created with IntelliJ IDEA. * * @author Demo_Null * @date 2020/12/9 * @description AUTO + NIO */
    @SpringJUnitConfig(locations = "classpath:application.properties")
    public class ActiveMQSend { 
    	// 这个端口可以自动检测多个协议，但是除 nio、tcp 外其他协议的编码方式不同，如下编码使用其他协议会报错
        private String URL = "tcp://127.0.0.1:61608";
    
        @Test
        public void send() throws JMSException { 
            // 创建工厂对象
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(URL);
            // 由工厂对象创建连接对象
            Connection connection = factory.createConnection();
            // 开启连接
            connection.start();
    
            // 创建会话
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 创建消息
            TextMessage textMessage = session.createTextMessage();
            textMessage.setText("我是消息生产者");
    
            // 创建生产者
            Destination destination = session.createQueue("myText");
            MessageProducer producer = session.createProducer(destination)
    
            // 发送消息
            producer.send(textMessage);
    
            // 关闭资源
            producer.close();
            session.close();
            connection.close();
        }
    }

[20201209182202276.png]: /images/20221120/549f5f03c7a74d0bb56bc9daa5cfd1e7.png
[20201209185206839.png]: /images/20221120/511a983b24d34e05ba545ef3afc65229.png
[20201209190708795.png]: /images/20221120/44f0447f05e54e2d8b68af4ac3efcff9.png