消息队列笔记

你的名字 2023-07-25 14:01 26阅读 0赞

### 目录 ###

*   *   *   *  为什么要使用消息队列？消息队列的好处？用途？
             *  使用消息队列有什么缺点？
             *  你还了解哪些消息队列？
             *  为什么你这两个项目分别使用了RocketMQ和Kafka？
             *  消息队列模型：
             *  RocketMQ模型（Kafka模型也是一样的）
             *  事务消息，为什么消息队列需要事务？
             *  消息队列是如何保证高可用的？
             *  检测消息是否丢失的办法
             *  RocketMQ如何保证消息不丢失的？如何保证可靠性的？
             *  Kafka如何保证消息不丢失的？如何保证可靠性的？
             *  重复发送消息，怎么保证消息不被重复消费呢？
             *  消息积压你是怎么处理的？
             *  你是怎么保证消息按顺序消费？

#### 为什么要使用消息队列？消息队列的好处？用途？ ####

消息队列用来解决系统之间的通信问题，主要功能就是收发消息，使用消息队列有几个好处：

1.  **解耦**：减少了系统之间的依赖关系，如果在一个系统A中直接调用另一个系统B的接口，如果B执行失败，A也会执行失败，两个系统高度耦合，不利于代码的扩展和维护。
2.  **异步**：提高系统响应速度，减少用户的等待时间。举例：假设我现在有一个下单功能，它包含了锁定库存，订单入库，给用户发短信3个操作，需要等这3个操作完成后才能给用户返回结果，用户需要等待较长的时间。我可以这样优化，在我这个下单功能中决定下单是否成功的操作是锁定库存，等锁定库存操作完成后就立刻返回结果给用户，然后将请求的数据放到消息队列中，由消息队列异步进行后续2步操作。好处是减少了用户等待的时间，充分利用了服务器的资源在短时间内处理了更多的请求。
3.  **削峰限流**：比如说秒杀系统秒杀时一秒钟有5000并发量，但是服务器每秒钟只能请求1000并发量，那多出来的4000个请求，可能会使系统崩溃，解决办法是把多出来的请求到消息队列里中，秒杀系统根据自己能够处理的请求数去消息对队列里拿数据。这样系统就不会崩溃。缺点是增加了系统调用链环节，导致总体响应时延变长。另外上下游系统需要将同步调用改为异步消息，增加了系统的复杂度。

#### 使用消息队列有什么缺点？ ####

1.  消息队列**时延**问题
2.  **降低系统的可用性**：系统引入的外部依赖越多，越容易挂掉。
3.  **系统复杂度提高**：使用 MQ 后可能需要保证消息没有被重复消费、消息没有丢失、保证消息传递的顺序性 等问题；
4.  **一致性问题**：A 系统处理完直接返回成功了，通过消息队列通知BCD执行，而且 B 和 D 两个系统也写库成功了，但 C 系统写库失败了，就造成数据不一致了。

#### 你还了解哪些消息队列？ ####

*  ActiveMQ：早期用的比较多，现在基本上没什么人用，社区也不是很活跃。吞吐量是万级，时延ms级别，有较低的概率会丢失消息。
 *  RabbitMQ：使用简单，开箱即用。性能也很好，社区活跃度高，吞吐量万级，是用erlang语言开发的，对于java开发者来说很难看懂，所以很难做扩展和二次开发，对于消息堆积的支持不是很好，消息堆积时，性能会急剧下降。如果对性能要求不高，用RabbitMQ即可。
 *  RocketMQ：吞吐量很高，达到了几十万级别，经过配置，消息可以做到0丢失，用java语言开发的，比较容易做扩展或二次开发，是目前比较多的消息队列。
 *  Kafka：吞吐量很高，达到了几十万级别，也可以做到消息0丢失，功能简单，一般用于大数据领域做实时计算和日志采集。

#### 为什么你这两个项目分别使用了RocketMQ和Kafka？ ####

*  RocketMQ它的吞吐量很高，时延低，稳定性高。适合处理在线业务，比如在交易系统中传递订单。所以我在秒杀系统中使用了它。
 *  而Kafka采用了异步批量设计，这种设计使得它的性能很高，但是带来的问题是它的响应时延比较高，当客户端发送一条消息时，Kafka 并不会立即发送出去，而是攒一批了再发送，所以 Kafka 不太适合在线业务的场景。在我的讨论社区项目中，如果有点赞，评论或者关注等操作，我需要去异步进行消息通知，对实时性要求不高，所以Kafka比较合适。

#### 消息队列模型： ####

①队列模型（点对点）：生产者-消费者，一条消息只能被一个消费者消费到  
②发布订阅模型：发布者-订阅者。

#### RocketMQ模型（Kafka模型也是一样的） ####

1.  RocketMQ中有生产者、消费者、主题等概念，每个主题可以包含多个队列，通过多个队列来实现并行生产和消费。这种情况下，只能在队列上保证消息的有序性，在主题层面无法保证消息顺序消费。
2.  RocketMQ中使一条消息能被多个消费者消费是通过消费组来实现的。在一个消费组中的多个消费者不能消费同一条消息，（同一个组内的消费者是竞争消费关系）但是一条消息可以被多个消费组消费。
3.  由于消息需要被不同的消费组进行多次消费，所以RocketMQ在队列上为每个消费组维护了一个消费位置，每成功消费一条消息，位置就加一。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzMzODUxOQ_size_16_color_FFFFFF_t_70]

#### 事务消息，为什么消息队列需要事务？ ####

消息队列中的事务，主要就是用来解决消息生产者和消息消费者的数据一致性问题。

*  假设一个下单过程，锁定库存成功后发消息到消息队列，订单系统收到消息去做订单入库操作。那么会有这两种可能，库存锁定成功，消息发送失败。库存锁定失败，消息发送成功。这两种情况都有可能造成数据不一致问题。所以就需要事务来保证这两个操作要么全部执行成功，要么全部失败。在分布式系统中要用分布式事务。
 *  比较常见的分布式事务实现有2PC（二阶段提交）、TCC（Try-Confirm-Cancel）和事务消息。每一种都有其特定的使用场景。
 *  事务消息使用的场景主要是那些需要异步更新数据，并且对数据实时性要求不太高的场景。事务消息需要消息队列提供相应的功能才能实现，具体实现如下：

1.  首先，订单系统（生产者）**在消息队列上开启一个事务**，然后订单系统给消息服务器发送一个“**半消息**”，这个半消息包含了完整的消息内容，只不过它在事务提交之前，对于消费者是不可见的。
2.  半消息发送成功后，订单系统就可以**执行本地事务**了，然后根据本地事务的执行结果决定**提交或者回滚事务消息**。如果本地事务执行成功，就提交事务消息，消费者就可以看到这条消息。如果本地事务执行失败，那就回滚事务消息，消费者就不会收到这条消息。这样就实现了要么都成功，要么都失败。

**如果提交或回滚事务消息时失败了怎么办？**

1.  RocketMQ增加了**事务反查机制**来解决事务消息提交失败的问题。如果订单系统（生产者）在提交或者回滚事务消息时发生网络异常，导致Broker没有收到提交或者回滚的请求，Broker会定期去Producer上去反查这个事务对应的本地事务的状态，然后根据反查结果决定提交或者回滚这个事务。
2.  为了支撑这个反查机制，我们的业务代码需要实现一个反查本地事务状态的接口，返回本地事务是执行成功还是失败。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzMzODUxOQ_size_16_color_FFFFFF_t_70 1]

#### 消息队列是如何保证高可用的？ ####

是通过多副本来实现的，每一个分区都有多个副本，生产者和消费者只和主副本交互，其他副本会同步数据，如果主副本所在的broker挂掉，会选出一个从副本作为leader，这样保证了高可用。

#### 检测消息是否丢失的办法 ####

*  可以利用消息队列的有序性来验证是否有消息丢失。Producer端在发消息时给每个消息附加一个连续递增的序号，然后在Consumer端来检查这个序号的连续性来判断是否有消息丢失，还可以通过缺失的序号来判断丢的是哪条消息。
 *  但是要注意的是像Kafka和RocketMQ这样的消息队列，是不保证在topic上的有序的，只能保证分区上是有序的，所以再发消息时必须指定分区并按区分别编号，并且检测时也要分区单独检测消息的连续性。

#### RocketMQ如何保证消息不丢失的？如何保证可靠性的？ ####

一条消息从生产到被消费，会经过三个阶段：生产阶段、存储阶段、消费阶段。要从这三个阶段保证消息不丢失。  
生产阶段：Producer创建消息，消息经过网络传输发送到Broker端。  
存储阶段：Broker将消息存储在磁盘中。如果是集群的话，消息还会被复制到其他节点上。  
消费阶段：Consumer从Broker上拉取消息，消息经过网络传输发送到Consumer上。

*  **生产阶段**：在生产阶段，是通过请求确认机制，来保证消息的可靠传递的。Producer发送消息给broker，当broker收到后会返回确认信息给Producer。所以生产者通过Broker返回的确认信息，判断消息在生产阶段没有丢失，如果丢失可以重发。（注意要正确处理返回值或者捕获异常，同步发送时，只要注意捕获异常即可；异步发送时，需要在回调方法里检查）
 *  **存储阶段**：在存储阶段，消息到了Broker端，会优先保存在内存中，然后立刻返回确认信息给生产者。随后Broker会定期批量地将消息从内存异步刷到磁盘。这种方式减少了IO次数，性能更好，但是如果Broker宕机，消息还未来得及刷入磁盘，就会出现消息丢失。
    
    如果对消息的可靠性要求非常高，可以通过配置Broker参数来避免因为宕机而丢消息。
    
    1.  首先需要配置Broker参数flushDiskType=SYNC\_FLUSH，意思是将消息保存机制修改为同步刷盘方式，也就是Broker收到消息后，就将消息存储到磁盘，存储成功再给生产者返回确认信息。
    2.  （主从部署）为了保证可用性，Broker通常采用一主多从部署方式，为了保证消息不丢失，还要将消息复制到slave节点。（默认方式下，消息写入master成功，就会返回确认信息给生产者，接着异步将消息复制到slave节点）此时如果master突然宕机不可恢复，那么还未复制到slave的消息就会丢失。为了防止消息丢失，可以采用同步复制方式，master节点把消息成功复制到slave节点之后，才会返回确认响应。需要配置参数brokerRole=SYNC\_MASTER。并且从节点也要设置为同步刷盘。
    
    以上配置虽然可以保证消息不丢失，但是会降低性能，生产中要综合选择。
 *  **消费阶段**：消费者从Broker中拉取消息进行消费，消费成功会发送确认响应给Broker，如果Broker未收到消费确认响应，消费者下次还会再次拉取到该消息，进行重试。

#### Kafka如何保证消息不丢失的？如何保证可靠性的？ ####

*  **生产阶段**：生产者发送消息给Broker后，通过Broker返回的信息来判断消息发送是否成功，如果失败的话就重新发送。  
    ① 使用send方法后，接着调用get方法获取调用结果，但是这样会让它变成同步操作。（不推荐）  
    ② 为send方法添加回调函数来获取结果，  
    要将Producer的retries重试次数设置大一些，可以保证出现网络问题的话可以自动重发消息，避免消息丢失。
 *  **消费阶段**：在分区中有一个offset的概念，表示消费者消费到的位置，默认配置下，当消费者拉取消息之后，消费者会自动提交offset。但是这样可能会产生这样一个问题，消费者刚拿到消息正准备消费时突然挂掉了，消息实际上没有被消费，但是offset却被提交了，就会导致消息丢失。  
    解决办法是关闭自动提交offset，在真正消费完消息之后再手动提交offset。（可能会带来重复消费问题）
 *  **存储阶段**：Kafka为分区引入了多副本机制，多个副本中有一个leader，其他的都是follower。消息会被先发送到leader副本，然后follower副本才能从leader副本中拉取消息进行同步。生产者和消费者只和leader副本交互，其他副本的存在只是为了保证消息存储的安全性。如果leader副本所在的broker挂掉，那么就要从follower副本中重新选出一个leader，但是如果leader的数据还有一部分没有被follower副本同步的话，就会造成消息丢失。

具体参数设置：  
①不要使用 producer.send(msg)，而要使用 producer.send(msg, callback)。记住，一 定要使用带有回调通知的 send 方法。  
②设置 retries 为一个较大的值。这里的 retries 同样是 Producer 的参数，对应前面提到 的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了 retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。  
③设置 acks = all。acks 是 Producer 的一个参数，代表了你对“已提交”消息的定义。 acks的默认值为1，为1代表消息被leader副本接收就算作被成功发送。配置为all代表所有的副本都接受这个消息之后消息才算被成功发送。  
④设置replication.factor>=3，为了保证数据安全性，一般会保证每个分区至少有3个副本。  
⑤设置min.insync.replicas>1，代表消息至少要被写入2个副本才算被发送成功。min.insync.replicas的默认值为1，在实际生产中尽量避免默认值为1。  
⑥确保 replication.factor > min.insync.replicas。如果两者相等，那么只要有一个副本挂 机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要 在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas + 1。  
⑦设置 unclean.leader.election.enable = false。这是 Broker 端的参数，意思是如果一个 Broker 落后原先的 Leader 太多，就不允许被设置为新的 Leader。  
⑧确保消息消费完成再提交。Consumer 端有个参数 enable.auto.commit，最好把它设 置成 false，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程 处理的场景而言是至关重要的。

#### 重复发送消息，怎么保证消息不被重复消费呢？ ####

让消费消息的操作具有幂等性，一个操作具有幂等性是指这个操作任意多次执行所产生的影响与一次执行的影响相同。具体的实现有：

1.  利用数据库的唯一约束实现幂等  
    每当要消费一条消息时，就先将其插入到数据库中，由于唯一性约束，相同的消息只会被插入一次。
2.  利用redis的SETNX命令也可以

#### 消息积压你是怎么处理的？ ####

设计系统时，要保证消费端的消费性能要高于生产者的发送性能，这样系统才能健康持续运行。  
消息积压产生的原因有2种：第一是消息生产速度过快，另一种是消息消费速度变慢。可以通过查看消息队列内置的监控数据，确定是生产端还是消费端的原因。

如果是消费端的原因，就进行消费端性能优化：

1.  首先可以优化消费业务逻辑，尽量减少冗余。还可以增加消费端的并发数，也就是扩容Consumer实例，也必须同步扩容主题中的队列数量，确保Consumer的实例数量和队列数量相等。
2.  如果是有大促或者抢购导致消息突增，短时间内不可能通过优化代码来解决，唯一的方法是通过增加消费者实例数来提升总体的消费能力。如果短时间内没有足够的服务器资源进行扩容，那么可以将系统降级，关闭一些不重要的业务，减少生产者生产的消息数量，让系统尽可能正常运转，服务一些重要的业务。

如果通过监控发现生产消息的速度和消费消息的速度都没有什么异常，那就需要检查一下消费端，是不是消费失败导致了一条消息反复消息。

如果监控到消费变慢了，需要检查一下消费实例，分析一下是什么原因导致消费变慢。可以检查一下日志看是否有大量的消费错误，还可以通过打印堆栈信息，看一下消费线程是不是卡在什么地方了，比如发生死锁或者等待资源。

#### 你是怎么保证消息按顺序消费？ ####

RocketMQ一个topic可以有多个分区，它可以分区中顺序消费，但是不能保证每个topic顺序消费，如果想要保证每个topic顺序消费的话，那么就只为一个topic设置一个分区。  
或者发送消息时指定发送的分区。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzMzODUxOQ_size_16_color_FFFFFF_t_70]: /images/20230528/60ca0cdcd01a463797248b8128ea4f97.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzMzODUxOQ_size_16_color_FFFFFF_t_70 1]: /images/20230528/d18aae7a9a554685b01b837844b5efdb.png