MQ（消息中间件）概述及 RabbitMQ 的基本介绍

男娘i 2023-10-13 10:28 7阅读 0赞

# 概述 #

## 消息中间件（MQ） ##

*  消息队列中间件是分布式系统中重要的组件，主要解决**应用解耦，异步消息，流量削锋等**问题，实现高性能，高可用，可伸缩和最终一致性架构。流量削锋 ： 削减峰值压力（秒杀，抢购）
 *  MQ（Message Queue，消息队列）是典型的生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。
 *  生产者：向消息中间件发送消息的一方
    
    消费者：从中间获取消息使用的一方
    
    broker： 消息中间件
 *  常见的消息中间件：
    
     *  RabbitMQ：基于 AMQP 协议，erlang 语言开发，稳定性好
        
        一款开源的消息管理系统，支持主流的操作系统，Linux、Windows、MacOX 等，支持多种开发语言，Java、Python、Ruby、.NET、PHP、C/C++、node.js 等
     *  RocketMQ：基于 JMS，阿里巴巴产品，目前交由 Apache 基金会
     *  ActiveMQ：基于JMS
     *  kafka：分布式消息系统，高吞吐量

## AMQP 和 JMS 标准 ##

实现 MQ 有两种主流方式：AMQP、JMS。

*  AMQP，即 Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。
    
    基于此协议的客户端与消息中间件可传递消息，并不受客户端中间件不同产品，不同的开发语言等条件的限制。
 *  JMS，通常而言提到 JMS（Java MessageService）实际上是指 JMS API。JMS 是由 Sun 公司早期提出的消息标准，旨在为 Java 应用提供统一的消息操作，包括 create、send、receive等。
    
    JMS 已经成为 java Enterprise Edition 的一部分。从使用角度看，JMS 和 JDBC 担任差不多的角色，用户都是根据相应的接口可以和实现了 JMS 的服务进行通信，进行相关的操作。

两者间的区别和联系：

*  JMS 是定义了统一的接口，来对消息操作进行统一
    
    AMQP 是通过规定协议来统一数据交互的格式
 *  JMS 限定了必须使用 Java 语言
    
    AMQP 只是协议，不规定实现方式，因此是跨语言的
 *  JMS 规定了两种消息模型 点对点、发布订阅
    
    AMQP 的消息模型更加丰富（7种），工作中常用的有 5 种

# RabbitMQ 简述 #

## 基本概念 ##

*  **Broker**：表示消息队列服务器实体
 *  **Message**：消息，消息是不具名的，它由消息头和消息体组成。
    
    消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。
 *  **Publisher**：消息的生产者，也是一个向交换器发布消息的客户端应用程序。
 *  **Virtual Host**：虚拟主机，表示一批交换器、消息队列和相关对象。
    
    虚拟主机是共享相同的身份认证和加密环境的独立服务器域。
    
    每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。
    
    vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 /
 *  **Exchange**：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
 *  **Binding**：绑定，用于消息队列和交换器之间的关联。
    
    一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
 *  **Queue**：消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。
    
    一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
 *  **Connection**：网络连接，比如一个 TCP 连接。
 *  **Channel**：信道，多路复用连接中的一条独立的双向数据流通道。
    
    信道是建立在真实的 TCP 连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。
    
    因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
 *  **Consumer**：消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

## 消息模型 ##

RabbitMQ 提供了 6 种消息模型，但是第 6 种其实是 RPC，并不是MQ。其中 3、4、5 这三种都属于订阅模型，只不过进行路由的方式不同。

![在这里插入图片描述][ad0933ceaf104ec7b60ca63e7c2d5f2d.png_pic_center]

### 基本消息模型（直连） ###

图示：

![在这里插入图片描述][9f208a2af4cf4fa6a871dd792e72b95f.png_pic_center]

*  P：生产者，也就是要发送消息的程序
 *  C：消费者，即消息的接受者，会一直等待消息到来。
 *  图中红色部分：queue（消息队列），类似一个邮箱，可以缓存消息；生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。

**示例代码**

*  生产者：
    
        // 连接工具类
        public class ConnectionUtil {
              
            // 建立与RabbitMQ的连接
            public static Connection getConnection() throws Exception {
              
                //定义连接工厂
                ConnectionFactory factory = new ConnectionFactory();
                //设置服务地址
                factory.setHost("127.0.0.1");
                //端口
                factory.setPort(5672);
                //设置账号信息，用户名、密码、vhost
                factory.setVirtualHost("/saas");
                factory.setUsername("saas");
                factory.setPassword("saas");
                // 通过工程获取连接
                Connection connection = factory.newConnection();
                return connection;
            }
        }
    
    生产者发送消息：
    
        public class Send {
              
        
            private final static String QUEUE_NAME = "simple_queue";
        
            public static void main(String[] argv) throws Exception {
              
                // 获取到连接
                Connection connection = ConnectionUtil.getConnection();
                // 从连接中创建通道，使用通道才能完成消息相关的操作
                Channel channel = connection.createChannel();
                // 声明（创建）队列
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                // 消息内容
                String message = "Hello World!";
                // 向指定的队列中发送消息
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                
                System.out.println(" [x] Sent '" + message + "'");
        
                //关闭通道和连接
                channel.close();
                connection.close();
            }
        }

### 工作（任务）模型 ###

工作模型，Work queues，也被称为（Task queues，任务模型） 。 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多，无法及时处理。此时就可以使用 work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

![在这里插入图片描述][7b717214b6744dbe86d71cff5e864a10.png_pic_center]

默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。

缺点：当其中一个消费比较慢，比如一个消费者已经消费完。但是另一个消费者还在一个个的接收。这样会造成消息的积累。

优化方案：

*  消费者：要关闭自动确认消息
 *  当前通道每一次只能消费一个消息
    
        // 每次只能消费一个消息|一次只接收一条未确认的消息
        channel.basicQos(1);
 *  消息确认：手动确认消息

**示例代码：**

*  生产者
    
        public class Provider {
              
            private static final String TASK_QUEUE_NAME = "task_queue";
         
            public static void main(String[] argv) throws Exception {
              
                // 获取连接对象
                Connection connection = RabbitMQUtils.getConnection();
                // 获取连接中的通道对象
                Channel channel = connection.createChannel();
                // 通过通道声明队列
                channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
                // 发送10条消息
                for (int i = 0; i < 20; i++) {
              
                    channel.basicPublish("", TASK_QUEUE_NAME, null, (i+"你好啊").getBytes());
                }
            }
        }
 *  消费者（创建2个，区别是是否带下面这个线程休眠代码）
    
        public class Customer2 {
              
            private static final String TASK_QUEUE_NAME = "task_queue";
         
            public static void main(String[] args) throws IOException {
              
                // 获取连接对象
                Connection connection = RabbitMQUtils.getConnection();
                // 获取连接中的通道对象
                Channel channel = connection.createChannel();
                // 每次只能消费一个消息
                channel.basicQos(1);
                // 通道绑定消息队列
                channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
                // 获取消息  参数2：消息自动确认，消费者自动向rabbitmq确认消息消费（只要消息队列有消息，就分配给消费者）
                channel.basicConsume(TASK_QUEUE_NAME, false, new DefaultConsumer(channel) {
              
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              
                        System.out.println("消费者-1：" + new String(body));
                        try {
              
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
              
                            e.printStackTrace();
                        }
                        // 参数1：确认队列中那个具体消息 参数2：是否开启多个消息同时确认
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                });
            }
        }

### 发布订阅模型 ###

![在这里插入图片描述][886c81e9089d4273985e54731470d95a.png_pic_center]

*  1个生产者，多个消费者
 *  每一个消费者都有自己的一个队列
 *  每个队列都要绑定到交换机
 *  生产者没有将消息直接发送到队列，而是发送到了交换机
 *  生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

X（Exchanges）：交换机。一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于 Exchange 的类型。Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

Exchange 类型有以下几种：

*  **Fanout**：广播，将消息交给所有绑定到交换机的队列
 *  **Direct**：定向，把消息交给符合指定 routing key 的队列
 *  **Topic**：通配符，把消息交给符合 routing pattern（路由模式）的队列

### 订阅模型-Fanout（广播）类型 ###

广播模式的消息发送流程：

*  可以有多个消费者
 *  每个消费者有自己的 queue（队列）
 *  每个队列都要绑定到 Exchange（交换机）
 *  生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
 *  交换机把消息发送给绑定过的所有队列
 *  队列的消费者都能拿到消息。实现一条消息被多个消费者消费

**示例代码：**

*  生产者
    
        public class Provider {
              
            public static void main(String[] args) throws IOException {
              
                // 获取连接对象
                Connection connection = RabbitMQUtils.getConnection();
                // 获取连接中的通道对象
                Channel channel = connection.createChannel();
         
                // 将通道声明指定交换机 参数1：交换机名称 参数2：交换机类型，fanout是广播类型
                channel.exchangeDeclare("logs","fanout");
                // 发送消息
                channel.basicPublish("logs","",null,"fanout type message".getBytes());
            }
        }
 *  消费者
    
        public class Customer1 {
              
            public static void main(String[] args) throws IOException {
              
                Connection connection = RabbitMQUtils.getConnection();
         
                Channel channel = connection.createChannel();
                //通道绑定交换机
                channel.exchangeDeclare("logs","fanout");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 绑定交换机和队列
                channel.queueBind(queue,"logs","");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){
              
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              
                        System.out.println("消费者1"+new String(body));
                    }
                });
            }
        }
    
        public class Customer2 {
              
            public static void main(String[] args) throws IOException {
              
                Connection connection = RabbitMQUtils.getConnection();
         
                Channel channel = connection.createChannel();
                //通道绑定交换机
                channel.exchangeDeclare("logs","fanout");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 绑定交换机和队列
                channel.queueBind(queue,"logs","");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){
              
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              
                        System.out.println("消费者2"+new String(body));
                    }
                });
            }
        }

### 订阅模型-Direct 类型（routing） ###

在 Fanout 模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，可能希望不同的消息被不同的队列消费。这时就要用到 Direct 类型的 Exchange。

在 Direct 模型下：

*  队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由 key）
 *  消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
 *  Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routingkey 与消息的 Routing key 完全一致，才会接收到消息

![在这里插入图片描述][cd7a53c1488c401fa7d0f440eef15cc1.png_pic_center]

场景举例：出现 error 时，既需要在控制台打印，又需要在日志中持久化

*  P：生产者，向 Exchange 发送消息， 发送消息时，会指定一 个 routing key
 *  x：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列
 *  C1：消费者，其所在队列指定了需要 routing key 为 error 的消息
 *  C2：消费者，其所在队列指定了需要 routing key 为 info、 error、 warning 的消息

**示例代码：**

*  生产者：
    
        public class Provider {
              
            public static void main(String[] args) throws IOException {
              
                // 创建connection
                Connection connection = RabbitMQUtils.getConnection();
                // 创建channel
                Channel channel = connection.createChannel();
                String exchangeName = "logs_direct";
                // 通道声明指定交换机 参数1：交换机名称 参数2：交换机类型，direct是路由模式
                channel.exchangeDeclare(exchangeName,"direct");
                // 发送消息
                channel.basicPublish(exchangeName,"error",null,("这是direct模型发布的基于route" +
                        " kye: ["+"info"+"]发送的消息").getBytes());
            }
        }
 *  消费者：
    
        public class Customer1 {
              
            public static void main(String[] args) throws IOException {
              
                // 1.建立连接
                Connection connection = RabbitMQUtils.getConnection();
                // 建立channel
                Channel channel = connection.createChannel();
                String exchangeName = "logs_direct";
                // 绑定交换机和通道
                channel.exchangeDeclare(exchangeName,"direct");
                // 创建队列
                String queue = channel.queueDeclare().getQueue();
                // 基于路由key ,绑定队列和交换机
                channel.queueBind(queue,exchangeName,"error");
                // 消费
                channel.basicConsume(queue,true,new DefaultConsumer(channel) {
              
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              
                        System.out.println("消费者1："+ new String(body));
                    }
                });
            }
        }
    
        public class Customer2 {
              
            public static void main(String[] args) throws IOException {
              
                // 1.建立连接
                Connection connection = RabbitMQUtils.getConnection();
                // 建立channel
                Channel channel = connection.createChannel();
                String exchangeName = "logs_direct";
                // 绑定交换机和通道
                channel.exchangeDeclare(exchangeName,"direct");
                // 创建队列
                String queue = channel.queueDeclare().getQueue();
                // 基于路由key ,绑定队列和交换机
                channel.queueBind(queue,exchangeName,"info");
                channel.queueBind(queue,exchangeName,"error");
                channel.queueBind(queue,exchangeName,"warning");
         
                // 消费
                channel.basicConsume(queue,true,new DefaultConsumer(channel) {
              
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              
                        System.out.println("消费者2："+ new String(body));
                    }
                });
            }
        }

### 订阅模型-Topic 类型 ###

Topic 类型的 Exchange 与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange 可以让队列在绑定 Routing key 的时候使用通配符。

这种模型 Routingkey 一般都是由一个或多个单词组成，多个单词之间以"."分割，例如: item.insert

![在这里插入图片描述][22bc5cc9e5624bcfa7ae08cf3b0e34f4.png_pic_center]

通配符：
    	* 	匹配不多不少恰好1个词
    	# 	匹配0个，1个或多个词
    如:
    	audit.#    匹配 audit.irs.corporate 或者 audit.irs 等
    	audit.*    只能匹配 audit.irs

**示例代码：**

*  生产者
    
        public class Provider {
              
            public static void main(String[] args) throws IOException {
              
                // 建立连接工厂
                Connection connection = RabbitMQUtils.getConnection();
                // 建立channel
                Channel channel = connection.createChannel();
                String exchangeName = "topics";
                // 交换机
                channel.exchangeDeclare(exchangeName,"topic");
                // 发消息
                String routekey = "user.save";
                channel.basicPublish(exchangeName,routekey,null,("这是topic动态路由模型发布的基于route" +
                        " key :["+routekey+"]").getBytes());
            }
        }
 *  消费者
    
        public class Customer1 {
              
            public static void main(String[] args) throws IOException {
              
                Connection connection = RabbitMQUtils.getConnection();
         
                Channel channel = connection.createChannel();
                //通道绑定交换机
                channel.exchangeDeclare("topics","topic");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 绑定交换机和队列
                channel.queueBind(queue,"topics","user.*");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){
              
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              
                        System.out.println("消费者1"+new String(body));
                    }
                });
            }
        }
    
        public class Customer2 {
              
            public static void main(String[] args) throws IOException {
              
                Connection connection = RabbitMQUtils.getConnection();
         
                Channel channel = connection.createChannel();
                //通道绑定交换机
                channel.exchangeDeclare("topics","topic");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 绑定交换机和队列
                channel.queueBind(queue,"topics","user.#");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){
              
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              
                        System.out.println("消费者2"+new String(body));
                    }
                });
            }
        }

# RabbitMQ 控制台 #

## 管理界面 ##

![在这里插入图片描述][8b5d9685f6494b6e8ab09713eab4dc78.png_pic_center]

## 添加用户 ##

![在这里插入图片描述][395ec72690df4109b69b35446eb11017.png_pic_center]

## 创建 Virtual Hosts 并且分配权限 ##

![在这里插入图片描述][160fdfac5230467db9c83705cc1a544b.png_pic_center]

[ad0933ceaf104ec7b60ca63e7c2d5f2d.png_pic_center]: https://img-blog.csdnimg.cn/ad0933ceaf104ec7b60ca63e7c2d5f2d.png#pic_center
[9f208a2af4cf4fa6a871dd792e72b95f.png_pic_center]: https://img-blog.csdnimg.cn/9f208a2af4cf4fa6a871dd792e72b95f.png#pic_center
[7b717214b6744dbe86d71cff5e864a10.png_pic_center]: https://img-blog.csdnimg.cn/7b717214b6744dbe86d71cff5e864a10.png#pic_center
[886c81e9089d4273985e54731470d95a.png_pic_center]: https://img-blog.csdnimg.cn/886c81e9089d4273985e54731470d95a.png#pic_center
[cd7a53c1488c401fa7d0f440eef15cc1.png_pic_center]: https://img-blog.csdnimg.cn/cd7a53c1488c401fa7d0f440eef15cc1.png#pic_center
[22bc5cc9e5624bcfa7ae08cf3b0e34f4.png_pic_center]: https://img-blog.csdnimg.cn/22bc5cc9e5624bcfa7ae08cf3b0e34f4.png#pic_center
[8b5d9685f6494b6e8ab09713eab4dc78.png_pic_center]: https://img-blog.csdnimg.cn/8b5d9685f6494b6e8ab09713eab4dc78.png#pic_center
[395ec72690df4109b69b35446eb11017.png_pic_center]: https://img-blog.csdnimg.cn/395ec72690df4109b69b35446eb11017.png#pic_center
[160fdfac5230467db9c83705cc1a544b.png_pic_center]: https://img-blog.csdnimg.cn/160fdfac5230467db9c83705cc1a544b.png#pic_center