rabbitmq知识点

野性酷女 2023-01-13 15:51 137阅读 0赞

## 消息队列 ##

### 消息队列两种模型 ###

*  队列模型：生产者往某个队列发送消息，一个队列可以存储多个生产者的消息，也可以有多个消费者，但消费者之间是竞争关系，即每条消息只能被一个消费者消费。
    
    ![dd00db2c6df360ac5556779b702f519e.png][]
 *  发布/订阅模型：为了实现一个消息被多个消费者消费。改模型是将消息发往一个Topic（主题）中，只要订阅了这个主题的订阅者都能消费这个消息。
    
    ![df943f273a304c67ad6b0b70933aa86e.png][]
 *  队列模型相当于一对一聊天，发布订阅相当于群聊。
 *  队列模型也可以通过消息全量存储至多个队列来解决一条消息被多个消费者消费的问题，但会有数据的冗余。
 *  发布订阅模型兼容队列模型，即只有一个消费者的情况下和队列模型基本一致。
 *  `RabbitMQ` 采用队列模型，`RocketMQ`和`Kafka` 采用发布/订阅模型。

### JMS ###

*  Java消息服务，2001年sun公司发布的，在个大厂商的MQ上面统一包装一层java的规范，我们只需要针对api编程即可，但JMS只适用于java语言
 *  在06年时，AMQP发布了，跨语言跨平台的。

### 什么是MQ？特点？ ###

*  用高效可靠的消息传递机制进行与平台无关的数据交流，支持分布式系统的集成。
 *  mq的特点：
    
     *  是一个独立运行的服务。生产者和消费者都要先跟此服务器建立连接
     *  采用队列作为数据结构，先进先出
     *  具有发布订阅的模型，消费者可以获取自己需要的消息。

### java中有这么多的对列的实现，为什么还要用mq呢？ ###

*  java中queue的实现不能跨进程，不能在分布式系统中使用，也没有持久化机制
 *  就像有了HashMap之后为什么还要redis做缓存是一样的。

### 消息队列的作用与使用场景 ###

*  异步通信，对于数据量大或者处理耗时长的操作，减少客户端等待，提升响应速度
    
     *  比如银行实时转账，其实是异步的，本行做了处理后还需要告诉其他行，其他行在做处理，然后再一步步返回，但实际上我们转账几秒就可以收到钱，是因为大部分的MQ有低延迟特性，能在短时间内处理非常多的消息
 *  系统解耦，对于改动影响大的系统，减少系统间的依赖
    
     *  耦合：系统内部或系统之间存在相互作用，依赖的情况。
     *  比如用户取消订单，此时订单系统需要告诉库存系统，支付系统，通知系统等，在我们的代码中这三个是串行执行的，如果在库存系统发生了异常，那么后面的系统就不会走了，但是库存出错应该是不影响退钱和通知用户的，此时系统和系统之间耦合很高，使用mq就可以把串行改成并行。订单系统分别给库存系统，支付系统，通知系统发消息，让他们去加库存，退钱，通知用户。
     *  当然，用多线程或线程池也可以实现，但每一个需要并行执行的地方都要引入线程，又会带来线程或线程池的管理问题
 *  流量削峰，对于出现瞬间的流量峰值的系统，保护应用和数据库。
    
     *  比如双十一这种千万级别的并发量，如果堆硬件的方式去解决那么在峰值过去后就会出现巨大的资源浪费，如果不堆硬件而是使用限流的方式保护我们的服务器又会导致订单的丢失(用户半天买不到就不买了。。)
     *  这时就要引入mq，先进先出，先把所有的流量接下来，转换成mq消息发送到消息队列服务器上，业务层就可以根据自己的消费速率去处理消息。

### 使用mq带来的问题？ ###

*  系统可用性降低：原来是两个节点的通信，现在还需要独立运行一个服务，如果mq服务出现问题，就导致请求失败
 *  系统复杂性提高：需要去学习mq，要考虑消息丢失和重复消息的问题，一旦消息没有被正确消费，就会带来数据一致性问题。

## Rabbitmq ##

### rabbitmq基本特性 ###

*  高可靠：提供持久化，发送应答，发布确认，高可用等特性
 *  灵活的路由：通过交换机实现消息的灵活路由
 *  支持多客户端：对主流的语言都支持
 *  集群与扩展：支持集群和负载
 *  高可用队列：通过镜像队列实现队列中数据的复制
 *  权限管理：提供了基于vhost和用户的权限管理
 *  插件系统：通过插件来增强rabbitmq的功能
 *  与spring集成：Spring对AMQP进行了封装

### rabbitmq工作模型 ###

*  ![5e8bbdadd02daf00507fdd135e6cac76.png][]
 *  broker：就是mq服务器，用于存储和转发消息。
 *  Connection：无论是生产者发送消息还是消费者接收消息，都必须要跟broker建立一个连接，并且是一个tcp的长连接
 *  Channel：如果所有的生产者和消费者都之间创建和释放tcp长连接的话，对于broker来说是很大的性能损耗，tcp连接是非常宝贵的资源，创建和释放也要消耗时间。所以引入了channel，在保持的tcp长连接里创建和释放channel，减少资源消耗
 *  Queue：存储消息，独立的进程，有自己的数据库（Mnesia）。
    
     *  消费者获取消息有两种模式，一种是push，只要服务器接收到消息就马上推给消费者，另一种是pull，消息存放在服务端，需要消费这主动获取消息，那么消费者需要写一个while循环不断从队列获取消息吗？不需要，可以基于事件机制，实现消费者对队列的监听。
     *  但是在rabbitmq中不会出现消息直接发给队列的情况，因为引入了交换机
 *  Exchange：实现消息的灵活路由，把消息发给Exchange，然后通过routing key将消息发给对列
 *  注意：交换机与队列，队列与消费者都是多对多关系
 *  vhost：虚拟机，当某个业务不想跟别的业务共用一个rabbitmq的服务器，这时不需要再另外安装一个rabbitmq服务器，只需要创建一个vhost就可以使用自己专属的交换机和队列了。实现隔离。
    
     *  还可以为不同用户分配vhost的权限，如给A用户分配名称叫A的vhost的权限，给root用户分配所有vhost权限

### rabbitmq交换机的类型与使用场景 ###

*  direct：直连交换机，需指定一个routing key，把消息发给完全匹配routing key的队列。不匹配，消息丢失。如果要保证消息不丢失，需要配置rabbitTemplate.setReturnCallback()方法，rabbitmq会把消息回发给生产者。或者将消息路由到备份交换机
    
     *  一个系统只和另一个系统交互时用。
 *  topic：主题交换机，在routing key中使用了通配符，\#表示0个或多个单词，\*表示一个单词
    
     *  如对消息做分级或过滤的需求
 *  fanout：广播交换机，不需要routing key，消息到达交换机后，交换机把消息发给所有绑定了此交换机的对列。
    
     *  通用类型消息

### TTL ###

*  time to live，如果消息放在队列里面一定时间还没有被消费的话，消息就会过期就会被删除。
 *  如何设置消息的过期时间呢？
    
     *  通过队列属性(x-message-ttl)设置统一的过期时间。
     *  发送消息时，单独为消息指定过期时间
     *  注意：如果同时在队列中制定了过期时间并在发送消息时也指定了过期时间，那么小的那个时间生效。

### 死信（Dead Letter）队列 ###

*  消息在某些情况下会变成死信，用于存储死信的队列叫做死信队列。用来路由死信的交换机叫死信交换机
 *  死信队列和死信交换机其实也是普通的队列和交换机

### 什么情况下消息会变成死信？ ###

*  消息被消费者拒绝
 *  消息过期
 *  队列达到最大长度。队列最大长度有两个属性：x-max-length（最大消息个数）或者x-max-length-bytes（最大存储的字节数）。达到其中一个的长度就会把对头（最先入队）的消息丢到死信交换机

### 死信队列如何使用 ###

*  在queue中有个属性：x-dead-letter-exchange可以指定死信交换机，当消息变成死信后就会把消息给死信交换机，死信交换机又绑定了死信队列，所以死信消息就跑到了死信队列中，然后被监听死信队列的消费者消费
 *  死信队列作用：可以实现延迟队列

### 延迟队列 ###

*  场景：如，有一个未付款的订单，想让它在15分钟后关闭
 *  解决方案：
    
     *  先存储到数据库，用定时任务扫描，如对比入库的时间和当前的时间有没有超过15分钟，有就关闭订单。
     *  死信队列实现
     *  使用rabbitmq-delayed-message-exchange插件实现
        
         *  使用插件比使用死信队列的好处（死信队列的缺点）？
            
             *  当有多的时间梯度比如每过1分钟，5分钟，10分钟…我们要做一些事时就要定义很多的交换机和队列来路由消息（消息可以死多次）。
             *  单独设置消息的TTL，可能造成消息阻塞，如第一条消息过期时间是30min，第二条过期时间是10分钟，10分钟后，由于第一条消息还未出队，导致第二条消息无法投递的情况

### 服务端流控 ###

*  当生产消息的速度远大于消费消息的速度时，会产生大量的消息堆积，占用系统资源，想要控制服务端接受的消息数量，就要用到服务端流控了。
 *  x-max-length, x-max-length-bytes这两个参数一旦超过设置的值就会丢弃对头的消息，并没有真正实现服务端限流。
 *  流控方法：
    
     *  内存控制。rabbitmq在启动时会检测机器的物理内存，默认当mq占用40%以上内存时，MQ会主动抛出内存警告并阻塞所有连接(Connections)，内存阈值可以修改。
     *  磁盘控制：当磁盘空间低于指定值(默认50M)或百分比（30%）时触发流控，不让生产者发送消息。
        
         *  rabbitmq的配置文件中可以改
         *  disk\_free\_limit.relative = 3.0 指定为磁盘的 30%  
            disk\_free\_limit.absolute = 2GB 指定2GB

### 消费端限流 ###

*  默认情况下，rabbitmq会尽可能快的把队列中的消息发给消费者，因为消费者会在本地缓存消息，消息过多就会导致oom，此时希望在一定数量的消息被消费完之前不再推送消息过来，就要用消费的流量限制措施。
 *  可以基于consumer或channel设置prefetch count的值，表示consumer端最大的unacked messages的数目，当超过这个数值的消息未被确认，rabbitmq会停止投递新的消息给改消费者。
 *  spring boot配置：spring.rabbitmq.listener.simple.prefetch=2

### 在一个工程中发送消息到多个rabbitmq服务中，怎么做？ ###

*  定义多个rabbitTemplate，注入不同的ConnectionFactory

### rabbitmq消息可靠性投递 ###

*  需要先发送消息，再修改数据库
 *  效率和可靠性是无法兼得的，如果要保证每一个环节都成功，就会对消息收发效率造成影响。所以如果是一些实时一致性要求不是特别高的场合，可以牺牲一些可靠性来换取效率。
 *  使用rabbitmq收发消息的时候，要注意这些环节
    
     *  怎么保证消息从生产者发送到broker
     *  保证exchange路由到queue
     *  消息存储在queue中，如何保证消息在队列稳定的存储呢
     *  怎么确保消费者收到了消息？

### 确保消息发送到了rabbitmq服务器 ###

*  rabbitmq提供了两种服务端确认机制。第一种是事务模式，第二种是确认模式
 *  事务模式：通过channel.txselect()方法把信道设置成事务模式，然后发消息给rabbitmq，然后调用channel.txCommit()方法，如果channel.txCommit()方法返回true，说明消息一定到达了rabbitmq中。如果出现异常，将其捕获通过channel.txRollback()方法实现事务回滚。
    
     *  缺点：是阻塞的，一条消息没有发送完毕不能发下一条，不建议使用
     *  spring boot中使用rabbitTemplate.setChannelTransacted(true)开启信道的事务模式
 *  确认模式：
    
     *  普通确认模式：生产者通过调用channel.confirmSelect()方法将信道设置为confirm模式，然后发送消息，当channel.waitForConfirms()返回true，就说明rabbitmq服务端接收到消息
        
         *  缺点：发送一条确认一条效率仍然低
     *  批量确认模式：生产者开启confirm模式后，先发送一批消息，只要channel.waitForConfirmsOrDie方法没有抛出异常，就代表消息都被服务端接受了
        
         *  缺点：批量发送消息数量设置多少好呢？太少效率低，太多可能导致最后一条失败那前面发的消息都要回滚
     *  异步确认模式：需要调用channel.addConfirmListener(new ConfirmListener() \{\})方法添加一个ConfirmListener，它不一定是多少条消息确认一次，可能5条也可能10条。如果消息被确认会走到handleAck()方法，如果未确认会走到handleNack()方法，并且用一个SortedSet（confirmSet）来维护没有被确认的消息。
        
         *  rabbitTemplate使用方式：从rabbitTemplate.setConfirmCallback()方法中设置confirmCallback，然后会收到一个ack，当ack是true说明broker收到了消息，false说明接收失败。这些数据可以记录下来，重发。
            
                rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { 
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) { 
                if (!ack) { 
                System.out.println("发送消息失败：" + cause);
                throw new RuntimeException("发送异常：" + cause);
                }
                }
                });

### 确保消息从交换机路由到队列 ###

*  当路由键错误，或队列不存在时会导致消息无法路由到正确的队列。
 *  两种方式解决，一种是让服务端把消息退给生产者，另一种是让交换机路由到另一个备份的交换机。
 *  消息回发方式：
    
     *  使用mandatory参数(消息是否一定要路由到正确的队列)等于true和ReturnListener实现。
     *  channel.addReturnListener(new ReturnListener() \{…\})
     *  rabbitemplate.setMandatory(true);
        
        rabbitTemplate.setReturnCallback()
 *  备份交换机方式：交换机上有一个参数，叫alternate-exchange，就是为当前交换机指定一个备份交换机。备份交换机绑定的有队列，队列又可以绑定消费者，然后就可以取到这条消息做下一步操作，如记录日志，写到数据库，给开发人员一个通知
    
     *  代码：再声明交换机时指定参数
     *      Map<String,Object> arguments = new HashMap<String,Object>();
            arguments.put("alternate-exchange","ALTERNATE_EXCHANGE"); // 指定交换机的备份交换机
            
            channel.exchangeDeclare("TEST_EXCHANGE","topic", false, false, false, arguments);

### 确保消息在队列中存储 ###

*  消息在队列中还没有被消费，然后rabbitmq服务出现了宕机，重启等问题，就可能导致内存中的消息丢失，所以要把消息本身和队列，交换机都保存到磁盘
 *  解决方案：
    
     *  队列持久化：定义队列时把durable参数设置成true
     *  交换机持久化：定义交换机时把durable参数设置成true
     *  消息持久化：原生qmqp协议的api上可以对消息的属性指定diliveryMode(2)表示持久化。springboot中把消系封装成一个message对象，message有一个properties属性调用方法setDeliveryMode()也可以持久化
     *  集群：如果rabbitmq是单节点，即使对交换机，队列，消息做了持久化，还是不能保证高可用，所以需要集群

### 确保消息投递到消费者 ###

*  如果消费者在处理消息时发生了异常，那消息就消费失败了，那么此时服务端应该要得知消费者对消息的处理情况，并决定是否重新投递消息。
 *  rabbitmq提供了消费者的消息确认机制（message acknowledgement），消费者可以自动或者手动发送ack给服务端。自动ack有问题，因为自动ack是消费者接收到消息就自动发ack了，那在消费者处理消息时抛出异常我们就不知道了，所以要在执行方法完成后再手动发送ack，rabbitmq会等待消费者显示地回复确认信号后才从队列中移除消息。
 *  如何设置手动ack？
    
     *  也是调用channel的方法 channel.basicAck()，也可以拒绝接收消息channel.basicReject()拒绝单条，channel.basicNack()批量拒绝，如果只有一个消费者，requeue参数为true时（设置消息重新入队）会造成消息重复消费，出现死循环，如果又多个消费者，会把消息给到别的消费者。所以requeue要慎用
     *  springboot中，消费者被封装成MessageLisener，调用MessageLisenerContainer对象的setAcknowledgeMode()方法设置应答方式，然后通过rabbitmq中的channel的basicAck方法发送ack
        
         *  NONE：自动ack
         *  MANUAL：手动ack
         *  AUTO：如果方法未抛出异常，发送ack

### 当消费者消费失败了，生产者如何知道？ ###

*  方式一：消费者处理完成后调用生产者自定义的api，比如修改数据库这条记录的状态。（破坏了解耦，不建议）
 *  方式二：消费者给生产者发送一条应答消息
 *  当生产者在约定的超时时间后还没有收到消费者的应答消息也没有收到接口回调，那么生产者可以认为消费者没有收到消息或消费失败，此时怎么办？
    
     *  重发：注意重发的间隔，重发次数。比如2分钟发一次，最多重发3次，否则造成消息堆积。
     *  如何重发？
        
         *  可以把消息登记到数据库中，然后用定时任务实现，发送一次消息的次数就递增一次
     *  重发是否发送一模一样的消息？
        
         *  需要根据业务场景来做，比如客户存款，钱已经被atm机吞了，所以这时要保证成功，发存款确认的报文。又如客户取款，atm还没有吐钞，此时要保证失败，不能扣减余额，发送冲正的报文

### 如何避免消息被重复处理（实现消息幂等性）？ ###

*  要在消费端来处理
 *  对每一条消息生成一个唯一的业务id，通过日志或消息落库来做重复控制，任何进来的消息和出去的消息都要经过表的检查（比如对比id）防止重复消息的处理。

### 消息重发如何实现 ###

*  在生产者那边发送消息前将消息记录到日志表中，当生产者收到消费者的回调或应答消息，就将表中那行消息状态改成发送完成，如果生产者一直没收到回调，就需要重发消息，根据日志+定时任务来重发消息。

### 当重试到次数上限后，还没成功，怎么办？ ###

*  人工去看看是什么问题。

### rabbitmq集群作用 ###

*  集群用于实现高可用和负载均衡
 *  高可用：当集群中某些mq服务器不可用时，客户端还可以连接到其他mq服务器
 *  负载均衡：单台服务器能处理的消息有限，可以把消息分发到多台mq服务器

### rabbitmq集群模式： ###

*  普通集群模式：不同的节点之间只会相互同步元数据，不复制消息内容，不能保证队列高可用。（一般不用）
 *  镜像队列模式：消息内容会在镜像节点间同步，可用性更高。但系统性能降低，节点太多的情况下同步代价大

### rabbitmq如何支持集群？ ###

*  应用做集群，需要面对数据同步和通信问题，erlang天生具备分布式特性，所以rabbitmq天然支持集群。rabbitmq的不同服务器上面有.erlang.cookie，只有当这个erlang.cookie是完全一致时，不同的节点之间才能实现集群

### rabbitmq集群节点类型 ###

*  磁盘节点：将元数据（交换机，队列，绑定关系，vhost）放在磁盘中
 *  内存节点：将元数据放在内存
 *  集群中至少需要一个磁盘节点用来持久化元数据，否则全部内存节点崩溃时，就无法同步元数据了，未指定类型的情况下，默认为磁盘节点。
 *  一般把应用连接到内存节点，读写快。磁盘节点用来备份。

### 实践经验总结 ###

#### 资源管理 ####

*  交换机和队列由谁创建
    
     *  运维管理员
 *  为什么都在消费者这边定义交换机和queue
    
     *  基于遵循谁使用谁管理的原则，但不是由代码创建的，是运维在我们写代码之前就已经创建好的。

#### 配置文件与命名规范 ####

*  元数据命名集中放在properties文件中，不要硬编码。
 *  命名体现元数据类型
    
     *  虚拟机命名：XXX\_VHOST
     *  交换机命名：XXX\_EXCHANGE
     *  队列命名：XXX\_QUEUE
 *  命名体现数据来源和去向
    
     *  如销售系统发往产品系统的交换机：SALE\_TO\_PRODUCT\_EXCHANGE

#### 调用封装 ####

*  可以对rabbitTemplate做进一步封装，简化消息发送

#### 信息落库+定时任务 ####

*  将需要发送的消息登记在数据库中（记录日志），可以实现消息的可追溯和重复控制，需要配合定时任务来实现

#### 生产环境运维监控 ####

*  虽然rabbitmq提供了简单的图形化界面，但如果对于系统性能，高可用等参数有一些定制化的监控需求的话，可以通过zabbix来实现监控。
 *  主要关注：磁盘，内存，连接数

#### 日志追踪 ####

*  rabbitmq可以通过firehose来记录消息流入流出情况，用于调试，排错

#### 在批量发送（如一次发10w）消息的场景中，如何减少连接数？ ####

*  可以通过合并消息内容，一次多发点内容来减少消息的数量，如用json数组
 *  建议单条消息不要超过4M

### Channel和vhost的作用是什么？ ###

*  channel：减少tcp连接资源的消耗
 *  vhost：实现资源隔离

### RabbitMQ的消息有哪些路由方式？适合在什么业务场景使用？ ###

*  同rabbitmq交换机的类型与使用场景

### 交换机与队列、队列与消费者的绑定关系是什么样的？ ###

*  交换机与队列是多对多的关系
 *  队列与消费者是多对多的关系，但是队列中的一条消息只会被一个消费者消费

### 无法被路由的消息，去了哪里？ ###

*  直接被丢弃了，可以使用回发，或备份的交换机去把数据保留下来

### 消息在什么时候会变成Dead Letter（死信）？ ###

*  消息过期了，无论是消息的过期时间导致的过期还是队列消息的过期时间导致的过期
 *  消息超过了队列的长度，队列的长度有两个属性，x-max-length(消息的数量)，x-max-length-bytes(消息的字节数)
 *  消费者调用了reject和Nack并且没有设置requeue

### 如果一个项目要从多个服务器接收消息，怎么做？如果一个项目要发送消息到多个服务器，怎么做？ ###

*  定义多个ConnectionFactory，再把它作为参数，生产者这边把它放到Template中去，消费者这边把它放到ListenerContainer上去，然后再把ListenerContainer配置在消费者类的属性上。

### rabbitmq如何实现延迟队列 ###

*  把消息登记到数据库，通过定时任务来扫描
 *  通过TTL+死信队列
 *  使用插件（推荐）

### 哪些情况会导致消息丢失？怎么解决？ ###

*  每一个环节都要去分析

### 一个队列最多可以存放多少条消息 ###

*  如果没在队列中配置消息的个数也没有配队列的最大字节数的话，取决于硬件的配置。就是磁盘和内存的大小

### 如何提高消息消费速率？ ###

*  增加消费者

### 如何动态地创建队列和消费者？ ###

*  在ListenerContainer中做的

### spring AMQP中消息怎么封装？用什么转换？ ###

*  用Message对象封装
 *  用MessageConvertor转换

### 如何保证消息的顺序性 ###

*  保证一个队列只能有一个消费者，因为如果有多个消费者那顺序性就无法保证了。

### rabbitmq的集群节点类型 ###

*  磁盘节点和内存节点，必须要有一个磁盘节点

### 如果有大量消息堆积怎么办？（有一个消费者就是很长时间没有发送ack应答，导致消息在队列堆积） ###

*  重启（代码解决了再重启）
 *  多创建几个消费者同时消费
 *  直接清空队列，重发消息。（如果已经影响了rabbitmq的性能）

### 设计一个MQ，思路是什么？ ###

*  从mq的本质入手：存储与转发
 *  存储：
    
     *  内存：用什么数据结构
     *  硬盘：用文件系统？用数据库？
 *  转发：要和生产者消费者建立连接，用哪些通信协议（tcp，http，amqp），又支持哪些路由方式呢？direct一对一，fanout一对多？发消息给消费者支持推模式还是拉模式？

### rabbitmq消费者消费消息支持推还是拉 ###

*  推拉都支持
 *  使用channel.basicConsume()可以设置推（投递）模式
 *  使用channel.basicGet()可以设置拉模式
 *  一般都使用推模式，要想实现高吞吐量，需要使用推模式，因为拉模式每次都要去消息中间件中拉取消息来消费，影响性能

[dd00db2c6df360ac5556779b702f519e.png]: /images/20221022/740132f5951b47188778f1afdcefd6bd.png
[df943f273a304c67ad6b0b70933aa86e.png]: /images/20221022/7a367ebc2d3f443f8e9d0bbd94ca0596.png
[5e8bbdadd02daf00507fdd135e6cac76.png]: /images/20221022/cd6f3088ec7049ac857028028111bdeb.png