大兴黄村网站建设,怎么快速建设小型外贸网站,深圳高端女装品牌,人才招聘网官网作者#xff1a;mousycodersegmentfault.com/a/1190000021054802消息队列连环炮项目里怎么样使用 MQ 的#xff1f;为什么要使用消息队列#xff1f;消息队列有什么优点和缺点#xff1f;kafka,activemq,rabbitmq,rocketmq 都有什么去呗#xff1f;如何保证消息队列高可用… 作者mousycodersegmentfault.com/a/1190000021054802消息队列连环炮项目里怎么样使用 MQ 的为什么要使用消息队列消息队列有什么优点和缺点kafka,activemq,rabbitmq,rocketmq 都有什么去呗如何保证消息队列高可用如何保证消息不被重复消费如何保证消息的可靠性传输如何保证消息的顺序性写一个消息队列架构设计消息队列技术选型解决的问题解耦异步削峰不用 MQ 系统耦合场景A 系统产生了一个比较关键的数据很多系统需要 A 系统将数据发过来强耦合(B,C,D,E 系统可能参数不一样、一会需要一会不需要数据A 系统要不断修改代码维护)A 系统还要考虑 B、C、D、E 系统是否挂了是否访问超时是否重试使用 MQ 系统解耦场景维护这个代码不需要考虑人家是否调用成功失败超时如果新系统需要数据直接从 MQ 里消费即可如果某个系统不需要这条数据就取消对 MQ 消息的消费即可。总结通过一个 MQ 的发布订阅消息模型(Pub/Sub), 系统 A 跟其他系统就彻底解耦了。不用 MQ 同步高延迟请求场景一般互联网类的企业对用户的直接操作一般要求每个请求都必须在 200ms以内对用户几乎是无感知的。使用 MQ 进行异步化之后的接口性能优化提高高延时接口没有用 MQ 时高峰期系统被打死的场景高峰期每秒 5000 个请求每秒对 MySQL 执行 5000 条 SQL(一般MySQL每秒 2000 个请求差不多了)如果MySQL被打死然后整个系统就崩溃用户就没办法使用系统了。但是高峰期过了之后每秒钟可能就 50 个请求对整个系统没有任何压力。使用 MQ 进行削峰的场景5000 个请求写入到 MQ 里面系统 A 每秒钟最多只能处理 2000 个请求(MySQL 每秒钟最多处理 2000 个请求)系统 A 从 MQ 里慢慢拉取请求每秒钟拉取 2000 个请求。MQ每秒钟 5000 个请求进来结果只有 2000 个请求出去结果导致在高峰期(21小时)可能有几十万甚至几百万的请求积压在 MQ 中这个是正常的因为过了高峰期之后每秒钟就 50 个请求但是系统 A 还是会按照每秒 2000 个该请求的速度去处理。只要高峰期一过系统 A 就会快速的将积压的消息给解决掉。算一笔账每秒积压在 MQ 里消息有 3000 条一分钟就会积压 18W 条消息一个小时就会积压 1000 万条消息。等高峰期一过差不多需要 1 个多小时就可以把 1000W 条积压的消息给处理掉架构中引入 MQ 后存在的问题系统可用性降低MQ 可能挂掉导致整个系统崩溃系统复杂性变高可能发重复消息导致插入重复数据消息丢了消息顺序乱了系统 B,C,D 挂了导致 MQ 消息积累磁盘满了一致性问题本来应该A,B,C,D 都执行成功了再返回结果A,B,C 执行成功 D 失败Kafka、ActiveMQ、RabbitMQ、RocketMQ 有什么优缺点建议中小型公司 RabbitMQ 大公司RocketMQ 大数据实时计算Kafka消息队列高可用RabbtitMQ 高可用RabbitMQ有三种模式单机模式 、普通集群模式、镜像集群模式单机模式demo级普通集群模式(非高可用)队列的元数据存在于多个实例中但是消息不存在多个实例中每次多台机器上启动多个 rabbitmq 实例每个机器启动一个。优点可以多个机器消费消息可以提高消费的吞吐量缺点可能会在 rabbitmq 内部产生大量的数据传输 可用性基本没保障queue 所在机器宕机就没办法消费了没有高可用性可言镜像集群模式(高可用非分布式)队列的元数据和消息都会存在于多个实例中每次写消息到 queue的时候都会自动把消息到多个实例的 queue 里进行消息同步。也就 是每个节点上都有这个 queue 的一个完整镜像(这个 queue的全部数据)。任何一个节点宕机了其他节点还包含这个 queue的完整数据其他 consumer 都可以到其他活着的节点上去消费数据都是 OK 的。缺点不是分布式的如果这个 queue的数据量很大大到这个机器上的容量无法容纳 。开启镜像集群模式方法管理控制台Admin页面下新增一个镜像集群模式的策略指定的时候可以要求数据同步到所有节点也可以要求同步到指定数量的节点然后你再次创建 queue 的时候 应用这个策略就 会自动将数据同步到其他的节点上去。Kafka 高可用架构broker进程就是kafka在每台机器上启动的自己的一个进程。每台机器机器上的broker进程就可以认为是 kafka集群中的一个节点。你创建一个 topic,这个topic可以划分为多个 partition,每个 partition 可以存在于不同的 broker 上每个 partition就存放一部分数据。这就是天然的分布式消息队列也就是说一个 topic的数据是分散放在 多个机器上的每个机器就放一部分数据。分布式的真正含义是每个节点只放一部分数据而不是完整数据(完整数据就是HA、集群机制)Kafka 0.8版本之前是没有 HA 机制的任何一个 broker 宕机了那么就缺失一部分数据。Kafka 0.8以后提供了 HA 机制就是 replica 副本机制。每个 partition的数据都会同步到其他机器上形成自己的多个 replica 副本。然后所有 replica 会选举一个 leader。那么生产者、消费者都会和这个 leader 打交道然后其他 replica 就是 follow。写的时候leader 负责把数据同步到所有 follower上去读的时候就直接读 leader 上的数据即可。如果某个 broker宕机了刚好也是 partition的leader那么此时会选举一个新的 leader出来大家继续读写那个新的 leader即可这个就 是所谓的高可用性。更多面试题面试题内容聚合leader和follower的同步机制写数据的时候生产者就写 leader然后 leader将数据落地写本地磁盘接着其他 follower 自己主动从 leader来pull数据。一旦所有 follower同步好数据了就会发送 ack给 leaderleader收到所有 follower的 ack之后就会返回写成功的消息给生产者。消费的时候只会从 leader去读但是只有一个消息已经被所有 follower都同步成功返回 ack的时候这个消息才会被消费者读到。消息队列重复数据MQ 只能保证消息不丢不能保证重复发送Kafka 消费端可能出现的重复消费问题每条消息都有一个 offset 代表 了这个消息的顺序的序号按照数据进入 kafka的顺序kafka会给每条数据分配一个 offset,代表了这个是数据的序号消费者从 kafka去消费的时候按照这个顺序去消费消费者会去提交 offset就是告诉 kafka已经消费到 offset153这条数据了 zk里面就记录了消费者当前消费到了 offset 几的那条消息假如此时消费者系统被重启重启之后消费者会找kafka让kafka把上次我消费到的那个地方后面的数据继续给我传递过来。更多面试题面试题内容聚合重复消息原因(主要发生在消费者重启后)消费者不是说消费完一条数据就立马提交 offset的而是定时定期提交一次 offset。消费者如果再准备提交 offset但是还没提交 offset的时候消费者进程重启了那么此时已经消费过的消息的 offset并没有提交kafka也就不知道你已经消费了 offset 153那条数据这个时候kafka会给你发offset152,153,154的数据此时 offset 152,153的消息重复消费了保证 MQ 重复消费幂等性幂等一个数据或者一个请求给你重复来多次你得确保对应的数据是不会改变的不能出错。思路拿数据要写库首先检查下主键如果有数据则不插入进行一次update如果是写 redis就没问题反正每次都是 set 天然幂等性生产者发送消息的时候带上一个全局唯一的id,消费者拿到消息后先根据这个id去 redis里查一下之前有没消费过没有消费过就处理并且写入这个 id 到 redis如果消费过了则不处理。基于数据库的唯一键保证 MQ 消息不丢MQ 传递非常核心的消息比如广告计费系统用户点击一次广告扣费一块钱如果扣费的时候消息丢了则会不断少钱积少成多对公司是一个很大的损失。RabbitMQ可能存在的数据丢失问题生产者写消息的过程中消息都没有到 rabbitmq在网络传输过程中就丢了。或者消息到了 rabbitmq但是人家内部出错了没保存下来。RabbitMQ 接收到消息之后先暂存在主机的内存里结果消费者还没来得及消费RabbitMQ自己挂掉了就导致暂存在内存里的数据给搞丢了。消费者消费到了这个消费但是还没来得及处理自己就挂掉了RabbitMQ 以为这个消费者已经处理完了。问题 1解决方案事务机制(一般不采用同步的生产者发送消息会同步阻塞卡住等待你是成功还是失败。会导致生产者发送消息的吞吐量降下来) channel.txSelecttry { //发送消息} catch(Exception e){ channel.txRollback; //再次重试发送这条消息} channel.txCommit;confirm机制(一般采用这种机制异步的模式不会阻塞吞吐量会比较高)先把 channel 设置成 confirm 模式发送一个消息到 rabbitmq发送完消息后就不用管了rabbitmq 如果接收到了这条消息就会回调你生产者本地的一个接口通知你说这条消息我已经收到了rabbitmq 如果在接收消息的时候报错了就会回调你的接口告诉你这个消息接收失败了你可以再次重发。public void ack(String messageId){}public void nack(String messageId){ //再次重发一次这个消息}问题 2 解决方案持久化到磁盘创建queue的时候将其设置为持久化的这样就可以保证 rabbitmq持久化queue的元数据但是不会持久化queue里的数据发送消息的时候将 deliveryMode 设置为 2将消息设置为持久化的此时 rabbitmq就会将消息持久化到磁盘上去。必须同时设置 2 个持久化才行。持久化可以跟生产者那边的 confirm机制配合起来只有消息被持久化到磁盘之后才会通知生产者 ack了 所以哪怕是在持久化到磁盘之前 rabbitmq挂了数据丢了生产者收不到 ack你也可以自己重发。缺点可能会有一点点丢失数据的可能消息刚好写到了 rabbitmq中但是还没来得及持久化到磁盘上结果不巧 rabbitmq挂了会导致内存里的一点点数据会丢失。更多面试题面试题内容聚合问题 3 解决方案原因消费者打开了 autoAck机制(消费到一条消息还在处理中还没处理完此时消费者自动 autoAck了通知 rabbitmq说这条消息已经消费了此时不巧消费者系统宕机了那条消息丢失了还没处理完而且 rabbitmq还以为这个消息已经处理掉了)解决方案关闭 autoAck,自己处理完了一条消息后再发送 ack给 rabbitmq,如果此时还没处理完就宕机了此时rabbitmq没收到你发的ack消息然后 rabbitmq 就会将这条消息重新分配给其他的消费者去处理。Kafka 可能存在的数据丢失问题消费端弄丢数据原因消费者消费到那条消息后自动提交了 offsetkafka以为你已经消费好了这条消息结果消费者挂了这条消息就丢了。例子消费者消费到数据后写到一个内存 queue里缓存下消息自动提交 offset重启了系统结果会导致内存 queue 里还没来得及处理的数据丢失。解决方法kafka会自动提交 offset那么只要关闭自动提交 offset在处理完之后自己手动提交可以保证数据不会丢。但是此时确实还是会重复消费比如刚好处理完还没提交 offset结果自己挂了此时肯定会重复消费一次 做好幂等即可。Kafka 丢掉消息原因kafka 某个 broker 宕机然后重新选举 partition 的 leader时此时其他的 follower 刚好还有一些数据没有同步结果此时 leader挂了然后选举某个 follower成 leader之后就丢掉了之前leader里未同步的数据。更多面试题面试题内容聚合例子kafka的leader机器宕机将 follower 切换为 leader之后发现数据丢了解决方案(保证 kafka broker端在 leader发生故障或者leader切换时数据不会丢)给 topic设置 replication.factor 这个值必须大于 1保证每个 partition 必须至少有 2 个副本在 kafka 服务端设置 min.insync.replicas 参数这个值必须大于 1这个是要求一个leader至少感知到有至少一个follower还跟自己保持联系没掉队这样才能确保 leader挂了还有一个follower保证至少一个 follower能和leader保持正常的数据同步。在 producer 端设置 acks all这个是要求每条数据必须是写入所有 replica 之后才能认为是写成功了。否则会生产者会一直重试此时设置 retries MAX(很大的重试的值),要求一旦写入失败就卡在这里(避免消息丢失)kafka 生产者丢消息按 2 的方案设置了 ack all一定不会丢。它会要求 leader 接收到消息所有的 follower 都同步 到了消息之后才认为本次写成功。如果没满足这个条件生产者会无限次重试 。消息队列顺序性背景mysql binlog 同步的系统在mysql里增删改一条数据对应出来了增删改 3 条binlog接着这 3 条binlog发送到 MQ 里面到消费出来依次执行起码是要保证顺序的吧不然顺序变成了 删除、修改、增加。日同步数据达到上亿mysql-mysql,比如大数据 team需要同步一个mysql库来对公司的业务系统的数据做各种复杂的操作。场景rabbitmq,一个queue,多个consumer这不明显乱了kafka,一个topic一个partition,一个consumer内部多线程这不也乱了RabbitMQ 消息顺序错乱RabbitMQ 如何保证消息顺序性需要保证顺序的数据放到同一个queue里Kafka 消息顺序错乱写入一个 partition中的数据一定是有顺序的。生产者在写的时候可以指定一个 key比如订单id作为key,那么订单相关的数据一定会被分发到一个 partition中区此时这个 partition中的数据一定是有顺序的。Kafka 中一个 partition 只能被一个消费者消费。消费者从partition中取出数据的时候 一定是有顺序的。Kafka 保证消息顺序性如果消费者单线程消费处理如果处理比较耗时处理一条消息是几十ms一秒钟只能处理几十条数据这个吞吐量太低了。肯定要用多线程去并发处理压测消费者4 核 8G 单机32 条线程最高每秒可以处理上千条消息消息队列延迟以及过期失效消费端出了问题不消费了或者消费极其慢。接着坑爹了你的消息队列集群的磁盘都快写满了 都没人消费怎么办积压了几个小时rabbitmq设置了消息过期时间后就没了怎么办例如每次消费之后都要写 mysql结果mysql挂了消费端 hang 不动了。消费者本地依赖的一个东西挂了导致消费者挂了。长时间没处理消费导致 mq 写满了。场景几千万条数据再 MQ 里积压了七八个小时快速处理积压的消息一个消费者一秒是 1000 条一秒 3 个消费者是 3000 条一分钟是 18W 条1000 多 W 条需要一个小时恢复。步骤先修复 consumer 的问题确保其恢复消费速度然后将现有的 consumer 都停掉新建一个topic,partition是原来的 10 倍临时建立好原先 10 倍或者 20 倍的 queue 数量然后写一个临时的分发数据的 consumer 程序这个程序部署上去消费积压的数据消费之后不做耗时的处理直接均匀轮询写入临时建立好的 10 倍数量的 queue接着临时征用 10 倍的机器来部署 consumer,每一批 consumer 消费一个临时 queue 的数据这种做法相当 于是临时将 queue 资源和 consumer 资源扩大 10 倍以正常 10 倍速度等快速消费完积压数据之后恢复原先部署架构 重新用原先的 consumer机器消费消息原来 3 个消费者需要 1 个小时可以搞定现在 30 个临时消费者需要 10 分钟就可以搞定。如果用的 rabbitmq并且设置了过期时间如果此消费在 queue里积压超过一定的时间会被 rabbitmq清理掉数据直接搞丢。这个时候开始写程序将丢失的那批 数据查出来然后重新灌入mq里面把白天丢的数据补回来。如果消息积压mq长时间没被处理掉导致mq快写完满了你临时写一个程序接入数据来消费写到一个临时的mq里再让其他消费者慢慢消费 或者消费一个丢弃一个都不要了快速消费掉所有的消息然后晚上补数据。如何设计消息队列中间件架构mq要支持可伸缩性快速扩容。设计一个分布式的 MQbroker-topic-partition每个 partition 放一个机器就存一部分数据。如果现在资源不够给 topic 增加 partition 然后做数据迁移增加机器。mq数据落磁盘避免进程挂了数据丢了顺序写这样就没有磁盘随机读写的寻址开销磁盘顺序读写的性能是很高的这个就是 kafka的思路。mq高可用性。多副本-leader follower- broker 挂了重新选举 leader 对外提供服务支持数据 0 丢失。ENDJava面试题专栏【61期】MySQL行锁和表锁的含义及区别(MySQL面试第四弹)【62期】解释一下MySQL中内连接外连接等的区别(MySQL面试第五弹)【63期】谈谈MySQL 索引B树原理以及建索引的几大原则(MySQL面试第六弹)【64期】MySQL 服务占用cpu 100%如何排查问题? (MySQL面试第七弹)【65期】Spring的IOC是啥?有什么好处?【66期】Java容器面试题谈谈你对 HashMap 的理解【67期】谈谈ConcurrentHashMap是如何保证线程安全的【68期】面试官对并发熟悉吗说说Synchronized及实现原理【69期】面试官对并发熟悉吗谈谈线程间的协作(wait/notify/sleep/yield/join)【70期】面试官对并发熟悉吗谈谈对volatile的使用及其原理我知道你 “在看”
