网站建设运营合同模板,招标网查询,顺德网站建设找顺的,h5海报免费制作软件文章目录 1、什么是消息队列#xff1f;2、消息队列有哪些使用场景#xff1f;#xff08;为什么使用消息队列#xff09;2.1 应用解耦2.2 流量削峰2.3 异步处理2.4 消息通讯2.5 远程调用 3、消息队列如何解决消息丢失问题#xff1f;3.1 生产者保证消息不丢失3.2 存储段不… 文章目录 1、什么是消息队列2、消息队列有哪些使用场景为什么使用消息队列2.1 应用解耦2.2 流量削峰2.3 异步处理2.4 消息通讯2.5 远程调用 3、消息队列如何解决消息丢失问题3.1 生产者保证消息不丢失3.2 存储段不丢消息3.3 消费者不丢消息 4、消息队列如何保证消息的顺序性4.1 单机4.2 集群4.2.1 生产者有序消费4.2.2 消费者有序消费 5、如何避免消息重复消费5.1 提高消费端的处理性能避免触发Balance5.2 使用ConsumerRebalanceListener再均衡监听器5.3 使用消息幂等性 6、如何解决幂等性问题6.1 概念6.2 问题6.3 如何解决 7、如何处理消息队列消息积压问题8、MQ技术选型9、如何保证数据一致性事务消息如何实现10、RabbitMQ的消息可靠传输如何保证11、RabbitMQ的消息如何实现路由 1、什么是消息队列 Message Queue 简称 MQ是一种应用间的通信方式由生产者Producer、代理Broker、消费者Consumer三者组成。 场景 异步实时性要求不严格的场景注册发送验证码、下单通知、发送优惠券等。不需要等待消息服务返回结果。应用解耦将相关但耦合度不高的系统联系起来。解决了各个系统可以采用不同的架构、语言来实现大大增加系统的灵活性。流量削峰应用在大流量入口且短时间内业务需求处理不完的服务中心为权衡高可用将大量并行业务发送到MQ起到大流量缓冲的作用。 类型ActiveMQ、RabbitMQ、Kafka、RocketMQ 中小型公司、低吞吐量一般用ActiveMQ、RabbitMQ。 大数据高吞吐量的大型公司一般选用用Kafka和RocketMQ。 2、消息队列有哪些使用场景为什么使用消息队列 2.1 应用解耦 将相关但耦合度不高的系统联系起来。解决了各个系统可以采用不同的架构、语言来实现大大增加系统的灵活性。扩充下游系统时不需要做大的调整。 2.2 流量削峰 场景特点应用在大流量入口且短时间内业务需求处理不完的服务中心 比如做秒杀需要避免流量暴涨打垮应用的风险。假设系统每秒最多处理2k个请求但实际每秒有5k个请求此时引入消息队列每秒从队列中拉取2k个请求处理就可以了。 消息积压 秒杀活动不可能每时每刻大请求量高峰期过去积压请求可以慢慢处理。队列长度超过最大数量可以直接抛弃用户请求走兜底业务 2.3 异步处理 将按顺序执行的业务同步执行大大减少响应速度。 2.4 消息通讯 内置了高效的通讯机制可以实现点对点消息队列、聊天室等。 2.5 远程调用 3、消息队列如何解决消息丢失问题 生产者产生消息-》Broker代理存储消息-》消费者消费消息 3.1 生产者保证消息不丢失 如果使用RocketMQ消息中间件生产者提供了三种发送方式同步、异步、单向 采用同步发送send消息返回成功状态则表示成功存储send消息异常或返回非成功状态可以重试可以使用事务消息RocketMQ事务消息机制就是为了保证零丢失设计的 3.2 存储段不丢消息 刷盘机制 生产者发送消息只有持久化到磁盘RocketMQ存储端才会返回一个成功ACK响应。但影响性能。 异步刷盘 只要消息写入PageCache缓存就返回一个成功ACK响应。提高了性能但一单及其断电就会丢失消息。 Broker一般集群部署有master主节点和slave从节点。 同步复制主节点和从节点都收到消息才返回成功ACK保证了消息不丢失但降低性能。 异步复制只要消息写入主节点就返回成功ACK速度快但会有丢失问题。 3.3 消费者不丢消息 消费者执行完业务逻辑在反馈给Broker消费成功。 4、消息队列如何保证消息的顺序性 4.1 单机 发送至同一个服务的MQ上发送至同一个服务的消费者且等到M1消费端ACK成功后M2再发送。 4.2 集群 生产者有序存储、消费者有序消费。 4.2.1 生产者有序消费 普通发送消息的模式下生产者采用轮询的方式均匀发送至不同队列中被不同的消费者消费。此时无法使用队列有序特性保证有序性。 解决方案投放消息支持自定义投放策略顺序消息必须使用同步发送的方式才能保证发送有序。 实现一个**MessageQueueSelector接口**使用Hash取模法来保证同一个订单在同一个队列中就行了 即通过订单ID%队列数量得到该ID的订单所投放的队列在队列列表中的索引然后该订单的所有消息都会被投放到这个队列中。 4.2.2 消费者有序消费 RockerMQ的MessageListener回调函数提供了两种消费模式有序消费模式MessageListenerOrderly和并发消费模式MessageListenerConcurrently。 在消费的时候还需要保证消费者注册MessageListenerOrderly类型的回调接口实现顺序消费如果消费者采用Concurrently并行消费则仍然不能保证消息消费顺序。 实际上每一个消费者的的消费端都是采用线程池实现多线程消费的模式即消费端是多线程消费。虽然MessageListenerOrderly被称为有序消费模式但是仍然是使用的线程池去消费消息。 MessageListenerConcurrently是拉取到新消息之后就提交到线程池去消费而MessageListenerOrderly则是通过加分布式锁和本地锁保证同时只有一条线程去消费一个队列上的数据。 messageQueue的本地synchronized锁 执行消费任务的开头获取本地锁对象objLock通过synchronized实现锁定。锁对象存储在MessageQueueLock.mqLockTable属性中结构为ConcurrentMapMessageQueue, Object一个messageQueue对应一个锁。这个锁保证同一时刻对于同一个队列只有一个线程去消费他。 ProcessQueue的本地consumeLock 执行真正的消费之前会获取ProcessQueue的本地consumeLock这个本地锁是一个ReentrantLock。 作用防止在消费过程中该消息队列因负载均衡而被分配给其他客户端导致两个客户端重复消费。 5、如何避免消息重复消费 生产端为保证消息可靠性可能想MQ重复发送消息直到拿到成功的ACK。 消费端流程拉取消息-》业务逻辑执行-》提交消费位移。 假设更新位移时消费者挂了这时另一个消费者就会拉取到重复的消息。 解决方案 5.1 提高消费端的处理性能避免触发Balance 多线程处理消息缩短单个消息消费时长。调整消息处理的超时时间。减少一次性从Broker上拉取数据的条数。 5.2 使用ConsumerRebalanceListener再均衡监听器 设定发生再均衡动作前后的一些准备工作和收尾工作。 5.3 使用消息幂等性 开启KafKa的幂等性功能将消息生成md5带唯一业务标记保存到Mysql或Redis中在处理消息之前先查Mysql或Redis进行判断是否已消费。 6、如何解决幂等性问题 6.1 概念 一个方法无论被多少次重复执行所期望的结果和第一次执行所期望的结果保持一致。 6.2 问题 用户重复提交、恶意攻击分布式系统中为避免数据丢失采用的超时重试机制 6.3 如何解决 本质保证接口的执行结果只影响一次后续再次调用不能对数据产生影响。 使用数据库唯一约束实现。使用Redis提供的setNX指令使用状态机实现即一条数据的完整运行状态的转换流程状态只会向前变更 7、如何处理消息队列消息积压问题 起因消费速度原小于生产速度 临时解决紧急扩容先保证消息都消费完。 修复consumer消费者问题确保恢复消费速度然后将现有consumer都停掉。新建一个topicpartition临时建立好原先10倍的queue数量。写一个临时分发数据的consumer程序部署上去消费积压的数据消费之后不做耗时处理直接轮询写入临时创建好的10倍数量的queue中。临时征用10倍数量的机器部署consumer每一批consumer消费一个临时queue数量。相当于临时将queue资源和consumer资源扩大10倍10倍速度消费。快速消费积压数据后恢复原先部署的程序。 8、MQ技术选型 指标KafkaRocketMQRabbitMQ单机吞吐量17.3w/s11.6w/s2.6w/s(消息做持久化)开发语言Scala/JavaJavaErlang维护ApacheAlibabaSpring订阅形式基于topic进行正则匹配基于topic、messageTag根据类型和属性正则匹配支持direct、topic、Headers、fanout四种模式持久化支持大量堆积支持大量堆积支持少量堆积顺序消息支持支持不支持集群方式天然Leader-Slave无状态集群每台服务器既是Master也是Slave常用“多对Master-Slave”开源版需要手动切换从节点变主节点支持简单集群“复制模式”对高级集群支持不好性能稳定性较差一般好 大数据领域实时计算、日志采集等用Kafka为业内标准RocketMQ阿里出品RabbitMQ开源稳定支持、活跃度高但不是Java开发 9、如何保证数据一致性事务消息如何实现 生产者发送消息到MQ服务器MQ收到消息后持久化到存储系统状态为代发送MQ服务器返回ACK确认到生产者此时还不触发推送事件生产者执行本地事务本地事务执行成功提交结果到MQ服务器执行失败发送rollback正常提交MQ更新消息状态为可送达rollback回滚则删除消息如果消息更新为可送达则MQ服务器push消息给消费者消费完成就回ACK。如果MQ服务器长时间没有收到生产者的commit或rollback会反查生产者根据查询结果执行最终状态 10、RabbitMQ的消息可靠传输如何保证 丢失的三种情况 生产者发送消息到MQ 服务的过程中丢失MQ服务收到消息后在持久化之前宕机导致数据丢失消费端收到消息还未处理 对于生产者生产者发送之后MQ提供了一个消息确认机制客户端根据消息的处理结果决定是否需要重新发送。 MQ 服务端开启消息持久化机制存入磁盘。 创建Queue设置持久化发送消息的时候将消息投递模式设置为持久化投递。 消费端将消息的自动确认机制修改为手动确认只会手动用消息确认方法才表示消息已被签收。 11、RabbitMQ的消息如何实现路由 本质一个基于AMQP协议实现的分布式消息中间件。 AMQP工作机制 生产者将消息发送到RabbitMQ Broker上的Exchange交换机上。Exchange交换机把收到的消息根据路由规则发给绑定的Queue队列。再把消息投递给订阅了这个队列的消费者完成消息的异步通讯。 Exchange 消息交换机定义了消息路由的规则规定消息与队列的关系。 Queue消息的载体每个消息可以根据路由规则到一个或多个队列中。 RoutingKey路由键发送消息时声明Exchange拿到路由键根据它和路由表里面的BindingKey进行匹配。 在RabbitMQ中有三种类型的Exchangedirect fanout和topic。 direct完整匹配是Routing key和Binding Key完全一致点对点发送。 fanout广播机制不急于路由键匹配将消息广播给绑定到当前交换机上的所有队列。 topic正则表达式匹配根据Routing key使用正则表达式进行匹配发送消息给符合规则的Queue。