MQ

mq使用场景

注册用户=>赠送优惠券
开票回调

kafka

group:对于某个topic,如果存在多个消费group,则每个group都会订阅到相同的消息,但是group内部只有固定某个consumer能消费
集群=>分区(类似分表-数据量太大)。多个broker就是多个partition,生产者发送消息到某个topic(逻辑概念并不真实存在)时,实际是按照默认规则随机发送到某个partition
如果存在3个partition,1个group中有3个消费者,则默认是一个消费者消费一个partition,不会交叉(性能低);如果2个消费者,则其中一个消费者消费两个partition,另一个消费一个partition;如果有4个消费者,则其中3个一对一,第四个消费者不消费。避免了锁竞争
当然,consumer代码中可以指定消费某个分区。
最好partition数量是consumer数量的整数倍。

kafka Rebalance

partition和consumer发生变化时
3种策略:随机(random),范围(range),strict(在尽量平均的基础上,尽量保持之前的配对不变-粘性)
由coordinator处理:
consumer group如何确定自己的coordinator是谁呢, 消费者向kafka集群中的任意一个broker发送一个
GroupCoordinatorRequest请求,服务端会返回一个负载最小的broker节点的id,并将该broker设置为coordinator

批量发送和批量获取

批量发送:现在内存中缓存,可以定时自动提交

offset

生产者随机发送到broker(partition),会返回一个自增的offerset(生产者不需要在broker上保存)
消费者每次消费,消费的offerset++,消费者会在broker上保存它,下次消费时读取下一个offset,如果从头读取offset就可以实现earliest数据的读取。
offset的存储:broker上有个专门的topic --- consumer_offsets,默认50个partition

副本机制

每个partition都有follow副本,自动从leader副本同步数据,leader副本挂掉后会重新选举(没有zookeeper那么严格和复杂)
副本数量不能超过broker的数量
leader和follow副本都维护在ISR集合中,谁挂掉就移除谁,如果follow延时过大也会被移除(有一个时间参数),移除后仍会从leader副本同步数据
leader宕机后,优先从ISR集合中选第一个当leader

kafka定位offset

partition分为很多LogSegment
先找到index文件[范围索引],从中找到log文件,最后从log文件中找到position

kafka既然有副本,为啥不做读写分离

数据一致性代价比加大

kafka清理策略

过期清理:按大小/时间清理
压缩策略

kafka磁盘存储

顺序写
大量页缓存
零拷贝

LEO和HW

数据丢失问题

解决方案:
引入了epoch的概念:(epoch, offset)

kafka日志清除和压缩

MQTT

微消息服务队列,面向移动端,支持超大量在线客户端接入(几万甚至几百万),单台设备的数据量很少(一般不会给客户端推送几条消息);
RocketMQ等,面向服务端,用户服务器和服务器之间的消息传递,一般公司内部服务器不会超过1万台,单台设备的数据量超大(服务器之间的消息传递量是巨大的)。

RabbitMQ

消息可靠性

  1. Producer => Exchange:事务消息(性能低),confirm模式(同步确认,异步确认,批量确认)
  2. Broker内部:Exchange/Queue/message的持久化,备份交换机和备份队列,集群
  3. Queue => Consumer:自动应答或手动应答

4层和7层负载

RabbitMQ的集群

普通集群

镜像集群:高可用

RocketMQ

要使用rocketmq,至少需要启动两个进程,nameserver、broker,前者是各种topic注册中心,后者是真正的broker
阿里云RocketMQ nameserver的端口默认是8080:factoryInfo.setFactoryProperty(factoryInfo.NAMESRV_ADDR, "http://onsaddr.cn-shanghai.mq-internal.aliyuncs.com:8080");
RocketMQ是基于长轮训来实现消息的pull
topic下通过tag做消息过滤

RocketMQ消息模式

同步、异步、OneWay(只发请求不等应答-效率最高,比如日志)

RocketMQ的Message Queue--类似kafka的partition

如果一个Topic 要发送 和接收的数据量非常大, 需要能支持增加并行处理的机器来提高处理速度,这时候一个Topic 可以根据 需求设置一个或多个Message Queue

image.png

RocketMQ消费端负载均衡

和kafka的partition一样,消费端也会针对Message Queue做负载均衡,使得每个消费者能够合理的消费多个分区 的消息。

RocketMQ默认16次衰减重试

RocketMQ的事务消息--可用来处理分布式事务

RocketMQ的消息存储

commitLog是顺序加锁写,随机读

当消费者进行消息消费时,会先读取consumerQueue, 逻辑消费队列ConsumeQueue保存了指
定Topic下的队列消息在CommitLog中的起始物理偏移量Offset,消息大小、和消息Tag的
HashCode值

RocketMQ发送顺序消息

只需要保证消息发送到同一个topic的同一个Queue,发送时指定MessageQueueSelector的key一样即可。
MessageQueueSelector的几个实现:SelectMessageQueueByHash(默认)、SelectMessageQueueByRandom、SelectMessageQueueByMachineRoom

public class SelectMessageQueueByHash implements MessageQueueSelector {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        int value = arg.hashCode();
        if (value < 0) {
            value = Math.abs(value);
        }

        value = value % mqs.size();
        return mqs.get(value);
    }
}

RocketMQ延时消息--18个延时级别

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
setDelayTimeLevel(3) => 延时10s

如何实现顺序消费

RocketMQ发送消息到同一个message queue即可,但是如果消息是多线程发送的,即使发送到同一个message queue,也有可能由于网络问题导致乱序。所以应该在消费方处理,比如状态机(这个没有想到具体例子),或者消费表做记录,攒齐7个再按标号执行。

有人压测过

2k的消息体,RocketMQ性能优于kafka

RocketMQ缺点

延迟队列写死18级不可定制


批量拉取多线程消费如何保证顺序

批量拉取的时候,同一个保序依据的消息只会拉取一条,多线程消费的速度并不会影响顺序消息的消费。

延迟消息应用

事务消息的回查——可以用到

消费限流

  1. 控制台控制group消费限流
  2. 代码里guava RateLimiter限制
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。