RabbitMQ--架构、相关概念、运转流程

缺乏、安全感 2021-08-30 01:45 330阅读 0赞

#### RabbitMQ架构 ####

RabbitMQ整体上是-一个生产者与消费者模型，主要负责接收、存储和转发消息。可以把消息传递的过程想象成:当你将一个包裹送到邮局，邮局会暂存并最终将邮件通过邮递员送到收件人的手上,RabbitMQ就好比由邮局、邮箱和邮递员组成的-一个系统。从计算机术语层面来说，RabbitMQ模型更像是一种交换机模型。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

生产者（Producer）  
生产者创建消息，然后发布到RabbitMQ中。消息一般可以包含2个部分:消息体和标签(Label)。消息体也可以称之为payload， 在实际应用中，消息体一般是一个带 有业务逻辑结构的数据，比如一个JSON字符串。当然可以进一-步对这个消息体进行序列化操作。消息的标签用来表述这条消息，比如一个交换器的名称和一个路由键。生产者把消息交由RabbitMQ,RabbitMQ之后会根据标签把消息发送给感兴趣的消费者( Consumer)。

消费者（Consumer）  
消费者连接到RabbitMQ服务器，并订阅到队列上。当消费者消费一条消息时，只是消费消息的消息体(payload)。 在消息路由的过程中，消息的标签会丢弃，存入到队列中的消息只有消息体，消费者也只会消费到消息体，也就不知道消息的生产者是谁，当然消费者也不需要知道。

Broker：消息中间件的服务结点  
对于RabbitMQ来说，一个RabbitMQ Broker可以简单地看作一个RabbitMQ服务节点，或者RabbitMQ服务实例。大多数情况下也可以将一个RabbitMQ Broker 看作一台 RabbitMQ服务器。

下图展示了生产者将消息存入RabbitMQBroker，以及消费者从Broker中消费数据的整个流程。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

Queue（队列）  
Queue是RabbitMQ的内部对象，用于存储消息：  
![在这里插入图片描述][20200515143455757.png]

RabbitMQ中消息都只能存储在队列中，这一点和Kafka这种消息中间件相反。Kafka 将消息存储在topic (主题)这个逻辑层面，而相对应的队列逻辑只是topic实际存储文件中的位移标识。RabbitMQ的生产者生产消息并最终投递到队列中,消费者可以从队列中获取消息并消费。

多个消费者可以订阅同一个队列,这时队列中的消息会被平均分摊(Round-Robin,即轮询)给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]  
RabbitMQ不支持队列层面的广播消费，如果需要广播消费，需要在其上进行二次开发，处理逻辑会变得非常复杂，同时也不建议这么做。

Exchange（交换器）  
生产者将消息发送到Exchange (交换器，通常也可以用大写的“X”来表示)，由交换器将消息路由到一个或者多个队列中。如果路由不到，或许会返回给生产者，或许直接丢弃。这里可以将RabbitMQ中的交换器看作一个简单的实体。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]

Connection和Channel  
无论是生产者还是消费者，都需要和RabbitMQ Broker 建立连接，这个连接就是一条TCP连接，也就Connection。一旦TCP连接建立起来，客户端紧接着可以创建一一个AMQP信道(Channel)， 每个信道都会被指派一个唯一的 ID。信道是建立在Connection之上的虚拟连接，RabbitMQ 处理的每条AMQP指令都是通过信道完成的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

我们完全可以直接使用Connection就能完成信道的工作，为什么还要引入信道呢?试想这样一个场景，一个应用程序中有很多个线程需要从RabbitMQ中消费消息，或者生产消息，那么必然需要建立很多个Connection,也就是许多个TCP连接。然而对于操作系统而言，建立和销毁TCP连接是非常昂贵的开销，如果遇到使用高峰，性能瓶颈也随之显现。RabbitMQ 采用类似NIO’ (Non-blocking I/O) 的做法，选择TCP连接复用，不仅可以减少性能开销，同时也便于管理。

每个线程把持一个信道，所以信道复用了Connection的TCP连接。同时RabbitMQ可以确保每个线程的私密性，就像拥有独立的连接一样。当每个信道的流量不是很大时，复用单一的Connection可以在产生性能瓶颈的情况下有效地节省TCP连接资源。但是当信道本身的流量很大时，这时候多个信道复用一个Connection就会产生性能瓶颈，进而使整体的流量被限制了。此时就需要开辟多个Connection,将这些信道均摊到这些Connection中，至于这些相关的调优策略需要根据业务自身的实际情况进行调节。

#### 运转流程 ####

生产者发送消息：

1.  生产者连接到RabbitMQ Broker, 建立一个连接(Connection),开启一个信道(Channel)
2.  生产者声明一个交换器，并设置相关属性，比如交换机类型、是否持久化等
3.  生产者声明一个队列并设置相关属性，比如是否排他、是否持久化、是否自动删除等
4.  生产者通过路由键将交换器和队列绑定起来
5.  生产者发送消息至RabbitMQ Broker,其中包含路由键、交换器等信息
6.  相应的交换器根据接收到的路由键查找相匹配的队列。
7.  如果找到，则将从生产者发送过来的消息存入相应的队列中
8.  如果没有找到，则根据生产者配置的属性选择丢弃还是回退给生产者
9.  关闭信道。
10. 关闭连接。

消费者接受消息：

1.  消费者连接到RabbitMQ Broker,建立以个连接(Connection),开启一个信道( Channel)。
2.  消费者向RabbitMQBroker请求消费相应队列中的消息，可能会设置相应的回调函数,以及做一些准备工作
3.  等待RabbitMQBroker回应并投递相应队列中的消息，消费者接收消息。
4.  消费者确认(ack) 接收到的消息。
5.  RabbitMQ 从队列中删除相应已经被确认的消息。
6.  关闭信道。

#### 简单示例 ####

生产者：

public class RabbitProducer { 
        private static final String EXCHANGE_NAME = "exchange_demo";
        private static final String ROUTING_KEY = "routingkey_demo";
        private static final String QUEUE_NAME = "queue_demo";
        private static final String IP_ADDRESS = "127.0.0.1";
        private static final int PORT = 5672;//RabbitMQ服务端默认端口号为5672
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection = factory.newConnection();
            //创建信道
            Channel channel = connection.createChannel();
            //创建一个type="direct"、持久化的、非自动删除的交换器
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
            //创建一个持久化、非排他的、非自动删除的队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //将交换器与队列通过路由键绑定
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            //发送一条持久化消息：hello world！
            String message = "Hello World!";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,
                    MessageProperties.PERSISTENT_TEXT_PLAIN,
                    message.getBytes());
            //关闭资源
            channel.close();
            connection.close();
        }
    }

消费者：

public class RabbitConsumer { 
    
        private static final String QUEUE_NAME = "queue_demo";
        private static final String IP_ADRESS = "127.0.0.1";
        private static int PORT = 5672;
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
            Address[] addresses = new Address[]{ 
                    new Address(IP_ADRESS, PORT)
            };
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUsername("guest");
            factory.setPassword("guest");
            //这里的连接方式与生产者的demo略有不同
            Connection connection = factory.newConnection(addresses);
            //创建信道
            Channel channel = connection.createChannel();
            Consumer consumer = new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("recv message: " + new String(body));
                    try { 
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(QUEUE_NAME, consumer);
            //等待回到函数执行完毕之后，关闭资源
            TimeUnit.SECONDS.sleep(5);
            channel.close();
            connection.close();
        }
    }

recv message: Hello World!

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/7059c195b3e841ff892447e44d1f0306.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/2d46eeed1a154b9a9ce1c49fff219b27.png
[20200515143455757.png]: /images/20210813/1badd628e88448409cdb4c530d736a16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/04cd4d3789c848699ddc5338eb6f5fcd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/b660ed5592ae47cabeca9a89a2f6b253.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/2799b299615c4121950eebb8f0e2923c.png