kafka第二次课！！！

Posted 2022-09-09 teayear

1，课程回顾
2，本章重点
kafka的整体工作流程
消息生产者写入消息过程
消息消费者消费要点
kafka的Java api

3，具体内容
3.1 kafka的整体工作流程
图片: https://uploader.shimo.im/f/i8WYA8tIRBMw7sAl.png

   Kafka中消息是以topic进行分类的，生产者生产消息，消费者消费消息，都是面向topic的。
topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，且每条数据都有自己的offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，以便出错恢复时，从上次的位置继续消费。

图片: https://uploader.shimo.im/f/gQv80XXFOZbytjlz.png

由于生产者生产的消息会不断追加到log文件末尾，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment对应两个文件——“.index”文件和“.log”文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。例如，first这个topic有三个分区，则其对应的文件夹为first-0,first-1,first-2。
00000000000000000000.index
00000000000000000000.log
00000000000000170410.index
00000000000000170410.log
00000000000000239430.index
00000000000000239430.log

3.1.1 分区的好处
1，方便集群的伸缩
每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了
2，可以提高并发
可以以Partition为单位读写，提高集群的读写速度
3.1.2 分区是如何分配到broker上
目的：

1. 保证所有的分区以及副本可以均衡在分布上所有的broker上
2. 保证同一个分区及其副本尽量不要分布在同一个broker上

让分区和broker进行排序 partition leader排序： p0 p1 p2 p3
partition 副本： p0r p1r p2r p3r
broker排序： br0 br1 br2
分区怎么分配到broker上： p0随机一个broker topic越多，每个topic都对应有一个p0 越平均的分配到不同的broker
leader p0->br1 p1->br2 p2 ->br0 p3->br1（p0随机，以后的放入下一个）
副本随机： p0r->br0 p1r->br1 p2r->br1 p3r->br2

3.1.3 副本（replication）好处
提高kafka的系统的可靠性和稳定性，同一个partitation对应一个或者多个副本，创建topic时就可以设置（–replication-factor 2）。没有副本，一旦当前保存消息的服务器宕机，就会造成消息丢失，如果有replication，当保存消息的服务器宕机后，从新选举新的leader,继续进行消息读写，不会造成消息丢失。
3.1.4 zk保存kafka数据的目录结构是什么样子的，在kafka集群中的作用
图片: https://uploader.shimo.im/f/etjqp4EUxDXOCZoG.png

1，broker在zk中注册：集群启动时，每个broker都会在/brokers/ids/下注册(创临时节点)，如果broker挂掉了，zk就会删除该节点。
2，topick会在zk中注册：创建topic时，每个topic都会在/brokers/topics/下注册,topic删除，节点失效。每个broker和topic的对应关系也是由zk进行维护。
3，consumer(消费者)在zk注册：当新的消费者都会zk进行注册，zk在/consumers/consumer-group/ 创建3个节点 ids offsets(偏移量) owners
ids: 记录当前消费者组所有的消费者id
offsets:消费者在消费topic每个partition时，消费到哪个位置（offset 偏移量）
owners:记录该消费者组消费的topic信息（订阅了哪些topic）
新版本无效
3.2 生产者如何写入消息的

图片: https://uploader.shimo.im/f/VXgxcIgPyyHzOlZu.png

       get  /brokers/topics/tpa/partitions/0/state     
"controller_epoch":7,"leader":2,"version":1,"leader_epoch":19,"isr":[2]   
  ack=-1/all  讲解画图实例
为保证producer发送的数据，能可靠的发送到指定的topic，topic的每个partition收到producer发送的数据后，都需要向producer发送ack（acknowledgement确认收到），如果producer收到ack，就会进行下一轮的发送，否则重新发送数据。

3.2.1消息写入时，放入分区的规则：
1),指定分区，直接按照指定分区写入
2,没有指定分区，但是消息中含有key(一般消息是key value的方式)，通过key的值进行hash运算，计算得到一个partition，写入到这个分区。（aaa hash运算后可能得到一个和aaa没有任何关系的一个数值123132，对分区的总数量取模，根据结果，得到分区）
3, 如果没有指定分区，key都没有，使用轮询(第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition 值，也就是常说的 round-robin 算法)，找出一个分区，并写入。
3.2.2 ack应答机制
0 producer不等待broker同步完成的确认，继续发送下一条(批)信息
提供了最低的延迟。但是最弱的持久性，当服务器发生故障时，就很可能发生数据丢失。例如leader已经死亡，producer不知情，还会继续发送消息broker接收不到数据就会数据丢失
1 producer要等待leader成功收到数据并得到确认，才发送下一条message。此选项提供了较好的持久性较低的延迟性。Partition的Leader死亡，follwer尚未复制，数据就会丢失
-1/all 意味着producer得到leader和所有follwer确认，才发送下一条数据
3.2.3 ISR(In Sync Replicas):
当ack配置-1时 leader收到数据，所有follower都开始同步数据，但有一个follower，因为某种故障，迟迟不能与leader进行同步，那leader就要一直等下去，直到它完成同步，才能发送ack。这个问题怎么解决呢？
Leader维护了一个动态的in-sync replica set (ISR)，意为和leader保持同步的follower集合。当ISR中的follower完成数据的同步之后，leader就会给follower发送ack。如果follower长时间未向leader同步数据，则该follower将被踢出ISR，该时间阈值由replica.log.time.max.ms参数设定。Leader发生故障之后，就会从ISR中选举新的leader。
3.3 消费者消费要点
消息的消费模式有两种：推送模式(push) 和拉取模式(pull)
kafka采取的拉取模式(pull),由消费者自己记录消费状态（消费者自己记录自己的消费位置（offset偏移量））,每个消费者相互独立的消费每个一个分区的消息。每个消费者消费完了后，对消息本身不做任何处理，决定该消息是否能被删除，跟消费者没有任何关系，与配置的消息过期时间和消息总容量的大小配置参数有关（log.retention.hours=168 log.retention.bytes=1G）。
消费者是以消费者组( consumer group)的方式，由一个或者多个消费者组成一个组，共 同消费一个topic(主题),在同一时刻，只能由同一个组的一个消费者去消费同一个分区。
pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。
3.4 kafka java api
3.4.1 创建项目，引入jar包

junit
junit
4.11
test

org.apache.kafka kafka-clients 2.7.0 org.apache.kafka kafka_2.12 2.7.0 3.4.2 编写kafka工具类，对topic的一系列操作 kafka.properties： #参考官网配置http://kafka.apachecn.org/documentation.html#configuration #这是一个用于建立初始连接到kafka集群的"主机/端口对"配置列表 bootstrap.servers=192.168.23.21:9092,192.168.23.22:9092,192.168.23.23:9092 #发送应答(0：收到后直接响应成功，1：leader保存则响应成功，-1/all:leader和isr列表保存后才响应成功) acks=all #失败重试次数 retries=0 #批量发送大小（默认：16384，16K） batch.size=16384 #发送延迟时间（默认：0） linger.ms=1 加载配置文件的工具类编写： package com.aaa.kafka.util; import java.io.IOException; import java.io.InputStream; import java.util.Map; import java.util.Properties; import java.util.Set; /** * @ fileName:LoadConfigurationUtil * @ description:通用加载配置工具类 * @ author:zhz * @ createTime:2022/1/11 10:17 * @ version:1.0.0 */ public class LoadConfigurationUtil /** * 通用加载properties文件方法 * @param path * @return */ public static Properties loadProperty(String path) Properties properties =new Properties(); InputStream inputStream = LoadConfigurationUtil.class.getResourceAsStream("/"+path); try properties.load(inputStream); catch (IOException e) e.printStackTrace(); return properties; /* public static void main(String[] args) Properties properties = loadProperty("kafka.properties"); //遍历Map集合，尽量使用这种方法 Set

KafkaJavaUtil：
一定要关闭AdminClient 否则创建不成功
package com.aaa.kafka.util;
import org.apache.kafka.clients.admin.;
import org.apache.kafka.common.KafkaFuture;
import java.util.;
import java.util.concurrent.ExecutionException;
/**

• @ fileName:KafkaJavaUtil
• @ description:kafka通用工具类
• @ author:zhz
• @ createTime:2022/1/11 10:24
• @ version:1.0.0
  /
  public class KafkaJavaUtil
  //多个方法公共，写为类属性
  private AdminClient adminClient;
  /*
  • 使用空构造，初始化AdminClient
    /
    public KafkaJavaUtil()
    //加载配置
    Properties properties = LoadConfigurationUtil.loadProperty(“kafka.properties”);
    //使用配置文件初始化对象
    adminClient = KafkaAdminClient.create(properties);
    
    /*
  • 获取topic列表
  • @return
    /
    public List getTopicList()
    try
    //列表选项
    ListTopicsOptions listTopicsOptions =new ListTopicsOptions();
    //设置是否显示kafka内部使用的topic //listInternal(true) __consumer_offsets 会显示 否则不会
    listTopicsOptions.listInternal(false);
    //返回topic列表
    ListTopicsResult listTopicsResult = adminClient.listTopics(listTopicsOptions);
    //名称集合对象
    KafkaFuture<Set> nameList = listTopicsResult.names();
    //获取到set集合
    Set setString = nameList.get();
    //定义List
    List topicNameList = new ArrayList<>(setString);
    return topicNameList;
    catch (InterruptedException e)
    e.printStackTrace();
    catch (ExecutionException e)
    e.printStackTrace();
    finally
    if(adminClient!=null)
    adminClient.close();
    
    
    return null;
    
    /*
  • 创建topic
  • @param topicName 名称
  • @param partitions 分区数量
  • @param replicationFactor 副本因子
  • @return
    /
    public boolean createTopic(String topicName,int partitions,short replicationFactor)
    try
    NewTopic topic =new NewTopic(topicName,partitions,replicationFactor);
    adminClient.createTopics(Arrays.asList(topic));
    //无异常创建成功 返回true
    return true;
    catch (Exception e)
    e.printStackTrace();
    finally
    if(adminClient!=null)
    adminClient.close();
    
    
    return false;
    
    /*
  • 删除topic
  • @param topicName
  • @return
    */
    public boolean deleteTopic(String topicName)
    try
    adminClient.deleteTopics(Arrays.asList(topicName));
    return true;
    catch (Exception e)
    e.printStackTrace();
    finally
    if(adminClient!=null)
    adminClient.close();
    
    
    return false;
    
    

KafkaJavaUtilTest：
package com.aaa.kafka.test;
import com.aaa.kafka.util.KafkaJavaUtil;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.util.List;
import static org.junit.Assert.*;
public class KafkaJavaUtilTest
private KafkaJavaUtil kafkaJavaUtil;
@Before
public void setUp() throws Exception
kafkaJavaUtil = new KafkaJavaUtil();

@Test
public void getTopicList() 
    List<String> topicList = kafkaJavaUtil.getTopicList();
    System.out.println(topicList);

@Test
public void createTopic() 
    boolean isSuc = kafkaJavaUtil.createTopic("tpaa", 3, Short.valueOf("2"));
    if (isSuc)
        System.out.println("成功");
    else 
        System.out.println("失败");
    

@Test
public void deleteTopic() 
    boolean isSuc = kafkaJavaUtil.deleteTopic("tpaa");
    if (isSuc)
        System.out.println("成功");
    else 
        System.out.println("失败");
    

@After
public void tearDown() throws Exception 

测试：
先启动 zk, 再启动kafka集群，然后测试
zkServer.sh start
jps
在allsession操作：
cd /usr/kafka
kafka-server-start.sh -daemon config/server.properties
jps
3.4.3 编写生产者，写入消息（带回调的写入，测试消息写入时，放入的分区规则）
生产者配置文件producer.properties：
#建立初始连接到kafka集群的"主机/端口对"配置列表
bootstrap.servers=cluster1:9092,cluster2:9092,cluster3:9092
#序列化
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
#此配置是 Producer 在确认一个请求发送完成之前需要收到的反馈信息的数量。
#ack =acknowledge反馈信息 值=0 1 -1 分别代表什么意思
#acks=0 如果设置为0，则 producer 不会等待服务器的反馈。该消息会被立刻添加到 socket buffer 中并认为已经发送完成。
#acks=1 如果设置为1，leader节点会将记录写入本地日志，并且在所有 follower 节点反馈之前就先确认成功。
#acks=all acks=-1与acks=all是等效的。 如果设置为all，这就意味着 leader 节点会等待所有同步中的副本(ISR)确认之后再确认这条记录是否发送完成。
acks=all
#若设置大于0的值，则客户端会将发送失败的记录重新发送，尽管这些记录有可能是暂时性的错误。
retries=3
#当将多个记录被发送到同一个分区时， Producer 将尝试将记录组合到更少的请求中。
batch.size=323840
#控制消息发送延时行为的，该参数默认值是0。表示消息需要被立即发送，无须关系batch是否被填满。
linger.ms=10
#Producer 用来缓冲等待被发送到服务器的记录的总字节数。
buffer.memory=33554432
#该配置控制 ProducerKafka.send()和KafkaProducer.partitionsFor() 允许被阻塞的时长。这些方法可能因为缓冲区满了或者元数据不可用而被阻塞。
max.block.ms=3000
#自定义指定消费topic名称
topicName=tp88
生产者代码类：
package com.aaa.kafka.demo;
import com.aaa.kafka.util.LoadConfigurationUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**

• @ fileName:ProducerKafka

• @ description:

• @ author:zhz

• @ createTime:2022/1/12 10:10

• @ version:1.0.0
  */
  public class ProducerKafka
  public static void main(String[] args)
  KafkaProducer kafkaProducer = null;
  try
  //加载配置文件并生成Properties
  Properties properties = LoadConfigurationUtil.loadProperty(“producer.properties”);
  //实例化KafkaProducer
  kafkaProducer = new KafkaProducer(properties);
  for (int i = 0; i < 100; i++)
  System.out.println(“发送了信息hello”+i);
  ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),“hello”+i);
  //发送消息
  kafkaProducer.send(producerRecord);
  //每个200毫秒发送一次
  Thread.sleep(200);
  
  System.out.println(“消息发送完毕！！！”);
  catch (Exception e)
  e.printStackTrace();
  finally
  if(kafkaProducer!=null)
  kafkaProducer.close();
  
  

  
  

带回调的生产者代码类
package com.aaa.kafka.demo;
import com.aaa.kafka.util.LoadConfigurationUtil;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Properties;
/**

• @ fileName:ProducerKafka

• @ description:

• @ author:zhz

• @ createTime:2022/1/12 10:10

• @ version:1.0.0
  */
  public class ProducerKafkaWithCallback
  public static void main(String[] args)
  KafkaProducer kafkaProducer = null;
  try
  //加载配置文件并生成Properties
  Properties properties = LoadConfigurationUtil.loadProperty(“producer.properties”);
  //实例化KafkaProducer
  kafkaProducer = new KafkaProducer(properties);
  for (int i = 0; i < 100; i++)
  System.out.println(“发送了信息hi,AAA”+i+“发送后的信息为：”);
  //3，不指定分区，也没有key
  //ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),“hello,qy141”+i);
  //2,不指定分区，但是有key
  ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),“AAA”+i,“hi,AAA”+i);
  //1,指定分区,跟key和value无关
  //ProducerRecord producerRecord =new ProducerRecord(properties.getProperty(“topicName”),2,“key”+i,“hi”+i);
  //发送消息,并带回调函数（根据回调函数可以看到该消息被存放分区和偏移量等信息）
  kafkaProducer.send(producerRecord, new Callback()
  @Override
  public void onCompletion(RecordMetadata recordMetadata, Exception e)
  System.out.println(“-------该消息的分区为：”+recordMetadata.partition()+“,偏移量为：”+recordMetadata.offset());
  
  );
  //每个200毫秒发送一次
  Thread.sleep(200);
  
  System.out.println(“消息发送完毕！！！”);
  catch (Exception e)
  e.printStackTrace();
  finally
  if(kafkaProducer!=null)
  kafkaProducer.close();
  
  

  
  

测试前需要在本地hosts配置文件加入：
C:\\Windows\\System32\\drivers\\etc\\hosts 添加下面配置
192.168.170.41 cluster1
192.168.170.42 cluster2
192.168.170.43 cluster3
3.4.4 编写消费者，消费消息（测试同组和不同组消费者的消费情况）
消费者配置文件consumer.properties：
#建立初始连接到kafka集群的"主机/端口对"配置列表
bootstrap.servers=cluster1:9092,cluster2:9092,cluster3:9092
#反序列化 消费者
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
#消费者组ID 测试是否是同组
group.id=groupThree
#如果为真，则用户的偏移量将在后台定期提交。
enable.auto.commit=true
#使用者偏移自动提交到Kafka的频率（毫秒）
auto.commit.interval.ms=1000
#当kafka中没有初始偏移或服务器上不再存在当前偏移量
#earliest:自动将偏移重置为最早偏移
#latest:自动将偏移重置为最新偏移

none:如果未找到使用者组的先前偏移量，则向使用者引发异常

#anything else: throw exception to the consumer.直接给消费者抛出异常
auto.offset.reset=earliest
#自定义指定消费topic名称
topicName=tp88
消费者类：
package com.aaa.kafka.demo;
import com.aaa.kafka.util.LoadConfigurationUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.time.temporal.TemporalUnit;
import java.util.Arrays;
import java.util.Properties;
/**

• @ fileName:ConsumerKafka

• @ description:消息消费者

• @ author:zhz

• @ createTime:2022/1/12 9:30

• @ version:1.0.0
  */
  public class ConsumerKafka
  public static void main(String[] args)
  KafkaConsumer kafkaConsumer = null;
  try
  //加载配置文件并生成Properties
  Properties properties = LoadConfigurationUtil.loadProperty(“consumer.properties”);
  //properties.put(“bootstrap.servers”,“cluster1:9092,cluster2:9092,cluster3:9092”);
  //实例化kafkaConsumer
  kafkaConsumer = new KafkaConsumer(properties);
  //指定当前消费者订阅的topic列表 可以是多个
  kafkaConsumer.subscribe(Arrays.asList(properties.getProperty(“topicName”)));
  //添加提示
  System.out.println(“消费者已经启动，等待生产者生产消息。。。。。。”);
  //编写循环接受消息
  while (true)
  //如果说topic中暂时没有生成者生产消息，就会隔间500毫秒拉取一次
  ConsumerRecords<String,String> consumerRecords = kafkaConsumer.poll(500);
  //循环拉取集合结果
  for (ConsumerRecord<String, String> consumerRecord : consumerRecords)
  System.out.println(“消息的key：”+consumerRecord.key()+“，value:”+consumerRecord.value()
  +“，offset偏移量：”+consumerRecord.offset()+“,partition消息所在分区：”+consumerRecord.partition());
  
  
  catch (Exception e)
  e.printStackTrace();
  finally
  //关闭资源
  if(kafkaConsumer!=null)
  kafkaConsumer.close();
  
  

  
  

  测试同组和不同组消费者的消费情况
  前1个控制台自成一组groupOne 2,3,4同组groupTwo 5,6同组groupThree

命令窗口，组：test-consumer-group
kafka-console-consumer.sh --bootstrap-server kafka1:9092 --from-beginning --topic myTopicB

4，知识点总结
5，本章面试题
kafka 如何减少数据丢失
https://www.cnblogs.com/huxi2b/p/6056364.html
Producer端：
#在Producer端用来存放尚未发送出去的Message的缓冲区大小。缓冲区满了之后可以选择阻塞发送或抛出异常，由block.on.buffer.full的配置来决定。
block.on.buffer.full = true
acks = all
#无限重试，直到你意识到出现了问题
retries = MAX_VALUE
#限制客户端在单个连接上能够发送的未响应请求的个数。设置此值是1表示kafka broker在响应请求之前client不能再向同一个broker发送请求。
max.in.flight.requests.per.connection = 1
#而不是send(record)方法自定义回调逻辑处理消息发送失败
使用KafkaProducer.send(record, callback)
#设置此参数是为了避免消息乱序
callback逻辑中显式关闭producer：close(0)
#关闭unclean leader选举，即不允许非ISR中的副本被选举为leader，以避免数据丢失
unclean.leader.election.enable=false
replication.factor = 3
#消息至少要被写入到这么多副本才算成功，也是提升数据持久性的一个参数。与acks配合使用
min.insync.replicas = 2
replication.factor > min.insync.replicas
Consumer端：
consumer端丢失消息的情形比较简单：如果在消息处理完成前就提交了offset，那么就有可能造成数据的丢失。由于Kafka consumer默认是自动提交位移的，所以在后台提交位移前一定要保证消息被正常处理了，因此不建议采用很重的处理逻辑，如果处理耗时很长，则建议把逻辑放到另一个线程中去做。为了避免数据丢失，现给出两点建议：
enable.auto.commit=false
消息处理完成之后再提交位移

消费者故障，出现活锁问题如何解决？
kafka 消费者正常情况下是订阅一个 topic 并且能够 poll 消息。该消费者会占用一个分区，同时需要定时向 zk 发送心跳监测，以证明自己活着。当消费者占用一个分区后，且能够正常发送心跳，但是不 poll 消息了，不再进行消息处理了，这种情况下就出现了活锁。
kafka 这边处理的时候会配置 max.poll.interval.ms 活跃监测机制。如果客户端调用 poll 的频率大于最大间隔，就会将当前客户端连接断开，让其它的消费者过来消费。

kafka 如何不消费重复数据？
  1，幂等操作，重复消费不会产生问题
  2，对每个partitionID，产生一个uniqueID,.只有这个partition的数据被完全消费，才算成功，否则失败回滚。下次若重复执行，就skip