消息队列_消息队列：kafka

概念

kafka是一个分布式的基于发布/订阅模式的消息队列，主要用于大数据实时处理领域。

要理解kafka首先要有分布式的概念，要有消息队列的概念。分布式系统最大的优势就是解耦和削峰，这种情况下，A系统生成了一个消息，B系统异步获取，那么就需要一个存放消息的消息队列(MQ)。

相比较传统的消息队列，消息被消费确认后会删除，而kafka有持久化功能，B系统消费完，C系统还可以再次消费。kafka默认消息保存168小时，即7天，可配置。

kafka的核心概念有：broker、topic、partition、leader、follower、ISR、acks、offset、lEO和HW，下面分别理解一下。

broker、topic、partition

Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker。
一个broker服务上可以有多个topic，topic是kafka消息队列的一个逻辑概念，partition是具体的物理概念，比如topic=first，partition=0，消息实际存储在broker的first-0目录下。
partition分区的优势：高并发的情况下，可以动态扩展，生产者往多个分区发生消息，能提高并发量。
生产者发送给kafka的消息是个ProducerRecord对象，ProducerRecord提供了6种构造器，如下：

不管使用哪一种构造器，最终实现如下：

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable headers) {
if (topic == null) {
throw new IllegalArgumentException("Topic cannot be null.");
    } else if (timestamp != null && timestamp < 0L) {
throw new IllegalArgumentException(String.format("Invalid timestamp: %d. Timestamp should always be non-negative or null.", timestamp));
    } else if (partition != null && partition < 0) {
throw new IllegalArgumentException(String.format("Invalid partition: %d. Partition number should always be non-negative or null.", partition));
    } else {
this.topic = topic;
this.partition = partition;
this.key = key;
this.value = value;
this.timestamp = timestamp;
this.headers = new RecordHeaders(headers);
    }
}

其中，topic必须指定，partition的值有以下参数确定：

1)指定partition值，消息直接放入指定的分区；

2)既没有partition也没有key，采用round-robin算法，随机生成一个整数，与topic的partition数取余得到partition值，后续在这个整数上自增；

3)没有指定partition但有key，将key的hash值与topic的Partition数进行取余得到partition值；

4)既有partition又有key，以partition为准，key失效。

leader、follower

每一个broker都有一个leader和多个follower，leader和ollower不能在同一个节点上，否则服务故障，主从都会失效。

producer可靠性保障和ISR

思考：producer向kafka发送消息，如何保障消息的可靠性？

有两种方案：

1)为了保障消息不丢失，至少要半数以follower上同步完成，broker发确认。

半数以上同步完成，有且只有一个follower才有可能获得半数以上投票，这样就不会存在选出来多个leader了，而且这半数以上的follower都完成了同步，能保证参与选举的follower(自己本身也是候选人)都满足成为leader的资格。

比如有7个follower，半数以上即4个follower同步完成，全部投选1号为leader，leader唯一可确定。如果选举是2，2,的结果，则没有任何一个follower有资格成为leader。

2)等待所有follower全部同步完成发确认，此时可以随便选举一个follower作为leader。

第一种方案，n个副本挂掉了，需要2n+1个副本才能恢复数据；

第二种方案，n个副本挂掉了，只需要n+1个副本就能恢复数据；

优选第二种方案：第一种方案当n+1个同步完成，n个挂掉，需要2n+1个副本，而每个partition的数据量都很大，会造成数据冗余。

但是第二种方案存在万一某个follower迟迟无法同步完成，会造成严重延迟，为了解决这个问题，kafka维护了一个动态follower集合——ISR：

ISR初始集合是所有follower，如果某个follower故障，在指定时间内无法同步，会被踢出ISR，这样只要保证ISR中的剩余follower同步完成就可以发送确认。

replica.lag.time.ms参数用来指定超时的时间阀值。

acks

acks用来指定broker什么时候发收到消息的确认：

0：producer不等待broker的ack，broker一收到消息还没写入就返回，一旦broker故障，消息会丢失，适用日志记录，对消息丢失容忍度比较高的场景；
1：producer等待broker的ack，broker的leader写入成功返回ack，如果follower同步成功之前leader挂掉，消息会丢失。
-1：producer等待broker的ack，broker的leader和所有follower都写入成功后发ack，但是此时leader挂掉了，没有发送ack，producer超时没收到ack，就会重发消息，造成消息重复。