Kafka 笔记

HuldaGnodim · 发表于 2022-11-5 12:39

消息队列

Kafka的应用场景

消息队列的两种模型

生产者、消费者模型
消息队列的模式

Kafka集群搭建

注意：

Kafka的生产者/消费者/工具

安装Kafka集群，可以测试以下
推荐大家开发的使用Kafka Tool

Kafka的基准测试工具

Kafka中提供了内置的性能测试工具

5000000 records sent, 11825.446943 records/sec (11.28 MB/sec), 2757.61 ms avg latency

Kafka Java API开发

生产者程序开发

创建连接
创建一个生产者对象KafkaProducer调用send方法发送消息（ProducerRecor，封装是key-value键值对）调用Future.get表示等带服务端的响应关闭生产者

public class KafkaProducerTest { public static void main(String[] args) throws ExecutionException, InterruptedException { // 1. 创建用于连接Kafka的Properties配置 Properties props = new Properties(); props.put("bootstrap.servers", "node1.itcast.cn:9092"); props.put("acks", "all"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 2. 创建一个生产者对象KafkaProducer KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props); // 3. 发送1-100的消息到指定的topic中 for(int i = 0; i < 100; ++i) { // 构建一条消息，直接new ProducerRecord ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", null, i + ""); Future<RecordMetadata> future = kafkaProducer.send(producerRecord); // 调用Future的get方法等待响应 future.get(); System.out.println("第" + i + "条消息写入成功！"); } // 4.关闭生产者 kafkaProducer.close(); }}消费者程序开发

/** * 消费者程序 * * 1.创建Kafka消费者配置 * Properties props = new Properties(); * props.setProperty("bootstrap.servers", "node1.itcast.cn:9092"); * props.setProperty("group.id", "test"); * props.setProperty("enable.auto.commit", "true"); * props.setProperty("auto.commit.interval.ms", "1000"); * props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); * props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); * * 2.创建Kafka消费者 * 3.订阅要消费的主题 * 4.使用一个while循环，不断从Kafka的topic中拉取消息 * 5.将将记录（record）的offset、key、value都打印出来 */public class KafkaConsumerTest { public static void main(String[] args) { // 1.创建Kafka消费者配置 Properties props = new Properties(); props.setProperty("bootstrap.servers", "node1.itcast.cn:9092"); // 消费者组（可以使用消费者组将若干个消费者组织到一起），共同消费Kafka中topic的数据 // 每一个消费者需要指定一个消费者组，如果消费者的组名是一样的，表示这几个消费者是一个组中的 props.setProperty("group.id", "test"); // 自动提交offset props.setProperty("enable.auto.commit", "true"); // 自动提交offset的时间间隔 props.setProperty("auto.commit.interval.ms", "1000"); // 拉取的key、value数据的 props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 2.创建Kafka消费者 KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props); // 3. 订阅要消费的主题 // 指定消费者从哪个topic中拉取数据 kafkaConsumer.subscribe(Arrays.asList("test")); // 4.使用一个while循环，不断从Kafka的topic中拉取消息 while(true) { // Kafka的消费者一次拉取一批的数据 ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5)); // 5.将将记录（record）的offset、key、value都打印出来 for (ConsumerRecord<String, String> consumerRecord : consumerRecords) { // 主题 String topic = consumerRecord.topic(); // offset：这条消息处于Kafka分区中的哪个位置 long offset = consumerRecord.offset(); // key\value String key = consumerRecord.key(); String value = consumerRecord.value(); System.out.println("topic: " + topic + " offset:" + offset + " key:" + key + " value:" + value); } } }}生产者使用异步方式生产消息

使用匿名内部类实现Callback接口，该接口中表示Kafka服务器响应给客户端，会自动调用onCompletion方法

// 二、使用异步回调的方式发送消息ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", null, i + "");kafkaProducer.send(producerRecord, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { // 1. 判断发送消息是否成功 if(exception == null) { // 发送成功 // 主题 String topic = metadata.topic(); // 分区id int partition = metadata.partition(); // 偏移量 long offset = metadata.offset(); System.out.println("topic:" + topic + " 分区id：" + partition + " 偏移量：" + offset); } else { // 发送出现错误 System.out.println("生产消息出现异常！"); // 打印异常消息 System.out.println(exception.getMessage()); // 打印调用栈 System.out.println(exception.getStackTrace()); } }});Kafka中的重要概念

broker
producer：生产者consumer：消费者
topic：主题，一个Kafka集群中，可以包含多个topic。一个topic可以包含多个分区
partition：Kafka集群的分布式就是由分区来实现的。一个topic中的消息可以分布在topic中的不同partition中replica：副本，实现Kafkaf集群的容错，实现partition的容错。一个topic至少应该包含大于1个的副本consumer group：消费者组，一个消费者组中的消费者可以共同消费topic中的分区数据。每一个消费者组都一个唯一的名字。配置group.id一样的消费者是属于同一个组中offset：偏移量。相对消费者、partition来说，可以通过offset来拉取数据

消费者组

幂等性

生产者消息重复问题
在Kafka中可以开启幂等性

事务编程

开启事务的条件

// 开启事务必须要配置事务的IDprops.put("transactional.id", "dwd_user");

// 配置事务的隔离级别props.put("isolation.level","read_committed");// 关闭自动提交，一会我们需要手动来提交offset，通过事务来维护offsetprops.setProperty("enable.auto.commit", "false");

生产者

如果使用了事务，不要使用异步发送

public class TransactionProgram { public static void main(String[] args) { // 1. 调用之前实现的方法，创建消费者、生产者对象 KafkaConsumer<String, String> consumer = createConsumer(); KafkaProducer<String, String> producer = createProducer(); // 2. 生产者调用initTransactions初始化事务 producer.initTransactions(); // 3. 编写一个while死循环，在while循环中不断拉取数据，进行处理后，再写入到指定的topic while(true) { try { // (1) 生产者开启事务 producer.beginTransaction(); // 这个Map保存了topic对应的partition的偏移量 Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>(); // 从topic中拉取一批的数据 // (2) 消费者拉取消息 ConsumerRecords<String, String> concumserRecordArray = consumer.poll(Duration.ofSeconds(5)); // (3) 遍历拉取到的消息，并进行预处理 for (ConsumerRecord<String, String> cr : concumserRecordArray) { // 将1转换为男，0转换为女 String msg = cr.value(); String[] fieldArray = msg.split(","); // 将消息的偏移量保存 // 消费的是ods_user中的数据 String topic = cr.topic(); int partition = cr.partition(); long offset = cr.offset(); int i = 1 / 0; // offset + 1：offset是当前消费的记录（消息）对应在partition中的offset，而我们希望下一次能继续从下一个消息消息 // 必须要+1，从能消费下一条消息 offsetMap.put(new TopicPartition(topic, partition), new OffsetAndMetadata(offset + 1)); // 将字段进行替换 if(fieldArray != null && fieldArray.length > 2) { String sexField = fieldArray[1]; if(sexField.equals("1")) { fieldArray[1] = "男"; } else if(sexField.equals("0")){ fieldArray[1] = "女"; } } // 重新拼接字段 msg = fieldArray[0] + "," + fieldArray[1] + "," + fieldArray[2]; // (4) 生产消息到dwd_user topic中 ProducerRecord<String, String> dwdMsg = new ProducerRecord<>("dwd_user", msg); // 发送消息 Future<RecordMetadata> future = producer.send(dwdMsg); try { future.get(); } catch (Exception e) { e.printStackTrace(); producer.abortTransaction(); }// new Callback()// {// @Override// public void onCompletion(RecordMetadata metadata, Exception exception) {// // 生产消息没有问题// if(exception == null) {// System.out.println("发送成功:" + dwdMsg);// }// else {// System.out.println("生产消息失败:");// System.out.println(exception.getMessage());// System.out.println(exception.getStackTrace());// }// }// }); } producer.sendOffsetsToTransaction(offsetMap, "ods_user"); // (6) 提交事务 producer.commitTransaction(); }catch (Exception e) { e.printStackTrace(); // (7) 捕获异常，如果出现异常，则取消事务 producer.abortTransaction(); } } } // 一、创建一个消费者来消费ods_user中的数据 private static KafkaConsumer<String, String> createConsumer() { // 1. 配置消费者的属性（添加对事务的支持） Properties props = new Properties(); props.setProperty("bootstrap.servers", "node1.itcast.cn:9092"); props.setProperty("group.id", "ods_user"); // 配置事务的隔离级别 props.put("isolation.level","read_committed"); // 关闭自动提交，一会我们需要手动来提交offset，通过事务来维护offset props.setProperty("enable.auto.commit", "false"); // 反序列化器 props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 2. 构建消费者对象 KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props); // 3. 订阅一个topic kafkaConsumer.subscribe(Arrays.asList("ods_user")); return kafkaConsumer; } // 二、编写createProducer方法，用来创建一个带有事务配置的生产者 private static KafkaProducer<String, String> createProducer() { // 1. 配置生产者带有事务配置的属性 Properties props = new Properties(); props.put("bootstrap.servers", "node1.itcast.cn:9092"); // 开启事务必须要配置事务的ID props.put("transactional.id", "dwd_user"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 2. 构建生产者 KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props); return kafkaProducer; }}Kafka中的分区副本机制

生产者的分区写入策略

轮询（按照消息尽量保证每个分区的负载）策略，消息会均匀地分布到每个partition
随机策略（不使用）按key写入策略，key.hash() % 分区的数量自定义分区策略（类似于MapReduce指定分区）

乱序问题

在Kafka中生产者是有写入策略，如果topic有多个分区，就会将数据分散在不同的partition中存储当partition数量大于1的时候，数据（消息）会打散分布在不同的partition中如果只有一个分区，消息是有序的

消费组Consumer Group Rebalance机制

触发时机
- 消费者数量发生变化
- topic的数量发生变化
- partition的数量发生变化
不良影响

消费者的分区分配策略

分区分配策略：保障每个消费者尽量能够均衡地消费分区的数据，不能出现某个消费者消费分区的数量特别多，某个消费者消费的分区特别少

Range分配策略（范围分配策略）：Kafka默认的分配策略
RoundRobin分配策略（轮询分配策略）
Striky粘性分配策略

副本的ACK机制

producer是不断地往Kafka中写入数据，写入数据会有一个返回结果，表示是否写入成功。这里对应有一个ACKs的配置。

根据业务情况来选择ack机制，是要求性能最高，一部分数据丢失影响不大，可以选择0/1。如果要求数据一定不能丢失，就得配置为-1/all。

分区中是有leader和follower的概念，为了确保消费者消费的数据是一致的，只能从分区leader去读写消息，follower做的事情就是同步数据，Backup。
高级API（High-Level API）、低级API（Low-Level API）

高级API就是直接让Kafka帮助管理、处理分配、数据
低级API：由编写的程序自己控制逻辑

Kafka原理

leader和follower

注意和ZooKeeper区分

AR\ISR\OSR

leader选举

Controller：controller是kafka集群的老大，是针对Broker的一个角色
Leader：是针对partition的一个角色
如果Kafka是基于ZK来进行选举，ZK的压力可能会比较大。例如：某个节点崩溃，这个节点上不仅仅只有一个leader，是有不少的leader需要选举。通过ISR快速进行选举。
leader的负载均衡

bin/kafka-leader-election.sh --bootstrap-server node1.itcast.cn:9092 --topic test --partition=2 --election-type preferredKafka读写流程

写流程
读流程

Kafka的物理存储

Kafka的数据组织结构
- segment
深入了解读数据的流程

消息传递的语义性

Flink里面有对应的每种不同机制的保证，提供Exactly-Once保障（二阶段事务提交方式）

Kafka的消息不丢失

消费者消费不丢失：重点控制offset

数据积压

数据清理&配额限速

数据清理
- Log Deletion（日志删除）：如果消息达到一定的条件（时间、日志大小、offset大小），Kafka就会自动将日志设置为待删除（segment端的后缀名会以 .delete结尾），日志管理程序会定期清理这些日志
- Log Compaction（日志合并）
配额限速

		自动登录	找回密码
密码			立即注册