后缀int网站,网站建设-易速通科技,宽带动态ip如何做网站访问,wordpress博消息中间件如何选型 RabbitMQ erlang开发#xff0c;对消息堆积的支持并不好#xff0c;当大量消息积压的时候#xff0c;会导致 RabbitMQ 的性能急剧下降。每秒钟可以处理几万到十几万条消息。 RocketMQ java开发#xff0c;面向互联网集群化功能丰富#xff0c;对在线业… 消息中间件如何选型 RabbitMQ erlang开发对消息堆积的支持并不好当大量消息积压的时候会导致 RabbitMQ 的性能急剧下降。每秒钟可以处理几万到十几万条消息。 RocketMQ java开发面向互联网集群化功能丰富对在线业务的响应时延做了很多的优化大多数情况下可以做到毫秒级的响应每秒钟大概能处理几十万条消息。 Kafka Scala开发面向日志功能丰富性能最高。当你的业务场景中每秒钟消息数量没有那么多的时候Kafka 的时延反而会比较高。所以Kafka 不太适合在线业务场景。 ActiveMQ java开发简单稳定性能不如前面三个。小型系统用也ok但是不推荐。推荐用互联网主流的。 消息中间件的作用 因为项目比较大做了分布式系统所有远程服务调用请求都是同步执行经常出问题所以引入了mq | 作用 | 描述 | | ---- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 解耦 | 系统耦合度降低没有强依赖关系 | | 异步 | 不需要同步执行的远程调用可以有效提高响应时间 | | 削峰 | 请求达到峰值后后端service还可以保持固定消费速率消费不会被压垮 | | 数据分发 | 通过消息队列可以让数据在多个系统更加之间进行流通。数据的产生方不需要关心谁来使用数据只需要将数据发送到消息队列数据使用方直接在消息队列中直接获取数据即可。原来是需要A系统分别调用B C D 系统。现在如果新增E那么需要修改代码,需要在A系统调用E系统。通过消息队列可以让数据在多个系统更加之间进行流通。数据的产生方不需要关心谁来使用数据只需要将数据发送到消息队列数据使用方直接在消息队列中直接获取数据即可 | RocketMQ角色 | 角色 | 作用 | | ---------- | -------------------------------------------------- | | Nameserver | 路由发现,无状态,动态列表。这也是和zookeeper的重要区别之一。zookeeper是有状态的。 | | Producer | 消息生产者负责发消息到Broker。 | | Broker | 就是MQ本身负责收发消息、消息存储等。 | | Consumer | 消息消费者负责从Broker上拉取消息进行消费消费完进行ack。 | 主要是发布订阅模式主要组件消息发送者、消息服务器(消息存储)、消息消费、。 NameServer NameServer 是整个 RocketMQ 的“大脑”它是 RocketMQ 的服务注册中心,所以 RocketMQ需要先启动 NameServer 再启动 Rocket 中的 Broker。 Broker 在启动时向所有 NameServer 注册(主要是服务器地址等)生产者在发送消息之前先从 NameServer 获取 Broker 服务器地址列表(消费者一 样)然后根据负载均衡算法从列表中选择一台服务器进行消息发送。 NameServer与每台 Broker 服务保持长连接 NameServer每间隔 30S检查 Broker 是否存活如果检测到 Broker 宕机则从路由注册表中将其移除。这样就可以实现RocketMQ的高可用。 Broker和NameServer的关系 Broker会向所有的NameServer上注册自己的信息而不是某一个是每一个全部 RocketMQ中的Topic和queue区别 queue就是来源于数据结构的FIFO队列。而Topic是个抽象的概念每个Topic底层对应N个queue而数据也真实存在queue上的。 消息分类Topic和Tag 消息消费是消费者可以对同一主题下的消息按照规则只消费自己感兴趣的消息可以支持在服务端与消费端的消息过滤机制。 消息标签二级消息类型用来进一步区分某个 Topic 下的消息分类。 Topic 与 Tag 都是业务上用来归类的标识区分在于 Topic 是一级分类而 Tag 可以理解为是二级分类。 以天猫交易平台为例订单消息和支付消息属于不同业务类型的消息分别创建 TopicOrder 和 TopicPay其中订单消息根据商品品类以不同的 Tag 再进行细分如电器类、男装类、女装类、化妆品类最后他们都被各个不同的系统所接收。 通过合理的使用 Topic 和 Tag可以让业务结构清晰更可以提高效率。 您可能会有这样的疑问到底什么时候该用 Topic什么时候该用 Tag 1)消息类型是否一致如普通消息、事务消息、定时(延时)消息、顺序消息不同的消息类型使用不同的 Topic无法通过 Tag 进行区分。 2)业务是否相关联没有直接关联的消息如淘宝交易消息京东物流消息使用不同的 Topic 进行区分而同样是天猫交易消息电器类订单、女 装类订单、化妆品类订单的消息可以用 Tag 进行区分。 3)消息优先级是否一致如同样是物流消息盒马必须小时内送达天猫超市 24 小时内送达淘宝物流则相对会慢一些不同优先级的消息用不同的 Topic 进行区分。 4)消息量级是否相当有些业务消息虽然量小但是实时性要求高如果跟某些万亿量级的消息使用同一个 Topic则有可能会因为过长的等待时间 而“饿死”此时需要将不同量级的消息进行拆分使用不同的 Topic。 消息存储 一般 MQ 核心就是消息的存储对存储一般来说两个维度:消息堆积能力和消息存储性能。RocketMQ 追求消息存储的高性能引入内存映射机制所有的主题消息顺序存储在同一个文件中。同时为了防止无限堆积引入消息文件过期机制和文件存储空间报警机制。 消息高可用 Rocket 关机、断电等情况下Rokcet 可以确保不丢失消息(同步刷盘机制不丢失异步刷盘会丢失少量)。 另外如果 Rocket 服务器因为 CPU、内存、主板、磁盘等关键设备损坏导致无法开机这个属于单点故障该节点上的消息全部丢失如果开启了 异步复制机制Rocket 可以确保只丢失很少量消息。 如果引入双写机制这样基本上可以满足消息可靠性要求极高的场景(毕竟两台主服务器同时故障的可能性还是非常小) 消息消费低延迟 RocketMQ在消息不发生消息堆积时以长轮询模式实现准实时的消息推送模式。 确保消息必须被消费一次 消息确认机制(ACK)来确保消息至少被消费一次一般 ACK 机制只能做到消息只被消费一次有重复消费的可能。 消息回溯 已经消费完的消息可以根据业务要求重新消费消息。 消息堆积 消息中间件的主要功能是异步解耦还有个重要功能是挡住前端的数据洪峰保证后端系统的稳定性这就要求消息中间件具有一定的 消息堆积能力RocketMQ 采用磁盘文件存储所以堆积能力比较强同时提供文件过期删除机制。 定时消息 定时消息定时消息是指消息发送到 Rocket Broker 上之后不被消费者理解消费要到等待一定的时间才能进行消费apache 的版本目 前只支持等待指定的时间才能被消费不支持任意精度的定时消息消费。(一个说法是任意精度的定时消息会带来性能损耗但是阿里云版本的 RocketMQ 却提供这样的功能充值收费优先策略) 消息重试机制 消息重试是指在消息消费时如果发送异常那么消息中间件需要支持消息重新投递RocketMQ 支持消息重试机制。 Message Key Key 一般用于消息在业务层面的唯一标识。对发送的消息设置好 Key以后可以根据这个 Key 来查找消息。比如消息异常消息丢失进行查找会很 方便。 RocketMQ 会创建专门的索引文件用来存储 Key 与消息的映射由于是 Hash 索引应尽量使 Key 唯一避免潜在的哈希冲突。 Tag 和 Key 的主要差别是使用场景不同Tag 用在 Consumer 代码中用于服务端消息过滤Key 主要用于通过命令进行查找消息 RocketMQ 并不能保证 message id 唯一在这种情况下生产者在 push 消息的时候可以给每条消息设定唯一的 key, 消费者可以通过 message key 保证对消息幂等处理。 消息被消费后会立即删除吗 不会RocketMQ每条消息都会持久化到CommitLog中每个Consumer连接到Broker后会维持消费进度信息当有消息消费后只是当前Consumer的消费进度(CommitLog的offset)更新了。 消息会堆积吗什么时候清理过期消息 先简单给大家说一下其实默认broker会启动后台线程这个后台线程会自动去检查CommitLog、ConsumeQueue文件因为这些文件都是多个的比如CommitLog会有多个ConsumeQueue也会有多个。 然后如果是那种比较旧的超过72小时的文件就会被删除掉也就是说默认来说broker只会给你把数据保留3天而已当然你也可以自己通过fileReservedTime来配置这个时间要保留几天的时间。 这个定时检查过期数据文件的线程代码在DefaultMessageStore这个类里他的start0方法中会调用一个addScheduleTask0方法里面会每隔10s定时调度执行一个后台检查任务我们看下面的源码片段 上面就可以看到了其实他是每隔10s就会执行一个调度任务 这个调度任务里就会执行DefaultMessageStore,this.dleanFilesPeriodically0方法其实就是会去周期性的清理掉磁盘上的数据文件也就是超过72小时的CommitLog、ConsumeQueue文件接着我们具体看看这里的清理逻辑他其实里面包含了清理CommitLo和ConsumeQueue的清理逻辑如下面源码片段。 在清理文件的时候他会具体判断一下如果当前时间是预先设置的凌晨4点就会触发删除文件的逻辑这个时间是默认的;或者是如果磁盘空间不足了就是超过了85%的使用率了立马会触发删除文件逻辑. 上面两个条件第一个是说如果磁盘没有满那么每天就默认一次会删除磁盘文件默认就是凌晨4点执行那个时候必然是业务低峰期因为凌晨4点大部分人都睡觉了无论什么业务都不会有太高业务量的。 第二个是说如果磁盘使用率超过85%了那么此时可以允许继续写入数据但是此时会立马触发删除文件的逻辑如果磁盘使用率超过90%了那么此时不允许在磁盘里写入新数悟立马删除文件。这是因为一旦磁盘满了那么你写入磁盘会失败此时你MO就彻底故障了 所以一且磁盘满了也会立马删除文件的 在删除文件的时候无非就是对文件进行遍历如果一个文件超过72小时都没修改过了此时就可以删除了哪怕有的消息你可能还没消费过但是此时也不会再让你消费了就直接删除掉。这就是RocketMQ的一整套文件删除的逻辑和和机制 4.6版本默认72小时后会删除不再使用的CommitLog文件 检查这个文件最后访问时间判断是否大于过期时间指定时间删除默认凌晨4点 RocketMQ消费模式有几种 消费模型由Consumer决定消费维度为Topic。 集群消费 1.一条消息只会被同Group中的一个Consumer消费 2.多个Group同时消费一个Topic时每个Group都会有一个Consumer消费到数据 广播消费 消息将对一 个Consumer Group 下的各个 Consumer 实例都消费一遍。即即使这些 Consumer 属于同一个Consumer Group 消息也会被 Consumer Group 中的每个 Consumer 都消费一次。 消费消息是push还是pull-重点看 RocketMQ没有真正意义的push都是pull虽然有push类但实际底层实现采用的是长轮询机制即拉取方式 broker端属性 longPollingEnable 标记是否开启长轮询。默认开启 // {link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()} // 看到没这是一只披着羊皮的狼名字叫PushConsumerImpl实际干的确是pull的活。 // 拉取消息结果放到pullCallback里 this.pullAPIWrapper.pullKernelImpl(pullCallback); 实时性和消息堆积如何取舍 PULL方式 由消费者客户端主动向消息中间件(MQ消息服务器代理)拉取消息采用Pull方式如何设置Pull消息的拉取频率需要重点去考虑举个例子来说可能1分钟内连续来了1000条消息然后2小时内没有新消息产生(概括起来说就是“消息延迟与忙等待”) 如果每次Pull的时间间隔比较久会增加消息的延迟即消息到达消费者的时间加长MQ中消息的堆积量变大若每次Pull的时间间隔较短但是在一段时间内MQ中并没有任何消息可以消费那么会产生很多无效的Pull请求的RPC开销影响MQ整体的网络性能 PUSH方式 由消息中间件(MQ消息服务器代理)主动地将消息推送给消费者采用Push方式可以尽可能实时地将消息发送给消费者进行消费。但是在消费者的处理消息的能力较弱的时候(比如消费者端的业务系统处理一条消息的流程比较复杂其中的调用链路比较多导致消费时间比较久。概括起来地说就是“慢消费问题”)而MQ不断地向消费者Push消息消费者端的缓冲区可能会溢出或者产生消息堆积导致异常 代码1:如果开启了长轮询模式则每次只挂起 5s然后就去尝试拉取。 代码2:如果不开启长轮询模式则只挂起一次挂起时间为 shortPollingTimeMills然后去尝试查找消息。 代码3:遍历 pullRequestTable如果拉取任务的待拉取偏移量小于当前队列的最大偏移量时执行拉取否则如果没有超过最大等待时间则等待否则返回未拉取到消息返回给消息拉取客户端 参考链接https://blog.csdn.net/prestigeding/article/details/79357818 为什么主动拉取不使用事件监听方式 事件驱动方式是建立好长连接由事件(发送数据)的方式来实时推送。 如果broker主动推送消息的话有可能push速度快消费速度慢的情况那么就会造成消息在consumer端堆积过多同时又不能被其他consumer消费的情况。 而pull的方式可以根据当前自身情况来pull不会造成过多的压力而造成瓶颈。 所以采取了pull的方式。消费者端的缓冲区可能会溢出导致异常 broker如何处理拉取请求的 Consumer首次请求Broker Broker中是否有符合条件的消息有 - 响应Consumer等待下次Consumer的请求没有 挂起consumer的请求即不断开连接也不返回数据使用consumer的offset DefaultMessageStore#ReputMessageService#run方法 每隔1ms检查commitLog中是否有新消息有的话写入到pullRequestTable当有新消息的时候返回请求PullRequestHoldService 来Hold连接每隔5s执行一次检查pullRequestTable有没有消息有的话立即推送 RocketMQ如何做负载均衡 通过Topic在多Broker中分布式存储实现。 producer端 发送端指定message queue发送消息到相应的broker来达到写入时的负载均衡 提升写入吞吐量当多个producer同时向一个broker写入数据的时候性能会下降消息分布在多broker中为负载消费做准备 默认策略是平均选择 producer维护一个自增index每次取节点会自增index向所有broker个数取余自带容错策略 其他实现 SelectMessageQueueByHash hash的是传入的argsSelectMessageQueueByRandomSelectMessageQueueByMachineRoom 没有实现 也可以自定义实现MessageQueueSelector接口中的select方法 MessageQueue select(final ListMessageQueue mqs, final Message msg, final Object arg); consumer端 采用的是平均分配算法来进行负载均衡。 其他负载均衡算法 平均分配策略(默认)(AllocateMessageQueueAveragely) 环形分配策略(AllocateMessageQueueAveragelyByCircle) 手动配置分配策略(AllocateMessageQueueByConfig) 机房分配策略(AllocateMessageQueueByMachineRoom) 一致性哈希分配策略(AllocateMessageQueueConsistentHash) 靠近机房策略(AllocateMachineRoomNearby) consumer和queue不对等的时候会发生什么 Consumer和queue会优先平均分配如果Consumer少于queue的个数则会存在部分Consumer消费多个queue的情况如果Consumer等于queue的个数那就是一个Consumer消费一个queue如果Consumer个数大于queue的个数那么会有部分Consumer空余出来白白的浪费了。
