RabbitMQ基础

电玩女神 2022-02-28 05:40 243阅读 0赞

> RabbitMQ是目前非常热门的消息中间件之一。

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

## 命令行 ##

> 命令行工具能方便地进行 RabbitMQ的管理

#### rabbitmq-server ####

*  `rabbitmq-server start` 启动RabbitMQ

#### rabbitmqctl ####

*  `rabbitmqctl reset` 移除所有数据
 *  `rabbitmqctl stop_app` 服务的停止
 *  `rabbitmqctl change_cluster_node_type disc | ram` 修改集群节点的存储形式
 *  `rabbitmqctl forget_cluster_node [--offline]` 忘记节点/节点摘除
 *  `rabbitmqctl rename_cluster_node oldName newName`

#### rabbitmq-plugins ####

*  `rabbitmq-plugins enable rabbitmq_management`

## 消息生产与消费模型 ##

> 这里通过Java代码，演示RabbitMQ最基本的数据流转

### 生产者基础代码 ###

public class Producer { 
    	public static void main(String[] args) throws Exception { 
    		
    		String exchangeName = "exchange_test";
    		String queueName = "test";
    		
    		// 创建一个 ConnectionFactory
    		ConnectionFactory factory = new ConnectionFactory();
    		
    		// 设置RabbitMQ参数
    		factory.setHost("localhost");
    		factory.setPort(5672);
    		factory.setVirtualHost("/");
    		
    		// 获取一个连接
    		Connection connection = factory.newConnection();
    		
    		// 创建Channel
    		Channel channel = connection.createChannel();
    		
    		// 创建一个"direct"类型、非排他、非自动删除的交换机
    		channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true, false, null);
    		
    		// 创建一个持久化、非排他的、非自动删除的队列
    		channel.queueDeclare(queueName, true, false, false, null);
    		
    		channel.queueBind(queueName, exchangeName, "key_test");
    		// 发送数据
    		String message = "Hello world!";
    		channel.basicPublish(exchangeName, "key_test", null, message.getBytes());
    		
    		// 释放资源，关闭连接
    		channel.close();
    		connection.close();
    		
    	}
    }

### 消费者基础代码 ###

public class Consumer { 
    
    	public static void main(String[] args) throws Exception { 
    		// 创建一个 ConnectionFactory
    		ConnectionFactory factory = new ConnectionFactory();
    		
    		// 设置RabbitMQ连接参数
    		factory.setHost("localhost");
    		factory.setPort(5672);
    		factory.setVirtualHost("/");
    		
    		// 获取一个连接
    		Connection connection = factory.newConnection();
    		
    		// 创建信道Channel
    		Channel channel = connection.createChannel();
    		
    		// 声明一个队列
    		String queueName = "test";
    
    		// 创建消费者
    		DefaultConsumer consumer = new DefaultConsumer(channel) { 
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException { 
    				System.out.println("Received message :" + new String(body));
    				try { 
    					TimeUnit.SECONDS.sleep(1);
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    				channel.basicAck(envelope.getDeliveryTag(), false);
    			}
    		};
    		
    		// 获取消息
    		channel.basicConsume(queueName, true, consumer);
    		
    		// 释放资源，关闭连接
    		channel.close();
    		connection.close();
    		
    	}
    }

*  ConnectionFactory：获取连接工厂
 *  Connection：获取一个连接
 *  Channel：得到数据通信信道
 *  Queue：具体的消息存储队列
 *  Producer：消息的生产
 *  Consumer：消费消息

> RabbitMQ的API设计的有些复杂，参数很多，在Java代码中显得凌乱。如果不对其中一些概念有认知，读起来会很吃力。

## 消息队列 Queue ##

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

*  消息队列，实际存储消息数据
 *  Durability：是否持久化，Durable -> 是，Transient -> 否
 *  Auto delete：自动删除，如果最后一个监听被移除以后，是否自动删除该Queue

## 消息 Message ##

*  服务器和应用程序之间传送的数据
 *  本质是一段数据，有Properties和Payload（Body）组成
 *  常用属性：delivery mode、headers（自定义属性）、content\_type、content\_encoding、priority：优先级、correlation\_id（可以当做唯一ID）

## 交换机 Exchange ##

> 接收消息，并根据路由键转发消息所绑定的队列

![交换机][20190324215217851.png]

在实际实现中，RabbitMQ将消息发送给Exchange而不是直接投递到队列中，交换机再把消息路由到队列中。

*  Name：交换机的名称
 *  Type：交换机类型 -> direct、topic、fanout、headers
 *  Durability：是否需要持久化
 *  Auto Delete：当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
 *  Internal：当前Exchange是否用于RabbitMQ内部使用，默认False。在扩展RabbitMQ的功能时才可能用到
 *  Arguments：扩展参数，用于扩展AMQP协议，自定义使用

### 交换机类型 ###

> RabbitMQ常用的交换机类型有四种：fanout、direct、topic、headers。实际上，AMQP协议还规定了其他的几种类型，只是并不常用。

#### direct 直连 ####

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

*  把消息路由到BindingKey和RoutingKey完全匹配的队列中

> 该方式比较严格，实际业务中并不常见

#### topic ####

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

*  所有发送到 Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上
 *  Exchange将RouteKey和某Topic进行模糊匹配，此时队列需要绑定一个Topic

**通配符匹配：**  
\- `#` 匹配一个或多个词：“log.\#” 可以匹配 “[log.info][]”,“log.info.msg”  
\- `*` 匹配一个词：“log.\#” 可以匹配"[log.info][]", “log.error”

> 次方式实际上是`direct`的一种扩展

#### fanout ####

*  不处理路由键，只需要简单的将队列绑定到交换机上
 *  发送到交换机的消息都会被转发到与该交换机绑定的所有队列上
 *  转发消息最快

#### headers ####

*  不依赖路由键的匹配
 *  通过消息头来路由
 *  不常用

> 此种方式局限性比较大，而且性能比较差，很少在业务中需要这种类型

## 绑定 Binding ##

*  Exchange和Exchange、Queue之间的连接关系
 *  Binding中可以包含RoutingKey或者参数

### BingingKey与RoutingKey ###

BingingKey与RoutingKey有时候可以看做是同一种东西。在`direct`模式下，二者必须匹配才能通信。在`topic`模式下，二者有些许差别。

*  BingdingKey可以看成一种特殊的RoutingKey
 *  需要使用绑定的时候，使用`BingdingKey`，如`channel.queueBind`方法
 *  需要发送消息的时候，使用`RoutingKey`，如`channel.basicPublish`方法

> 一般，并不需要刻意地去分清它们

## Channel 信道 ##

> 它是建立在Connection上的一种虚拟连接

它建立在TCP连接中，数据流动都是在Channel中进行的。也就是说，一般情况是程序起始建立TCP连接，第二步就是建立这个Channel。Channel存在的意义重大，对于操作系统来说，建立和关闭TCP连接是有代价的，频繁的建立关闭TCP连接对于系统的性能有很大的影响，而且TCP的连接数也有限制，这也限制了系统处理高并发的能力。但是，在TCP连接中建立Channel是没有上述代价的，RabbitMQ采用类似NIO的做法，使得TCP连接复用，不但减少开销，还便于管理。

> RabbitMQ中的概念比较多，交换机、队列、绑定、路由键。。。它们都是AMQP协议中的组成部分

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9qaWFuZ2NoYW8uYmxvZy5jc2RuLm5ldA_size_16_color_FFFFFF_t_70]: /images/20220228/00b321dfe6d8447b95878ecb72354462.png
[20190324215333351.png]: /images/20220228/839525a9f7134baf8ebdcd7a0aeee164.png
[20190324215217851.png]: /images/20220228/e26c28b307bd49e1a6448ffc6c6236e7.png
[20190324215307177.png]: /images/20220228/b477c1dbb2ac4bb4be66c1cb863ef49b.png
[20190324215317541.png]: /images/20220228/492ee8459f044380a6ae0a237c339e41.png
[log.info]: http://log.info