Kafka入门使用_记忆中逝去的往事_kafka入门

大大的周 02-07 993

本文为学习B站教程所总结出来的笔记！！！ kafka入门到精通教程|kafka快速入门

Kafka的定义

Message Queue（MQ），消息队列中间件；

Kafka是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，因其可水平扩展和高吞吐率而被广泛使用！

Kafka的安装与使用（Windows）

1、运行zookeeper，下载与安装可见：https://blog.csdn.net/z1790424577/article/details/106660764

2、下载kafka，地址为：https://kafka.apache.org/downloads 下载完成并解压缩后，打开控制台进入kafka安装目录输入如下命令启动kafka：

.\bin\windows\kafka-server-start.bat .\config\server.properties

server.properties核心配置详解：

属性默认值描述broker.id0每个broker都可以??个唯?的?负整数id进?标识log.dirs/tmp/kafka-logskafka存放数据的路径。这个路径并不是唯?的，可以是多个，路径之间只需要使?逗号分隔即可；每当创建新partition时，都会选择在包含最少partitions的路径下进?。listenersPLAINTEXT://:9092server接受客户端连接的地址及端口zookeeper.connectlocalhost:2181kafka连接zookeeper的地址log.retention.hours168每个?志?件删除之前保存的时间。默认数据保存时间对所有topic都?样。num.partitions1创建topic的默认分区数default.replication.factor1?动创建topic的默认副本数量min.insync.replicas1当producer设置acks为-1时，min.insync.replicas指定replicas的最?数?（必须确认每?个repica的写数据都是成功的），如果这个数?没有达到，producer发送消息会产?异常delete.topic.enablefalse是否允许删除主题

3、测试基本使用，控制台输入如下命令（Windows需在/bin/windows/目录下执行）：

#1、创建主题（topic） kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test #为主题创建多个分区，由--partitions参数指定 kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 2 --topic test #2、查看已创建topic kafka-topics.bat --list --bootstrap-server localhost:9092 #查看topic详细信息 kafka-topics.bat --describe --bootstrap-server localhost:9092 --topic testA #3、创建生产者 kafka-console-producer.bat --broker-list localhost:9092 --topic test #4、创建消费者 kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning #从头开始消费 #5、查看消费组及信息 kafka-consumer-groups.bat --bootstrap-server localhost:9092 --list #查看当前主题下有哪些消费组 kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group testGroup #查看消费组中的具体信息

往生产者窗口写入消息，消费者窗口也能同步的接收到消息：消费者组指标描述：

Currennt-offset：当前消费组的已消费偏移量Log-end-offset：主题对应分区消息的结束偏移量(HW)Lag：当前消费组未消费的消息数 Kafka的基本概念名称描述Topic用户定义并配置在Kafka服务器，用于建立生产者和消息者之间的订阅关系：生产者发送消息到指定的Topic下，消息者从这Topic下消费消息；Broker消息中间件处理节点，?个Kafka节点就是?个Broker，?个或者多个Broker可以组成?个Kafka集群Producer消息的发送方，负责生成消息并发送到KafkaConsumer消息的使用方，负责消费Kafka服务器上的消息Partition?个Topic可以分为多个Partition（一个主题中的消息量是非常大的，因此可以通过分区的设置，来分布式存储这些消息），每个Partition内部消息是有序的ConsumerGroup每个Consumer属于?个特定的Consumer Group，?条消息可以被多个不同的Consumer Group消费，但是?个Consumer Group中只能有?个Consumer能够消费该消息Replication-Factor副本是对分区的备份。在集群中，不同的副本会被部署在不同的broker上

分区图示： Producer通过?络发送消息到Kafka集群，然后Consumer来进?消费，如下图：副本是对分区的备份。在集群中，不同的副本会被部署在不同的broker上，查看topic详细信息如下：通过查看主题信息，其中的关键数据：

replicas：当前副本存在的broker节点；leader：副本里的概念，leader专?用来接收消息。接收到消息，其他follower通过poll的方式来同步数据，每个partition都在不同的Broker上。消息发送方要把消息发给哪个broker？就看副本的leader是在个broker上面；follower：follower负责从leader同步数据，不提供读写；isr：可以同步的broker节点和已同步的broker节点，存放在isr集合中；

Kafka中的细节

消息是按照发送的顺序进行存储，因此消费者在消费消息时可以指明主题中消息的偏移量（默认情况下，是从最后一个消息的下一个偏移量开始消费）；如果多个消费者在同?个消费组，那么只有?个消费者可以收到订阅的topic中的消息（换?之，同?个消费组中只能有?个消费者收到?个topic中的消息）；不同的消费组订阅同?个topic，那么不同的消费组中只有?个消费者能收到消息；

Kafka消息存放目录结构如下： __consumer_offsets是Kafka内部主题，默认创建50个分区（可以通过offsets.topic.num.partitions设置），主要用于存储消费者的偏移量，图示如下：消费者会定期将自己消费分区的offset提交给__consumer_offsets，key是consumerGroupId+topic+分区号，value就是当前offset的值，并且kafka会定期清理topic里的消息仅保留最新的那条数据，通过如下公式可以选出consumer消费的offset要提交到__consumer_offsets的哪个分区：

hash(consumerGroupId) % __consumer_offsets主题的分区数

文件具体作用如下：

00000000000000000000.log：消息数据；00000000000000000000.index：用于根据位移值快速查找消息所在文件位置；00000000000000000000.timeindex：用于根据时间戳快速查找特定消息的位移值； Kafka的Java客户端-生产者

1、引入依赖（建议版本与Kafka一致）

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>3.1.0</version> </dependency>

2、生产者发送消息的基本实现

public class MyProducer { private final static String TOPIC_NAME = "my-topic"; public static void main(String[] args) throws ExecutionException, InterruptedException { Properties props = new Properties(); //Kafka地址 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092"); //把发送的key从字符串序列化为字节数组 props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); //把发送消息value从字符串序列化为字节数组 props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); //缓冲区大小设置--kafka默认会创建?个消息缓冲区，?来存放要发送的消息，缓冲区是32m props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); //拉取设置--kafka本地线程会去缓冲区中?次拉16k的数据，发送到broker props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); //如果线程拉不到16k的数据，间隔10ms也会将已拉到的数据发到broker props.put(ProducerConfig.LINGER_MS_CONFIG, 10); //ack参数配置，适用于同步发送情况，下面细讲 //props.put(ProducerConfig.ACKS_CONFIG, "1"); //重试次数 props.put(ProducerConfig.RETRIES_CONFIG, 3); //重试间隔设置，发送失败会重试，默认重试间隔100ms props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300); Producer<String, String> producer = new KafkaProducer<>(props); //发送主题与内容 ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC_NAME, "123"); //发送到指定分区 //ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME,0, "1","123"); //未指定分区，则会通过业务key的hash运算，算出消息往哪个分区上发 ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME,"2","123"); //同步发消息，在收到kafka的ack告知发送成功之前一直处于阻塞状态 //RecordMetadata metadata = producer.send(producerRecord).get(); //=====阻塞======= //System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset()); //异步发消息 producer.send(producerRecord, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception != null) { System.err.println("发送消息失败：" + exception.getStackTrace()); } if (metadata != null) { System.out.println("异步方式发送消息结果：" + "topic-" +metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" + metadata.offset()); } } }); //因为是异步发，避免main线程立马结束，故休眠1s Thread.sleep(1000); } } //同步方式发送消息结果：topic-my-topic|partition-0|offset-1 //异步方式发送消息结果：topic-testA|partition-0|offset-14

new ProducerRecord()有多个构造方法，如果有指定分区，那么消息就会被发送到指定分区。如果未指定分区，则会通过key的hash运算，算出消息往哪个分区上发。如果既没有key也没有指定分区（或者key为null），那么消息将会随机发送到一个分区；

关于生产者的ack参数配置在同步发送的前提下，?产者在获得集群返回的ack之前会?直阻塞。那么集群什么时候返回ack呢？

此时ack有3个配置：

ack = 0：kafka-cluster不需要任何的broker收到消息，就?即返回ack给?产者，最容易丢消息的，效率是最?的；ack = 1（默认）：多副本之间的leader已经收到消息，并把消息写?到本地的log中，才会返回ack给?产者，性能和安全性均衡；ack = -1/all：依赖配置min.insync.replicas(默认为1，推荐配置?于等于2)，例如min.insync.replicas=2此时就需要leader和?个follower同步完后，才会返回ack给?产者，这种?式最安全，但性能最差； Kafka的Java客户端-消费者 public class MyConsumer { private final static String TOPIC_NAME = "testA"; private final static String CONSUMER_GROUP_NAME = "console-consumer-9076"; public static void main(String[] args) { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092"); // 消费分组名 props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME); //配置序列化 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName()); //创建一个消费者的客户端 KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props); // 消费者订阅主题列表 consumer.subscribe(Arrays.asList(TOPIC_NAME)); //指定分区消费 //consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0))); //从头消费 //consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,0))); //指定offset消费 //consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10); //poll() API 是拉取消息的?轮询 ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis( 1000 )); for (ConsumerRecord<String, String> record : records) { System.out.printf("收到消息：partition = %d,offset = %d, key =%s, value = %s%n", record.partition(),record.offset(), record.key(), record.value()); } } } //收到消息：partition = 0,offset = 15, key =2, value = 123

关于消费者?动提交和?动提交offset

消费者?论是?动提交还是?动提交，都需要把所属的消费组+消费的某个主题+消费的某个分区及消费的偏移量，这样的信息提交到集群的_consumer_offsets主题??；

?动提交：消费者poll消息下来以后就会?动提交offset

// 是否?动提交offset，默认就是true props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); // ?动提交offset的间隔时间 props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

?动提交：把?动提交的配置改成false

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

?动提交?分成了两种：

1、?动同步提交：在消费完消息后调?同步提交的?法，当集群返回ack前?直阻塞，返回ack后表示提交成功，执?之后的逻辑

//所有的消息已消费完 if (records.count() > 0) {//有消息 // ?动同步提交offset，当前线程会阻塞直到offset提交成功 // ?般使?同步提交，因为提交之后?般也没有什么逻辑代码了 consumer.commitSync();//=======阻塞=== 提交成功 }

2、?动异步提交：在消息消费完后提交，不需要等到集群ack，直接执?之后的逻辑，可以设置?个回调?法，供集群调?

if (records.count() > 0) {//有消息 // ?动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后?的程序逻辑 consumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception exception) { if (exception != null) { System.err.println("Commit failed for " + map); System.err.println("Commit failed exception: " + exception.getStackTrace()); } } }); }

?轮询poll消息

默认情况下，消费者?次会poll500条消息

//?次poll最?拉取消息的条数，可以根据消费速度的快慢来设置 props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); //如果两次poll的时间如果超出了30s的时间间隔，kafka会认为其消费能?过弱，将其踢出消费组。将分区分配给其他消费者 props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000); ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis(1000)); ...后续逻辑

代码中设置了?轮询的时间是1000毫秒，意味着：

如果?次poll到500条，就直接往下执行；如果这?次没有poll到500条且时间在1秒内，那么?轮询继续poll，要么到500 条，要么到1s ，如果多次poll都没达到500条，且1秒时间到了，那么也往下执行‘

如果两次poll的间隔超过30s，集群会认为该消费者的消费能?过弱，该消费者被踢出消费组，触发rebalance机制，rebalance机制会造成性能开销。

消费者的健康状态检查

消费者每隔1s向kafka集群发送?跳，集群发现如果有超过10s没有续约的消费者，将被踢出消费组，触发该消费组的rebalance机制，将该分区交给消费组?的其他消费者进?消费；

//consumer给broker发送?跳的间隔时间 props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000); //kafka如果超过10秒没有收到消费者的?跳，则会把消费者踢出消费组，进?rebalance，把分区分配给其他消费者。 props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

指定时间消费

根据时间，去所有的partition中确定该时间对应的offset，然后去所有的partition中找到该offset之后的消息开始消费

/** *指定消费30分钟前-now 所创建的消息 */ public static void consumerBefore30Min(KafkaConsumer<String, String> consumer){ List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME); //从1?时前开始消费 long fetchDataTime = System.currentTimeMillis() - 1000 * 60 * 30; Map<TopicPartition, Long> map = new HashMap<>(); for (PartitionInfo par : topicPartitions) { map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDataTime); } Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map); for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) { TopicPartition key = entry.getKey(); OffsetAndTimestamp value = entry.getValue(); if (key == null || value == null) { continue; } Long offset = value.offset(); System.out.println("partition-" + key.partition() + "|offset-" + offset); System.out.println(); //根据消费?的timestamp确定offset if (value != null) { consumer.assign(Arrays.asList(key)); consumer.seek(key, offset); } } }

新消费组的消费offset规则

新消费组中的消费者在启动以后，默认会从当前分区的最后?条消息的offset+1开始消费（消费新消息）。可以通过以下的设置，让新的消费者第?次从头开始消费。之后开始消费新消息（最后消费的位置的偏移量+1）

Latest：默认配置，消费新消息earliest：第?次从头开始消费。之后开始消费新消息 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); Springboot中使?Kafka

1、引入依赖

<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>

2、编写配置?件

server: port: 8081 spring: kafka: bootstrap-servers: 127.0.0.1:9092 producer: retries: 3 batch-size: 16384 buffer-memory: 33554432 acks: 1 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer consumer: group-id: default-group enable-auto-commit: false auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer max-poll-records: 500 listener: # 当每?条记录被消费者监听器（ListenerConsumer）处理之后提交 # RECORD # 当每?批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交 # BATCH # 当每?批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间?于TIME时提交 # TIME # 当每?批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量?于等于COUNT时提交 # COUNT # TIME | COUNT　有?个条件满?时提交 # COUNT_TIME # 当每?批poll()的数据被消费者监听器（ListenerConsumer）处理之后, ?动调?Acknowledgment.acknowledge()后提交 # MANUAL # ?动调?Acknowledgment.acknowledge()后?即提交，?般使?这种 # MANUAL_IMMEDIATE ack-mode: MANUAL_IMMEDIATE

3、编写消息?产者

@RestController @RequestMapping("/msg") public class MyKafkaController { private final static String TOPIC_NAME = "testA"; @Autowired private KafkaTemplate<String,String> kafkaTemplate; @RequestMapping("/send") public String sendMessage(){ kafkaTemplate.send(TOPIC_NAME,0,"key","this is a message!"); return "send success!"; } }

4、编写消费者

@Component public class MyConsumer { @KafkaListener(topics = "testA") /* @KafkaListener(groupId = "testGroup", topicPartitions = { @TopicPartition(topic = "topic1", partitions = {"0", "1"}), @TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100")) },concurrency = "3")//concurrency就是同组下的消费者个数，就是并发消费数，建议?于等于分区总数*/ public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) { String value = record.value(); System.out.println(value); System.out.println(record); //?动提交offset ack.acknowledge(); } }

启动项目，执行http://localhost:8081/msg/send，控制台打印：

this is a message! ConsumerRecord(topic = testA, partition = 0, leaderEpoch = 0, offset = 17, CreateTime = 1644403908306, serialized key size = 3, serialized value size = 18, headers = RecordHeaders(headers = [], isReadOnly = false), key = key, value = this is a message!) Kafka集群中的controller、rebalance、HW

1、controller

每个broker启动时会向zk创建?个临时序号节点，获得的序号最?的那个broker将会作为集群中的controller，负责这么?件事：

当集群中有?个副本的leader挂掉，需要在集群中选举出?个新的leader，选举的规则是从isr集合中最左边获得。当集群中有broker新增或减少，controller会同步信息给其他broker当集群中有分区新增或减少，controller会同步信息给其他broker

2、rebalance（重平衡）机制

前提是：消费者没有指明分区消费。当消费组里消费者和分区的关系发生变化，那么就会触发rebalance机制，这个机制会重新调整消费者消费哪个分区。

在触发rebalance机制之前，消费者消费哪个分区有三种策略：

range：通过公示来计算某个消费者消费哪个分区轮询：大家轮着消费sticky：在触发了rebalance后，在消费者消费的原分区不变的基础上进行调整。

range与轮询会将现有的消费关系全部去除并且重新分配，对性能肯定会有所影响！

3、HW（High WaterMark 高水位）和LEO

LEO是某个副本最后消息的消息位置（log-end-offset）

HW是已完成同步的位置。消息在写?broker时，且每个broker完成这条消息的同步后，hw才会变化。在这之前消费者是消费不到这条消息的。在同步完成之后，HW更新之后，消费者才能消费到这条消息，这样的?的是防?消息的丢失；

Kafka相关问题优化