kafka: 基础概念回顾（生产者客户端和机架感知相关内容）

本文主要是介绍kafka: 基础概念回顾（生产者客户端和机架感知相关内容），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、kafka生产者客户端

1、整体架构：数据发送流程

在这里插入图片描述
（1）生产者

拦截器
生产者的拦截器可以在消息发送前做一些拦截工作对数据进行相应的处理，比如：消息过滤、消息内容修改等。

package org.apache.kafka.clients.producer;
import org.apache.kafka.common.Configurable;
public interface ProducerInterceptor<K, V> extends Configurable {//在将消息序列化和计算分区之前会调⽤该⽅法，⽤来对消息进⾏相应的定制化操作，如修改消息内容public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);//在消息被应答之前或者消息发送失败时调⽤该⽅法，优先于⽤⼾设定的Callback之前执⾏，如统计消息发送成功或失败的次数public void onAcknowledgement(RecordMetadata metadata, Exception exception);public void close();
}

序列化器
分区器

二、kafka数据可靠性保证

1、LEO和HW

2、工作流程

3、Leader Epoch

三、粘性分区策略

四、机架感知

1、概念

2、机架感知分区分配策略

3、验证

（1）验证目标

机架感知特性将同⼀分区的副本分散到不同的机架上
rack机制消费者可以消费到follower副本中的数据

（2）参数配置
broker端配置：

配置名：broker.rack=my-rack-id
- 解释：broker属于的rack
配置名：replica.selector.class
- 解释：ReplicaSelector实现类的全名，包括路径 (⽐如 RackAwareReplicaSelector 即按 rack id 指定消费)

Client端配置：
client.rack

consumer端配置
配置名：client.rack
解释：这个参数需要和broker端指定的 broker.rack 相同，表⽰去哪个rack中获取数据。
默认：null

（3）环境准备：kafka集群

kafka实例数: 4
两个kafka实例broker.rack配置为0，另外两个kafka实例broker.rack配置为了2，broker端配置如下：

server1：
broker.id=0broker.rack=0
replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelectorserver2：
broker.id=1
broker.rack=0
replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelectorserver3
broker.id=2
broker.rack=2
replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelectorserver4
broker.id=3
broker.rack=2
replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelector

启动kafka集群，服务端⽇志信息：
在这里插入图片描述

验证一：机架感知特性将同一分区的副本分散到不同的机架上

创建topic rack02，副本被分配到了broker1和2

创建topic rack03 副本被分配到了0和3

验证二：客⼾端（消费者）验证：rack机制消费者可以消费到follower副本中的数据

验证代码如下：

package person.xsc.train.producer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import person.xsc.train.client.KafkaConsumerClient;
import person.xsc.train.constant.KafkaConstant;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
public class Demo {public static KafkaConsumer<String, String> kafkaConsumer;public static void main(String[] args) {Properties properties = new Properties();properties.put(KafkaConstant.BOOTSTRAP_SERVERS, "localhost:9093,localhosproperties.put(KafkaConstant.GROUP_ID, "test01");properties.put(KafkaConstant.ENABLE_AUTO_COMMIT, "true");properties.put(KafkaConstant.AUTO_COMMIT_INTERVAL_MS, "1000");properties.put(KafkaConstant.KEY_DESERIALIZER, StringDeserializer.class.properties.put(KafkaConstant.VALUE_DESERIALIZER, StringDeserializer.clasproperties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");properties.put(ConsumerConfig.CLIENT_RACK_CONFIG, "0");properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");kafkaConsumer = KafkaConsumerClient.createKafkaClient(properties);receiveMessage("rack02");}public static void receiveMessage(String topic) {TopicPartition topicPartition0 = new TopicPartition(topic, 0);kafkaConsumer.assign(Arrays.asList(topicPartition0));while(true) {// Kafka的消费者⼀次拉取⼀批的数据ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll//System.out.println("开始打印消息！");// 5.将将记录（record）的offset、key、value都打印出来for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {// 主题String topicName = consumerRecord.topic();int partition = consumerRecord.partition();// offset：这条消息处于Kafka分区中的哪个位置long offset = consumerRecord.offset();// key\valueString key = consumerRecord.key();String value = consumerRecord.value();System.out.println(String.format("topic: %s, partition: %s, offs}}}
}

前置背景：
Topic rack02的partition 0分区的副本为broker2（对应的rack为2）和broker1（对应的rack为0），其中broker2为leader（在⾮rack机制下仅能消费到leader中的数据）。

在上述代码中，消费者配置中限制了rack为0，消费的分区为0，因此映射到broker1。通过测试可验证在rack机制下消费者可以消费到folloer副本中的数据，测试如下：
在这里插入图片描述