Kafka 消息中间件

一、消息队列的应用场景

1. 异步处理

• 将串行执行通过消息中间件实现异步执行。
  

2. 系统解耦

• 通过消息中间件，实现两个服务间解耦，降低耦合性。
  

3. 流量削峰

• 将请求存入消息队列，慢慢消费，避免高并发同时到达服务器。
  

4. 日志处理（大数据领域常见）

二、Kafka 简介

1. Kafka的概念

kafka官网：http://kafka.apache.org/

1. Producers: 可以有很多的应用程序，将消息数据放入Kafka集群中
2. Consumers: 可以有很多的应用程序，将消息数据从Kafka集群中拉取出来
3. Connectors: Kafka的连接器可以将数据库中的数据导入到Kafka，也可以将Kafka的数据到出到数据库中
4. Stream Processors: 流处理器可以从Kafka中拉取数据，也可以将数据写入到Kafka中

• producer：发布消息的对象称之为主题生产者（Kafka topic producer）
• topic：Kafka将消息分门别类，每一类的消息称之为一个主题（Topic）
• consumer：订阅消息并处理发布的消息的对象称之为主题消费者（consumers）
• broker：已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理（Broker）。 消费者可以订阅一个或多个主题（topic），并从Broker拉数据，从而消费这些已发布的消息。

2. 消息中间件的对比、选择

• 对比

• 选择

3. Kafka安装配置

Kafka对于zookeeper是强依赖，保存kafka相关的节点数据，所以安装Kafka之前必须先安装zookeeper

• Docker安装zookeeper

下载镜像：

docker pull bitnami/zookeeper
1

创建容器:

docker run -d --name zookeeper -p 2181:2181 -e ALLOW_ANONYMOUS_LOGIN=yes  bitnami/zookeeper
1

• Docker安装Kafka

下载镜像：

docker pull bitnami/kafka
1

创建容器:

docker run -d --name kafka \
-e KAFKA_ADVERTISED_HOST_NAME=192.168.137.136 \
-e KAFKA_ZOOKEEPER_CONNECT=192.168.137.136:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.137.136:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
-e KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
-e ALLOW_PLAINTEXT_LISTENER=yes \
--net=host bitnami/kafka
1
2
3
4
5
6
7
8

• 安装Kafka可视化web工具 – kafdrop 便于操作消息中间件

下载镜像：

docker pull obsidiandynamics/kafdrop
1

创建容器:

docker run -d --name kafdrop -p 9100:9100\
    -e JVM_OPTS="-Xms32M -Xmx64M -Dserver.port=9100" \
    -e KAFKA_BROKERCONNECT=192.168.137.136:9092\
    -e SERVER_SERVLET_CONTEXTPATH="/" \
    obsidiandynamics/kafdrop
1
2
3
4
5

访问 http://192.168.137.136:9100/可以看到Kafdrop界面

4. Kafka的基准测试

https://www.bilibili.com/video/BV19y4y1b7Uo?p=6&vd_source=c581024b8cd9585ec6a75c56ac05571a

三、Kafka入门案例

1. 创建一个maven项目

2. 引入依赖

<dependency>
    <groupId>org.apache.kafkagroupId>
    <artifactId>kafka-clientsartifactId>
    <version>2.4.0version>
dependency>
1
2
3
4
5

3. 生产者代码

/**
 * 生产者
 */
public class ProducerQuickStart {

    public static void main(String[] args) {
        // 1.kafka的配置信息
        Properties properties = new Properties();
        // kafka的连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.136:9092");
        // 发送失败，失败的重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 5);
        // 批量发送的阈值，先缓存达到阈值后一起发送到中间件
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 存留时间，如果消息带了一秒了，就发送，不管大美达到阈值
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        // 消息发送成功的确认,all-主从都搞定才返回成功, 0-只管发，1-只要主成功了就返回成功
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        // 消息key的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 消息value的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 2.生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        // 封装发送的消息
        ProducerRecord<String, String> record = new ProducerRecord<String, String>("itheima-topic", "100001", "hello kafka");

        // 3.发送消息
        producer.send(record);

        // 4.关闭消息通道，必须关闭，否则消息发送不成功
        producer.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

4. 消费者代码

/**
 * 消费者
 */
public class ConsumerQuickStart {

    public static void main(String[] args) {
        // 1.添加kafka的配置信息
        Properties properties = new Properties();
        // kafka的连接地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.136:9092");
        // 消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");
        // 消息的反序列化器
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 2.消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        // 3.订阅主题
        consumer.subscribe(Collections.singletonList("itheima-topic"));

        // 当前线程一直处于监听状态
        while (true) {
            //4.获取消息
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : consumerRecords) {
                System.out.println(record.key() + "=====" + record.value());
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

5. 总结

• 生产者发送消息，同一个组中的多个消费者只能有一个消费者接收消息

• 生产者发送消息，如果有多个组，每个组中只能有一个消费者接收消息,如果想要实现广播的效果，可以让每个消费者单独有一个组即可，这样每个消费者都可以接收到消息

四、Kafka高可用设计

1. 集群

• Kafka 的服务器端由被称为 Broker 的服务进程构成，即一个 Kafka 集群由多个 Broker 组成
• 这样如果集群中某一台机器宕机，其他机器上的 Broker 也依然能够对外提供服务。这其实就是 Kafka 提供高可用的手段之一
• key

​ Kafka根据传递消息的key来进行分区的分配，分区器会使用key的哈希值（采用Murmur2Hash算法）对partition数量取模，决定要把消息发送到哪个partition上，即hash(key) % numPartitions，这就保证了相同key的消息一定会被路由到相同的分区。

​ 如果你没有指定key，Kafka会随机找一个分区发送无key的消息（各个版本可能会不同），然后把这个分区号加入到缓存中以备后面直接使用。

• Zookeeper

  kafka对zookeeper是强依赖的，是以zookeeper作为基础的，即使不做集群，也需要zk的支持。Kafka通过Zookeeper管理集群配置，选举leader。

• 分区Partition

  主题（topic）可以被分为多个分区（partition），一个分区（partition）只能属于单个主题（topic），一个主题（topic）至少有一个分区（partition），同一个主题（topic）下的不同分区（partition）包含的消息是不同的。同一个主题（topic）中的分区（partition）有可能会分布在多个机器（Broker ）上，由此来实现 kafka 的伸缩性。

  每个主题（topic）至少有一个分区（partition）。分区（partition）中的数据使用多个文件进行存储。kafka保证数据分区有序而不是主题有序，即：分区（partition）中的数据是有序的，分区（partition）之间的数据是没有顺序的。在需要严格保证消息的消费顺序的场景下，需要将分区（partition）数目设为1。

2. 备份机制(Replication）

Kafka 中消息的备份又叫做 副本（Replica）

Kafka 定义了两类副本：

• 领导者副本（Leader Replica）
• 追随者副本（Follower Replica）

同步方式

ISR（in-sync replica）需要同步复制保存的follower

如果leader失效后，需要选出新的leader，选举的原则如下：

第一：选举时优先从ISR中选定，因为这个列表中follower的数据是与leader同步的

第二：如果ISR列表中的follower都不行了，就只能从其他follower中选取

极端情况，就是所有副本都失效了，这时有两种方案

第一：等待ISR中的一个活过来，选为Leader，数据可靠，但活过来的时间不确定

第二：选择第一个活过来的Replication，不一定是ISR中的，选为leader，以最快速度恢复可用性，但数据不一定完整

五、Kafka生产者详解

1. 发送类型

• 同步发送

  使用send()方法发送，它会返回一个Future对象，调用get()方法进行等待，就可以知道消息是否发送成功

RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
System.out.println(recordMetadata.offset());
1
2

• 异步发送

  调用send()方法，并指定一个回调函数，服务器在返回响应时调用函数

//异步消息发送
producer.send(kvProducerRecord, new Callback() {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if(e != null){
            System.out.println("记录异常信息到日志表中");
        }
        System.out.println(recordMetadata.offset());
    }
});
1
2
3
4
5
6
7
8
9
10

2. 参数详解

• ack

代码的配置方式：

//ack配置  消息确认机制
prop.put(ProducerConfig.ACKS_CONFIG,"all");
1
2

参数的选择说明

• retries

生产者从服务器收到的错误有可能是临时性错误，在这种情况下，retries参数的值决定了生产者可以重发消息的次数，如果达到这个次数，生产者会放弃重试返回错误，默认情况下，生产者会在每次重试之间等待100ms

代码中配置方式：

//重试次数
prop.put(ProducerConfig.RETRIES_CONFIG,10);
1
2

• 消息压缩

默认情况下， 消息发送时不会被压缩。

代码中配置方式：

//数据压缩
prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
1
2

使用压缩可以降低网络传输开销和存储开销，而这往往是向 Kafka 发送消息的瓶颈所在。

3. 消息存储

3.1 存储方式

为了规避随机读写带来的时间消耗，kafka 采用顺序写的方式存储数据。kafka还有一个性能策略：零拷贝

消息从发送到落地保存，broker 维护的消息日志本身就是文件目录，每个文件都是二进制保存，生产者和消费者使用相同的格式来处理。在消费者获取消息时，服务器先从硬盘读取数据到内存，然后把内存中的数据原封不动的通过 socket 发送给消费者。即：操作系统将数据从页缓存传输到 socket。

3.2 日志文件

kafka 使用日志文件的方式来保存生产者和发送者的消息，每条消息都有一个 offset 值来表示它在分区中的偏移量。Kafka 中存储的一般都是海量的消息数据，为了避免日志文件过大，一个分片 并不是直接对应在一个磁盘上的日志文件，而是对应磁盘上的一个目录，这个目录的命名规则是_。
log分段：

每个分片目录中，kafka 通过分段的方式将 数据 分为多个 LogSegment，一个 LogSegment 对应磁盘上的一个日志文件（00000000000000000000.log）和一个索引文件(如上：00000000000000000000.index)，其中日志文件是用来记录消息的。索引文件是用来保存消息的索引。每个LogSegment 的大小可以在server.properties 中log.segment.bytes=107370 (设置分段大小,默认是1gb)选项进行设置。

segment 的 index file 和 data file 2 个文件一一对应，成对出现，后缀".index"和“.log”分别表示为 segment 索引文件、数据文件.命名规则：partion 全局的第一个 segment从 0 开始，后续每个 segment 文件名为上一个 segment文件最后一条消息的 offset 值进行递增。数值最大为 64 位long 大小，19 位数字字符长度，没有数字用 0 填充
第一个 log 文件的最后一个 offset 为:5376,所以下一个segment 的文件命名为: 0000000000000005377.log。对应的 index 为 00000000000000005376.index

kafka 这种分片和分段策略，避免了数据量过大时，数据文件文件无限扩张带来的隐患，更有助于消息文件的维护以及被消费的消息的清理。
在 partition 中通过 offset 查找 message过程
根据 offset 的值，查找 segment 段中的 index 索引文件。由于索引文件命名是以上一个文件的最后一个offset 进行命名的，所以，使用二分查找算法能够根据offset 快速定位到指定的索引文件
找到索引文件后，根据 offset 进行定位，找到索引文件中的匹配范围的偏移量position。（kafka 采用稀疏索引的方式来提高查找性能）

得到 position 以后，再到对应的 log 文件中，从 position处开始查找 offset 对应的消息，将每条消息的 offset 与目标 offset 进行比较，直到找到消息
比如说，我们要查找 offset=2490 这条消息，那么先找到00000000000000000000.index, 然后找到[2487,49111]这个索引，再到 log 文件中，根据 49111 这个 position 开始查找，比较每条消息的 offset 是否大于等于 2490。最后查找到对应的消息以后返回

3.3 有效期

据消息的保留时间，当消息在 kafka 中保存的时间超过了指定的时间，就会触发清理过程。

根据 topic 存储的数据大小，当 topic 所占的日志文件大小大于一定的阀值，则可以开始删除最旧的消息。
通过 log.retention.bytes 和 log.retention.hours 这两个参数来设置，当其中任意一个达到要求，都会执行删除。默认的保留时间是：7 天
kafka会启动一个后台线程，定期检查是否存在可以删除的消息

六、Kafka消费者详解

1. 消费者组

• 消费者组（Consumer Group） ：指的就是由一个或多个消费者组成的群体
• 一个发布在Topic上消息被分发给此消费者组中的一个消费者
  • 所有的消费者都在一个组中，那么这就变成了queue模型
  • 所有的消费者都在不同的组中，那么就完全变成了发布-订阅模型

2. 消息有序性

应用场景：

• 即时消息中的单对单聊天和群聊，保证发送方消息发送顺序与接收方的顺序一致
• 充值转账两个渠道在同一个时间进行余额变更，短信通知必须要有顺序

topic分区中消息只能由消费者组中的唯一一个消费者处理，所以消息肯定是按照先后顺序进行处理的。但是它也仅仅是保证Topic的一个分区顺序处理，不能保证跨分区的消息先后处理顺序。 所以，如果你想要顺序的处理Topic的所有消息，那就只提供一个分区。

3. 提交偏移量

kafka不会像其他JMS队列那样需要得到消费者的确认，消费者可以使用kafka来追踪消息在分区的位置（偏移量）

消费者会往一个叫做_consumer_offset的特殊主题发送消息，消息里包含了每个分区的偏移量。如果消费者发生崩溃或有新的消费者加入群组，就会触发再均衡

正常的情况

如果消费者2挂掉以后，会发生再均衡，消费者2负责的分区会被其他消费者进行消费

再均衡后不可避免会出现一些问题

问题一：

如果提交偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理。

问题二：

如果提交的偏移量大于客户端的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。

如果想要解决这些问题，还要知道目前kafka提交偏移量的方式：

提交偏移量的方式有两种，分别是自动提交偏移量和手动提交

• 自动提交偏移量

当enable.auto.commit被设置为true，提交方式就是让消费者自动提交偏移量，每隔5秒消费者会自动把从poll()方法接收的最大偏移量提交上去

• 手动提交 ，当enable.auto.commit被设置为false可以有以下三种提交方式
  • 提交当前偏移量（同步提交）
  • 异步提交
  • 同步和异步组合提交

1.提交当前偏移量（同步提交）

把enable.auto.commit设置为false,让应用程序决定何时提交偏移量。使用commitSync()提交偏移量，commitSync()将会提交poll返回的最新的偏移量，所以在处理完所有记录后要确保调用了commitSync()方法。否则还是会有消息丢失的风险。

只要没有发生不可恢复的错误，commitSync()方法会一直尝试直至提交成功，如果提交失败也可以记录到错误日志里。

while (true){
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record.value());
        System.out.println(record.key());
        try {
            consumer.commitSync();//同步提交当前最新的偏移量
        }catch (CommitFailedException e){
            System.out.println("记录提交失败的异常："+e);
        }

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

2.异步提交

手动提交有一个缺点，那就是当发起提交调用时应用会阻塞。当然我们可以减少手动提交的频率，但这个会增加消息重复的概率（和自动提交一样）。另外一个解决办法是，使用异步提交的API。

while (true){
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record.value());
        System.out.println(record.key());
    }
    consumer.commitAsync(new OffsetCommitCallback() {
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
            if(e!=null){
                System.out.println("记录错误的提交偏移量："+ map+",异常信息"+e);
            }
        }
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3.同步和异步组合提交

异步提交也有个缺点，那就是如果服务器返回提交失败，异步提交不会进行重试。相比较起来，同步提交会进行重试直到成功或者最后抛出异常给应用。异步提交没有实现重试是因为，如果同时存在多个异步提交，进行重试可能会导致位移覆盖。

举个例子，假如我们发起了一个异步提交commitA，此时的提交位移为2000，随后又发起了一个异步提交commitB且位移为3000；commitA提交失败但commitB提交成功，此时commitA进行重试并成功的话，会将实际上将已经提交的位移从3000回滚到2000，导致消息重复消费。

try {
    while (true){
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.value());
            System.out.println(record.key());
        }
        consumer.commitAsync();
    }
}catch (Exception e){+
    e.printStackTrace();
    System.out.println("记录错误信息："+e);
}finally {
    try {
        consumer.commitSync();
    }finally {
        consumer.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

七、参考资料

https://www.bilibili.com/video/BV19y4y1b7Uo/?p=3&spm_id_from=pageDriver&vd_source=c581024b8cd9585ec6a75c56ac05571a