消息队列核心知识点_rocketmq

消息队列核心知识点_rocketmq
强烈推介IDEA2020.2破解激活,IntelliJ IDEA 注册码,2020.2 IDEA 激活码

Part I

1、如何选择消息队列?

(1) 如果对消息队列功能和性能都没有很高的要求,只需一个开箱即用易维护的产品,建议使用RabbitMQ。

  • 有良好的运维界面,仅仅只是使用消息队列功能,用于异步和业务模块解耦,对性能要求不是很高。

(2)如果系统使用消息队列的场景是处理在线业务,比如在交易系统中用消息队列传递订单,那么使用RocketMQ。

  • 阿里内部也是使用RocketMQ作为支撑其业务的消息队列,经历多次“双十一”考验,它的性能、稳定性和可靠性都是值得信赖的。

  • 有非常活跃的中文社区,大多数问题都可以找到中文答案。使用Java语言开发,它的贡献者大多数为中国人,源代码相对比较易懂或易进行扩展和二次开发。

  • RocketMQ对在线业务的响应时延做了很多优化,大多数情况下可以做到毫秒级的响应,如果你的应用场景很在意响应时延,那应该选择使用RocketMQ。

(3)如果需要处理海量的消息,想收集日志、监控信息或是前端埋点这类数据,或应用场景大量使用了大数据、流计算(做事后的统计分析)相关的开源产品,那kafka是最适合的。

2、MQ是用来干嘛的?有什么作用?

MQ其实就是一种系统,独立部署,让系统之间通过发消息和收消息来进行异步的调用

作用:提升性能、系统解耦、流量消峰

  • 提升性能:
    一个请求调用了A、B两个系统,执行业务逻辑各需要20 、200毫秒,那么处理这个请求一共需要220毫秒
    引入MQ后:发送消息给MQ的速度是很快的(没有业务逻辑、没有数据库操作),所以引入MQ后,20多毫秒就可以返回结果给用户了。
  • 系统解耦:
    系统A和系统B通过同步调用的模式耦合在了一起,一旦系统B出现故障,很可能会影响系统A也有故障
    而且系统A还得去关心系统B的故障,去处理对应的异常,这是很麻烦的。
    引入MQ后:B如果出现了故障,对系统A根本没影响,系统A也感觉不到,B自己处理自己的问题!
  • 流量消峰:
    如果高并发访问系统A(A没有数据库操作),A调用B(B有数据库操作),那么瓶颈在B,因为数据库操作是比较耗时的。
    同样的机器配置下,如果数据库可以抗每秒6000请求,MQ至少可以抗每秒几万请求。因为数据库复杂,需要支持事务、复杂的SQL查询等
    引入MQ后:A系统依赖支持高并发的MQ,B也依赖MQ,此时B可以用自己的合适的速度访问MQ,即B系统流量被消峰了。整个系统的性能由A决定,而不速度慢的B决定

3、什么是RabbitMQ?

RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue 高级消息队列协议 )的开源实现,能够实现异步消息处理。

4、什么是RocketMQ?

RocketMQ是一个低延时、高可靠、可伸缩、易于使用的分布式消息中间件,是由阿里巴巴开源捐献给Apache的顶级项目。RocketMQ具有高吞吐、低延迟、海量消息堆积等优点,同时提供顺序消息、事务消息、定时消息、消息重试于追踪等功能,非常适合在电商、金融等领域使用。

5、RocketMQ包含了几个核心部分?

NameServer集群、Broker集群、生产者、消费者

  • NameServer
    负责管理所有的Broker消息
    让生产者和消费者鬼知道集群里有哪些Broker,然后与之通信
  • Broker
    实现数据多副本存储和高可用,使用主从架构
  • 生产者
    向MQ发送消息
  • 消费者
    从MQ获取消息

6、RocketMQ持久化原理

现代的磁盘都是高性能的,写速度并不一定比网络的数据传输速度慢。比如 SSD 固态硬盘顺序写的速度可以达到 1500 MB/s,就算是普通磁盘,如果性能比较高的话,顺序写的速度可以达到 450MB/s~600MB/s。但是这是顺序写的速度。但实际上磁盘采用的是随机写,在随机写的情况下,磁盘的写入速度急速下降,磁盘的随机写速度可能只有几百KB/s,这远远要慢于网络传输速度,所以它并不能满足高性能的要求。
RocketMQ 在持久化的设计上,采取的是消息顺序写、随机读的策略,利用磁盘顺序写的速度,让磁盘的写速度不会成为系统的瓶颈。并且采用 MMPP 这种“零拷贝”技术,提高消息存盘和网络发送的速度。极力满足 RocketMQ 的高性能、高可靠要求。
在 RocketMQ 持久化机制中,涉及到了三个角色:
CommitLog:消息真正的存储文件,所有消息都存储在 CommitLog 文件中。
ConsumeQueue:消息消费逻辑队列,类似数据库的索引文件。
IndexFile:消息索引文件,主要存储消息 Key 与 offset 对应关系,提升消息检索速度。
写入:CommitLog -->文件是存放消息数据的地方,所有的消息都将存入到 CommitLog 文件中。生产者将消息发送到 RocketMQ 的 Broker 后,Broker 服务器会将消息顺序写入到 CommitLog 文件中,这也就是 RocketMQ 高性能的原因,因为我们知道磁盘顺序写特别快,RocketMQ 充分利用了这一点,极大的提高消息写入效率。
读取:Queue–>但是同一主题的消息在 CommitLog 文件中可能是不连续的,那么消费者消费消息的时候,需要将 CommitLog 文件加载到内存中遍历查找订阅主题下的消息,频繁的 IO 操作,性能就会急速下降。为了解决这个问题,RocketMQ 引入了 Queue 文件。Queue 文件可以看作是索引文件,类似于 MySQL 中的二级索引。在存放了同一主题下的所有消息,消费者消费的时候只需要去对应的 Queue 组中取消息即可。

Part II

1、如何保证消息队列的高可用?

高可用HA(High Availability)是分布式系统架构设计中必须考虑的因素之一,它通常是指通过设计减少系统不能提供服务的时间。
RocketMQ 部署方式:

  • 单个 Master。这种方式风险较大,一旦Broker重启或者宕机时,会导致整个服务不可用,不建议线上环境使用。
  • 多 Master 模式(2m-noslave)。一个集群无 Slave,全是 Master,例如 2 个 Master 或者 3 个 Master
    优点:配置简单,单个Master 宕机或重启维护对应用无影响,在磁盘配置为RAID10 时,即使机器宕机不可恢 复情况下,由与 RAID10磁盘非常可靠,消息也不会丢(异步刷盘丢失少量消息,同步刷盘一条不丢)。性能最高。
    缺点:单台机器宕机期间,这台机器上未被消费的消息在机器恢复之前不可订阅,消息实时性会受到受到影响。
  • 多 Master 多 Slave 模式,异步复制(2m-2s-async)。每个 Master 配置一个 Slave,有多对Master-Slave,HA。采用异步复制方式,主备有短暂消息延迟,毫秒级。
    优点:即使磁盘损坏,消息丢失的非常少,且消息实时性不会受影响,因为Master 宕机后,消费者仍然可以从 Slave消费,此过程对应用透明。不需要人工干预。性能同多 Master 模式几乎一样。
    缺点:Master 宕机,磁盘损坏情况,会丢失少量消息。
  • 多 Master 多 Slave 模式,同步双写(2m-2s-sync)。每个 Master 配置一个 Slave,有多对Master-Slave,HA。采用同步双写方式,主备都写成功,向应用返回成功。
    优点:数据与服务都无单点,Master宕机情况下,消息无延迟,服务可用性与数据可用性都非常高
    缺点:性能比异步复制模式略低,大约低10%左右,发送单个消息的RT会略高。目前主宕机后,备机不能自动切换为主机,后续会支持自动切换功能。

2、如何保证消息不被重复消费(如何保证消息消费时的幂等性)?

其实重复消费不可怕,可怕的是你没考虑到重复消费之后,怎么保证幂等性。

幂等性,通俗点说,就一个数据,或者一个请求,给你重复来多次,你得确保对应的数据是不会改变的,不能出错。

其实还是得结合业务来思考,我这里给几个思路:

  • 比如你拿个数据要写库,你先根据主键查一下,如果这数据都有了,你就别插入了,update一下好吧
  • 比如你是写redis,那没问题了,反正每次都是set,天然幂等性
  • 比如你不是上面两个场景,那做的稍微复杂一点,你需要让生产者发送每条数据的时候,里面加一个全局唯一的id,类似订单id之类的东西,然后你这里消费到了之后,先根据这个id去比如redis里查一下,之前消费过吗?如果没有消费过,你就处理,然后这个id写redis。如果消费过了,那你就别处理了,保证别重复处理相同的消息即可。

3、如何保证消息的可靠性传输(如何处理消息丢失的问题)?

可以从三个方面来分析rocket的消息可靠性

  • Producer端消息丢失
    首先消息分为普通消息(同步,异步,单向发送),定时延时消息,顺序消息(不能保证全局有序,只能保证同一个quen中有序),事物消息。
    producer端防止消息发送失败,可以采用同步阻塞式的发送(也就是发送同步消息),同步的检查Brocker返回的状态是否持久化成功,发送超时或者失败,则会默认重试2次,rocker选择了确保消息一定发送成功,但有可能发生重复投递。
    如果是异步发送消息,会有一个回调接口,当brocker存储成功或者失败的时候,也可以在这里根据返回状态来决定是否需要重试(当然这个是需要我们自己来实现的)。
  • Brocker端消息丢失
    rocketmq一般都是先把消息写到PageCache中,然后再持久化到磁盘上,数据从pagecache刷新到磁盘有两种方式,同步和异步。
    同步刷盘方式:消息写入内存的 PageCache后,立刻通知刷盘线程刷盘,然后等待刷盘完成,刷盘线程执行完成后唤醒等待的线程,返回消息写成功的状态。这种方式可以保证数据绝对安全,但是吞吐量不大。
    异步刷盘方式(默认):消息写入到内存的 PageCache中,就立刻给客户端返回写操作成功,当 PageCache中的消息积累到一定的量时,触发一次写操作,将 PageCache中的消息写入到磁盘中。这种方式吞吐量大,性能高,但是 PageCache中的数据可能丢失,不能保证数据绝对的安全。
  • Cousmer端消息丢失
    cousmer端默认是消息之后自动返回消费成功确认ack,但是这时如果我们的程序执行失败了,数据不就丢失了吗?
    所以我们可以将自动提交(AutoCommit)消费响应,设置为在代码中手动提交,只有真正消费成功之后再通知brocker消费成功,然后更新消费唯一offset或者删除brocker中的消息。

4、如何保证消息的顺序性?

  • rabbitmq:拆分多个queue,每个queue一个consumer,就是多一些queue而已,确实是麻烦点;或者就一个queue但是对应一个consumer,然后这个consumer内部用内存队列做排队,然后分发给底层不同的worker来处理
  • rocketmq:同一条queue里面保证FIFO的。把消息确保投递到同一条queue。一个订单的顺序流程是:创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中,消费时,同一个OrderId获取到的肯定是同一个队列。

5、如何解决消息队列的延时以及过期失效问题?消息队列满了以后该怎么处理?

(1)一般这个时候,只能操作临时紧急扩容了,具体操作步骤和思路如下:

  • 先修复consumer的问题,确保其恢复消费速度,然后将现有cnosumer都停掉;
  • 新建一个topic,partition是原来的10倍,临时建立好原先10倍或者20倍的queue数量;
  • 然后写一个临时的分发数据的consumer程序,这个程序部署上去消费积压的数据,消费之后不做耗时的处理,直接均匀轮询写入临时建立好的10倍数量的queue‘;
  • 接着临时征用10倍的机器来部署consumer,每一批consumer消费一个临时queue的数据;
  • 这种做法相当于是临时将queue资源和consumer资源扩大10倍,以正常的10倍速度来消费数据;
  • 等快速消费完积压数据之后,得恢复原先部署架构,重新用原先的consumer机器来消费消息。

假设你用的是rabbitmq,rabbitmq是可以设置过期时间的,就是TTL,如果消息在queue中积压超过一定的时间就会被rabbitmq给清理掉,这个数据就没了。那这就是第二个坑了。这就不是说数据会大量积压在mq里,而是大量的数据会直接搞丢。
假设1万个订单积压在mq里面,没有处理,其中1000个订单都丢了,你只能手动写程序把那1000个订单给查出来,手动发到mq里去再补一次。

(2)如果走的方式是消息积压在mq里,那么如果你很长时间都没处理掉,此时导致mq都快写满了,咋办?这个还有别的办法吗?没有,谁让你第一个方案执行的太慢了,你临时写程序,接入数据来消费,消费一个丢弃一个,都不要了,快速消费掉所有的消息。

6、如果让你写一个消息队列,该如何进行架构设计啊?说一下你的思路。

比如说这个消息队列系统,我们来从以下几个角度来考虑一下:

  • 可伸缩性。就是需要的时候快速扩容,就可以增加吞吐量和容量,参照一下kafka的设计理念,broker -> topic -> partition,每个partition放一个机器,就存一部分数据。如果现在资源不够了,简单啊,给topic增加partition,然后做数据迁移,增加机器,不就可以存放更多数据,提供更高的吞吐量了。
  • 高可用性。
  • 高可靠性。考虑一下这个mq的数据要不要落地磁盘吧?那肯定要了,落磁盘,才能保证别进程挂了数据就丢了。那落磁盘的时候怎么落啊?顺序写,这样就没有磁盘随机读写的寻址开销,磁盘顺序读写的性能是很高的,这就是kafka的思路。
本文来源jackaroo2020,由架构君转载发布,观点不代表Java架构师必看的立场,转载请标明来源出处:https://javajgs.com/archives/7023

发表评论