简单电商网站模板下载,抖音的商业营销手段,手机如何建设网站,网站建设及第三方支付文章目录 Kafka、RabbitMQ、RocketMQ 之间的区别是什么#xff1f;性能数据可靠性服务可用性功能 RabbitMQ如何保证消息不丢失#xff1f;Kafka 的架构说一下#xff1f;Kafka 怎么保证消息是有序的#xff1f;Kafka 怎么解决重复消费#xff1f;Kafka 怎么保证消息不丢失… 文章目录 Kafka、RabbitMQ、RocketMQ 之间的区别是什么性能数据可靠性服务可用性功能 RabbitMQ如何保证消息不丢失Kafka 的架构说一下Kafka 怎么保证消息是有序的Kafka 怎么解决重复消费Kafka 怎么保证消息不丢失RocketMQ 如何监听消息的RocketMQ 常见的面试题哪个环节会有消息丢失的可能RocketMQ消息零丢失方案1、生产者使用事务消息机制保证消息零丢失 2、**RocketMQ**配置同步刷盘**Dledger**主从架构保证**MQ**自身不会丢消息3、消费者端不要使用异步消费机制4、RocketMQ特有的问题NameServer挂了如何保证消息不丢失 使用RocketMQ如何快速处理积压消息1、如何确定RocketMQ有大量的消息积压2、如何处理大量积压的消息 RocketMQ的消息轨迹RocketMQ 如何处理流量削峰使用RocketMQ如何保证消息顺序 Kafka、RabbitMQ、RocketMQ 之间的区别是什么
性能
消息中间件的性能主要衡量吞吐量Kafka的吞吐量比RabbitMQ要高出1~2个数量级RabbitMQ的单机QPS在万级别Kafka的单机QPS能够达到百万级别。RocketMQ单机写入TPS单实例约7万条/秒单机部署3个Broker可以跑到最高12万条/秒消息大小10个字节Kafka如果开启幂等、事务等功能性能也会有所降低。
数据可靠性
Kafka与RabbitMQ 都具备多副本机制数据可靠性较高。RocketMQ支持异步实时刷盘同步刷盘同步Replication拷贝异步Replication数据可靠性高。
服务可用性
Kafka采用集群部署分区与多副本的设计使得单节点宕机对服务无影响并且支持消息容量的线性提升。RabbitMQ支持集群部署集群节点数量有多种规格。RocketMQ是分布式架构可用性高。
功能
Kafka与RabbitMQ都是比较主流的两款消息中间件具备消息传递的基本功能但在一些特殊的功能方面存在差异RocketMQ在阿里集团内部有大量的应用在使用。
RabbitMQ如何保证消息不丢失 Kafka 的架构说一下
整个架构中包括三个角色 ⚫ 生产者Producer消息和数据生产者。 ⚫ 代理Broker缓存代理Kafka的核心功能。 ⚫ 消费者Consumer消息和数据消费者。 Kafka给Producer和Consumer提供注册的接口数据从Producer发送到BrokerBroker承担一个中间缓存和分发的作用负责分发注册到系统中的Consumer。
Kafka 怎么保证消息是有序的
消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量offset。Kafka 通过偏移量offset来保证消息在分区内的有序性。发送消息的时候指定key/Partition。
Kafka 怎么解决重复消费
⚫ 生产者发送每条数据的时候里面加一个全局唯一的id消费到了之后先根据这个id去比对Redis里查一下之前是否消费过如果没有消费过就处理然后把这个id写入到Redis中。如果消费过就别处理了。 ⚫ 基于数据库的唯一键来保证重复数据不会重复插入多条。因为有唯一键约束了重复数据插入只会报错不会导致数据库中出现脏数据。
Kafka 怎么保证消息不丢失
生产者丢失消息的情况
生产者(Producer) 调用send方法发送消息之后消息可能因为网络问题并没有发送过去。 为了确定消息是发送成功我们要判断消息发送的结果Kafka 生产者(Producer) 使用 send 方法发送消息实际上是异步的操作我们可以通过 get()方法获取调用结果但是这样也让它变为了同步操作可以采用为其添加回调函数的形式示例代码如下 ListenableFutureSendResultString, Object future kafkaTemplate.send(topic, o); future.addCallback(result - logger.info(“生产者成功发送消息到topic:{} partition:{}的消息”, result.getRecordMetadata().topic(), result.getRecordMetadata().partition()), ex - logger.error(“生产者发送消失败原因{}”, ex.getMessage())); Producer 的retries重试次数设置一个比较合理的值一般是 3 但是为了保证消息不丢失的话一般会设置比较大一点。设置完成之后当出现网络问题之后能够自动重试消息发送避免消息丢失。另外建议还要设置重试间隔因为间隔太小的话重试的效果就不明显了网络波动一次你3次一下子就重试完了
消费者丢失消息的情况 当消费者拉取到了分区的某个消息之后消费者会自动提交了 offset。自动提交的话会有一个问题试想一下当消费者刚拿到这个消息准备进行真正消费的时候突然挂掉了消息实际上并没有被消费但是 offset 却被自动提交了。 解决办法也比较粗暴我们手动关闭自动提交 offset每次在真正消费完消息之后再自己手动提交 offset 。 但是细心的朋友一定会发现这样会带来消息被重新消费的问题。比如你刚刚消费完消息之后还没提交 offset结果自己挂掉了那么这个消息理论上就会被消费两次。
Kafka 弄丢了消息 试想一种情况假如 leader 副本所在的 broker 突然挂掉那么就要从 follower 副本重新选出一个 leader 但是 leader 的数据还有一些没有被 follower 副本的同步的话就会造成消息丢失。 当我们配置了 unclean.leader.election.enable false 的话当 leader 副本发生故障时就不会从 follower 副本中和 leader 同步程度达不到要求的副本中选择出 leader 这样降低了消息丢失的可能性。
RocketMQ 如何监听消息的 RocketMQ 常见的面试题
链接1 (20条消息) RocketMQ 常见面试问题_rocketmq面试题_柚子茶1990的博客-CSDN博客 主要是保证消息的不丢失。
哪个环节会有消息丢失的可能
我们考虑一个通用的MQ场景 其中124三个场景都是跨网络的而跨网络就肯定会有丢消息的可能。 然后关于3这个环节通常MQ存盘时都会先写入操作系统的缓存page cache中然后再由操作系统异步的将消息写入硬盘。这个中间有个时间差就可能会造成消息丢失。如果服务挂了缓存中还没有来得及写入硬盘的消息就会丢失。 这个是MQ场景都会面对的通用的丢消息问题。那我们看看用Rocket时要如何解决这个问题
RocketMQ消息零丢失方案
1、生产者使用事务消息机制保证消息零丢失
这个结论比较容易理解因为RocketMQ的事务消息机制就是为了保证零丢失来设计的并且经过阿里的验证肯定是非常靠谱的。 但是如果深入一点的话我们还是要理解下这个事务消息到底是不是靠谱。我们以最常见的电商订单场景为例来简单分析下事务消息机制如何保证消息不丢失。我们看下下面这个流程图 1、为什么要发送个half消息有什么用
这个half消息是在订单系统进行下单操作前发送并且对下游服务的消费者是不可见的。那这个消息的作用更多的体现在确认RocketMQ的服务是否正常。相当于嗅探下RocketMQ服务是否正常并且通知RocketMQ我马上就要发一个很重要的消息了你做好准备。
2.half消息如果写入失败了怎么办
如果没有half消息这个流程那我们通常是会在订单系统中先完成下单再发送消息给MQ。这时候写入消息到MQ如果失败就会非常尴尬了。而half消息如果写入失败我们就可以认为MQ的服务是有问题的这时就不能通知下游服务了。我们可以在下单时给订单一个状态标记然后等待MQ服务正常后再进行补偿操作等MQ服务正常后重新下单通知下游服务。
3.订单系统写数据库失败了怎么办 这个问题我们同样比较下没有使用事务消息机制时会怎么办如果没有使用事务消息我们只能判断下单失败抛出了异常那就不往MQ发消息了这样至少保证不会对下游服务进行错误的通知。但是这样的话如果过一段时间数据库恢复过来了这个消息就无法再次发送了。当然也可以设计另外的补偿机制例如将订单数据缓存起来再启动一个线程定时尝试往数据库写。而如果使用事务消息机制就可以有一种更优雅的方案。 如果下单时写数据库失败(可能是数据库崩了需要等一段时间才能恢复)。那我们可以另外找个地方把订单消息先缓存起来(Redis、文本或者其他方式)然后给RocketMQ返回一个UNKNOWN状态。这样RocketMQ就会过一段时间来回查事务状态。我们就可以在回查事务状态时再尝试把订单数据写入数据库如果数据库这时候已经恢复了那就能完整正常的下单再继续后面的业务。这样这个订单的消息就不会因为数据库临时崩了而丢失。
4.half消息写入成功后RocketMQ挂了怎么办
我们需要注意下在事务消息的处理机制中未知状态的事务状态回查是由RocketMQ的Broker主动发起的。也就是说如果出现了这种情况那RocketMQ就不会回调到事务消息中回查事务状态的服务。这时我们就可以将订单一直标记为新下单的状态。而等RocketMQ恢复后只要存储的消息没有丢失RocketMQ就会再次继续状态回查的流程。
5.下单成功后如何优雅的等待支付成功
在订单场景下通常会要求下单完成后客户在一定时间内例如10分钟内完成订单支付支付完成后才会通知下游服务进行进一步的营销补偿。 如果不用事务消息那通常会怎么办 最简单的方式是启动一个定时任务每隔一段时间扫描订单表比对未支付的订单的下单时间将超过时间的订单回收。这种方式显然是有很大问题的需要定时扫描很庞大的一个订单信息这对系统是个不小的压力。 那更进一步的方案是什么呢是不是就可以使用RocketMQ提供的延迟消息机制。往MQ发一个延迟1分钟的消息消费到这个消息后去检查订单的支付状态如果订单已经支付就往下游发送下单的通知。而如果没有支付就再发一个延迟1分钟的消息。最终在第十个消息时把订单回收。这个方案就不用对全部的订单表进行扫描而只需要每次处理一个单独的订单消息。 那如果使用上了事务消息呢我们就可以用事务消息的状态回查机制来替代定时的任务。在下单时给Broker返回一个UNKNOWN的未知状态。而在状态回查的方法中去查询订单的支付状态。这样整个业务逻辑就会简单很多。我们只需要配置RocketMQ中的事务消息回查次数(默认15次)和事务回查间隔时间(messageDelayLevel)就可以更优雅的完成这个支付状态检查的需求。
6、事务消息机制的作用
整体来说在订单这个场景下消息不丢失的问题实际上就还是转化成了下单这个业务与下游服务的业务的分布式事务一致性问题。而事务一致性问题一直以来都是一个非常复杂的问题。而RocketMQ的事务消息机制实际上只保证了整个事务消息的一半他保证的是订单系统下单和发消息这两个事件的事务一致性而对下游服务的事务并没有保证。但是即便如此也是分布式事务的一个很好的降级方案。目前来看也是业内最好的降级方案。
2、RocketMQ配置同步刷盘Dledger主从架构保证MQ自身不会丢消息
1、同步刷盘 这个从我们之前的分析就很好理解了。我们可以简单的把RocketMQ的刷盘方式 flushDiskType配置成同步刷盘就可以保证消息在刷盘过程中不会丢失了。
2、Dledger的文件同步 在使用Dledger技术搭建的RocketMQ集群中Dledger会通过两阶段提交的方式保证文件在主从之间成功同步。 简单来说数据同步会通过两个阶段一个是uncommitted阶段一个是commited阶段。 Leader Broker上的Dledger收到一条数据后会标记为uncommitted状态然后他通过自己的DledgerServer组件把这个uncommitted数据发给Follower Broker的DledgerServer组件。 接着Follower Broker的DledgerServer收到uncommitted消息之后必须返回一个ack给Leader Broker的Dledger。然后如果Leader Broker收到超过半数的Follower Broker返回的ack之后就会把消息标记为committed状态。 再接下来 Leader Broker上的DledgerServer就会发送committed消息给Follower Broker上的DledgerServer让他们把消息也标记为committed状态。这样就基于Raft协议完成了两阶段的数据同步。
3、消费者端不要使用异步消费机制
正常情况下消费者端都是需要先处理本地事务然后再给MQ一个ACK响应这时MQ就会修改Offset将消息标记为已消费从而不再往其他消费者推送消息。所以在Broker的这种重新推送机制下消息是不会在传输过程中丢失的。但是也会有下面这种情况会造成服务端消息丢失
DefaultMQPushConsumer consumer new DefaultMQPushConsumer(please_rename_unique_group_name_4);
consumer.registerMessageListener(new MessageListenerConcurrently() {Overridepublic ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs,ConsumeConcurrentlyContext context) {new Thread(){public void run(){//处理业务逻辑System.out.printf(%s Receive New Messages: %s %n, Thread.currentThread().getName(), msgs);}};return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
});
这种异步消费的方式就有可能造成消息状态返回后消费者本地业务逻辑处理失败造成消息丢失的可能。
4、RocketMQ特有的问题NameServer挂了如何保证消息不丢失
NameServer在RocketMQ中是扮演的一个路由中心的角色提供到Broker的路由功能。但是其实路由中心这样的功能在所有的MQ中都是需要的。kafka是用zookeeper和一个作为Controller的Broker一起来提供路由服务整个功能是相当复杂纠结的。而RabbitMQ是由每一个Broker来提供路由服务。而只有RocketMQ把这个路由中心单独抽取了出来并独立部署。 这个NameServer之前都了解过集群中任意多的节点挂掉都不会影响他提供的路由功能。那如果集群中所有的NameServer节点都挂了呢 有很多人就会认为在生产者和消费者中都会有全部路由信息的缓存副本那整个服务可以正常工作一段时间。其实这个问题大家可以做一下实验当NameServer全部挂了后生产者和消费者是立即就无法工作了的。至于为什么可以回顾一下我们之前的源码课程去源码中找找答案。 那再回到我们的消息不丢失的问题在这种情况下RocketMQ相当于整个服务都不可用了那他本身肯定无法给我们保证消息不丢失了。我们只能自己设计一个降级方案来处理这个问题了。例如在订单系统中如果多次尝试发送RocketMQ不成功那就只能另外找给地方(Redis、文件或者内存等)把订单消息缓存下来然后起一个线程定时的扫描这些失败的订单消息尝试往RocketMQ发送。这样等RocketMQ的服务恢复过来后就能第一时间把这些消息重新发送出去。整个这套降级的机制在大型互联网项目中都是必须要有的。
完整分析过后整个RocketMQ消息零丢失的方案其实挺简单
生产者使用事务消息机制。Broker配置同步刷盘Dledger主从架构消费者不要使用异步消费。整个MQ挂了之后准备降级方案
使用RocketMQ如何快速处理积压消息
1、如何确定RocketMQ有大量的消息积压
在正常情况下使用MQ都会要尽量保证他的消息生产速度和消费速度整体上是平衡的但是如果部分消费者系统出现故障就会造成大量的消息积累。这类问题通常在实际工作中会出现得比较隐蔽。例如某一天一个数据库突然挂了大家大概率就会集中处理数据库的问题。等好不容易把数据库恢复过来了这时基于这个数据库服务的消费者程序就会积累大量的消息。或者网络波动等情况也会导致消息大量的积累。这在一些大型的互联网项目中消息积压的速度是相当恐怖的。所以消息积压是个需要时时关注的问题。 对于消息积压如果是RocketMQ或者kafka还好他们的消息积压不会对性能造成很大的影响。而如果是RabbitMQ的话那就惨了大量的消息积压可以瞬间造成性能直线下滑。 对于RocketMQ来说有个最简单的方式来确定消息是否有积压。那就是使用web控制台就能直接看到消息的积压情况。 在Web控制台的主题页面可以通过 Consumer管理 按钮实时看到消息的积压情况。 另外也可以通过mqadmin指令在后台检查各个Topic的消息延迟情况。 还有RocketMQ也会在他的 ${storePathRootDir}/config 目录下落地一系列的json文件也可以用来跟踪消息积压情况。
2、如何处理大量积压的消息
其实我们回顾下RocketMQ的负载均衡的内容就不难想到解决方案。 如果Topic下的MessageQueue配置得是足够多的那每个Consumer实际上会分配多个MessageQueue来进行消费。这个时候就可以简单的通过增加Consumer的服务节点数量来加快消息的消费等积压消息消费完了再恢复成正常情况。最极限的情况是把Consumer的节点个数设置成跟MessageQueue的个数相同。但是如果此时再继续增加Consumer的服务节点就没有用了。 而如果Topic下的MessageQueue配置得不够多的话那就不能用上面这种增加Consumer节点个数的方法了。这时怎么办呢 这时如果要快速处理积压的消息可以创建一个新的Topic配置足够多的MessageQueue。然后把所有消费者节点的目标Topic转向新的Topic并紧急上线一组新的消费者只负责消费旧Topic中的消息并转储到新的Topic中这个速度是可以很快的。然后在新的Topic上就可以通过增加消费者个数来提高消费速度了。之后再根据情况恢复成正常情况。 在官网中还分析了一个特殊的情况。就是如果RocketMQ原本是采用的普通方式搭建主从架构而现在想要中途改为使用Dledger高可用集群这时候如果不想历史消息丢失就需要先将消息进行对齐也就是要消费者把所有的消息都消费完再来切换主从架构。因为Dledger集群会接管RocketMQ原有的CommitLog日志所以切换主从架构时如果有消息没有消费完这些消息是存在旧的CommitLog中的就无法再进行消费了。这个场景下也是需要尽快的处理掉积压的消息。
RocketMQ的消息轨迹
RocketMQ默认提供了消息轨迹的功能 另外也支持客户端自定义轨迹数据存储的Topic。 在客户端的两个核心对象 DefaultMQProducer和DefaultMQPushConsumer他们的构造函数中都有两个可选的参数来打开消息轨迹存储 enableMsgTrace是否打开消息轨迹。默认是false。 customizedTraceTopic配置将消息轨迹数据存储到用户指定的Topic 。
RocketMQ 如何处理流量削峰
连接地址 [(23条消息) RocketMQ实现流量削峰_mq削峰_ClareTung的博客-CSDN博客](https://blog.csdn.net/qq_36135928/article/details/121146631?ops_request_miscrequest_idbiz_id102utm_termRocketMQ 如何处理流量肖峰utm_mediumdistribute.pc_search_result.none-task-blog-2allsobaiduweb~default-0-121146631.142v80insert_down1,201v4add_ask,239v2insert_chatgptspm1018.2226.3001.4187)
实现的思路 总体思路
http请求到gateway网关gateway通过RouteLocator配置路由信息转发到mq服务mq服务接收到数据后将数据发送到对应的topicmq消费者监听消息拉取消息进行业务处理处理完后将msgId作为key结果作为value存入redis 中mq服务发送消息的线程发送完消息后挂起每隔一秒根据msgId去redis查询消费结果
使用RocketMQ如何保证消息顺序
为什么要保证消息的有序例如如果我们有个大数据系统需要对业务系统的日志进行收集分析这时候为了减少对业务系统的影响通常都会通过MQ来做消息中转。而这时候对消息的顺序就有一定的要求了。
MQ的顺序问题分为全局有序和局部有序 全局有序整个MQ系统的所有消息严格按照队列先入先出顺序进行消费。 局部有序只保证一部分关键消息的消费顺序。
