zoukankan      html  css  js  c++  java
  • RocketMQ之消息中间件需要解决的问题

    消息中间件需要解决哪些问题

    1.Publish/Subscribe(发布订阅)

    发布订阅是消息中间件最基本的功能

    2.Message Priority(消息优先级)

    在消息队列中,每条消息都有不同的优先级,优先级高的先投递。

    由于rocketmq的所有消息都是持久化的,按照优先级排序开销会非常大,所以不支持持久化。但是可以配置一个优先级高的队列和一个普通的队列,将不同的消息发送到不同的队列。

    优先级问题可以分为两类:

    1. 只要达到优先级目的即可,不需要严格划分优先级。通常将优先级划分为高、中、低等几个等级。每个优先级用不同的topic表示。发送消息通过不同的topic来表示优先级。缺点是对业务的优先级做了妥协。
    2. 严格的优先级。如果让MQ解决此问题,会对MQ的性能造成很多影响。这里要确保一点,业务上是否确实需 要这种严格的优先级,如果将优先级压缩成几个,对业务的影响有多大?

    3.Message Order(消息有序)

    消息有序指的是一类消息消费时,能按照发送的顺序来消费。例如:一个订单产生了 3 条消息,分别是订单创 建,订单付款,订单完成。消费时,要按照这个顺序消费才能有意义。但是同时订单之间是可以并行消费的。

    rocketmq严格保证消息有序性。

    4.Message Filter(消息过滤)

    • Broker端消息过滤

        在broker中按照consumer的要求过滤,优点是减少了对应consumer的无用消息的传输。但增加了broker的负担,使得实现变复杂。

    • Consumer端消息过滤

        这种过滤方式可由应用完全自定义实现,但是缺点是很多无用的消息要传输到 consumer端。

    5.Message Persistence(消息持久化)

    消息中间件通常采用几种方式持久化:

    1. 持久化到数据库
    2. 持久化到KV存储
    3. 文件记录形式的持久化,例如kafka、rocketmq
    4. 对内存数据做持久化镜像

    前三种持久化方式都具有将内存队列 buffer 进行扩展的能力,后一种则当broker挂掉重启后仍然能将之前内存的数据恢复出来。

    rocketmq参考了kafka的持久化方式,充分利用Linux文件系统内存cache来提高性能。

    6.Message Reliablity(消息可靠性)

    影响消息可靠性的几种情况:

    1. broker正常关闭
    2. broker异常crash
    3. OS crash
    4. 机器掉电,但能立即恢复供电情况
    5. 机器不能开机(硬件损坏)
    6. 磁盘设备损坏

    前面1-4四种情况都属于硬件资源可立即恢复的情况。rocketmq在这四种情况下能保证消息不丢,或丢失少量数据(依赖刷盘方式是同步还是异步)。

    5-6两种情况属于单点故障,且不能恢复,一旦发生,在此单点上的消息全部丢失。rocketmq在这两种情况下,通过异步复制,可保证99%的消息不丢。通过同步双写技术可以完全避免单点,但会影响性能,适合对消息可靠性要求极高的场景,如与钱相关的应用。

    7.Low Latency Messaging(低延迟)

    在消息不堆积情况下,消息到达 broker 后,能立刻到达 consumer。

    rocketmq使用长轮询 pulll 方式,可保证消息非常实时,消息实时性不低于 push。

    8.At least Once(至少一次) 

    指每个消息必须投递一次

    rocketmq的consumer先pull消息到本地,消息完成后,才向服务器返回ask。如果没有消费,一定不会ask消息。

    9.Exactly Only Once(只有一次)

    • 发送消息阶段不允许重复发送
    • 消费消息阶段不允许重复消费

    只有两个条件都满足,才能认为消息是去除的。而要实现以上两点,在分布式环境下,无疑会产生巨大开销。

    rocketmq并不保证此特性,而是要求在业务上去重,即消费消息做到幂等性。

    10.Broker 的 Buffer

    broker中的buffer通常指broker中一个队列中的内存buffer大小。如果buffer满了怎么办?

    CORBA Notification 规范中处理方式:

    • RejectNewEvents:拒绝新来的消息,向producer返回错误码
    • 按照特定策略丢弃已有消息

    rockermq的队列都是持久化磁盘,buffer大小是磁盘容量,且数据定期清理。

    11.回溯消费

    回溯消息指consumer成功消费的消息由于业务上的需求,需要重新消费。要支持此功能,broker在向consumer投递消息后,消息仍要保留。。并且重新消费一般是按照时间维度,例如由于 Consumer 系统故障, 恢复后需要重新消费 1 小时前的数据,那么 Broker 要提供一种机制,可以按照时间维度来回退消费进度。

    rocketmq支持按照时间回溯消息,时间维度精确到毫秒,可以向前回溯,也可以向后回溯。

    12.消息堆积

    消息中间件的主要功能是异步解耦,挡住前端的数据洪峰,保证后端系统的稳定性。这要求消息中间件具有一定的消息堆积能力。

    消息堆积分两种:

    • 一种是消息堆积在内存buffer中,一旦超过内存buffer,可以根据丢弃策略来丢弃消息。这种消息堆积能力主要在于内存buffer的大小。且消息堆积后,性能下降不会太大。因为内存中数据多少对于对外提供的访问能力影响有限。
    • 第二种是消息堆积在持久化存储系统中,如:数据库,KV存储,文件记录形式。当消息不能在内存cache中命中时,要不可避免的访问磁盘,从而产生大量读IO,读IO的吞吐量直接决定了消息堆积后的访问能力

    13.分布式事务

    rocketmq采用第一阶段发送Prepared消息时,拿到消息的offset,第二阶段通过offset访问消息,并修改状态。offset就是数据的地址。

    rocketmq实现事务方式,没有通过KV存储做,而是通过offset方式,存在一个显著缺陷,即通过offset更改数据,会令系统的脏页过多。

    14.定时消息

    定时消息指消息放到broker后,不能立即被consumer消费,需到特定时间点或等待特定时间后才能被消费。

    如果支持任意时间精度,在broker层,就必须做消息排序,涉及到持久化,排序就会产生大量性能开销。

    rocketmq支持定时消息,但不支持任意精度,只支持特定level,如:5s,10s,1m等

    15.消息重试

    consumer消费消息失败后,要提供一种重试机制,令消息再消费一次。

    consumer消费消息失败有几种情况:

    1. 由于消息本身原因,如反序列化失败,消息数据本身没法处理(话费充值,当前手机号被注销,不能充值)等。这种错误通常需要跳过这条消息,再消费其他消息,这条失败的消息即使立刻重试消费,基本不可能成功,所以最好提供一种定时重试机制。
    2. 由于依赖的下游应用用服务不可用,如数据库连接不可用,网络原因等。这种错误即使跳过当前失败的消息,消费其他消息同样会报错。

    这种情况建议应用sleep 30s,再消费下条消息,从而减轻broker重试消息的压力。

  • 相关阅读:
    (原)ubuntu16在torch中使用caffe训练好的模型
    (原)Ubuntu16中卸载并重新安装google的Protocol Buffers
    (原)lua提示cannot load incompatible bytecode
    (原)ubuntu上安装nvidia及torch的nccl
    Ubuntu修改grub菜单引导选项和等待时间
    Servelet 简介
    JAVA JUC 线程池
    JAVA JUC synchronized 锁的理解
    JAVA JUC 读写锁
    JAVA JUC 线程按顺序执行
  • 原文地址:https://www.cnblogs.com/yushangzuiyue/p/9684000.html
Copyright © 2011-2022 走看看