小白入门kafka-附源码

现在消息队列的主流 kafaka面试过程中经常被提到，所以进行了学习和总结现在分享给大家，如有疑问可以评论沟通。 

基础API使用源码和spirngboot+kafka集成源码：https://files.cnblogs.com/files/liyanbofly/kafaka_basic_SpringbootPractice.rar?t=1667259540

                                kafka基础API使用

一、kafka和zookeeper启动说明
  1.1 zookeeper启动 ：本机按装目录zookeeper 在/usr/local/zookeeperxxx      ./zkServer.sh start
  1.2 kafka 启动：本机安装目录 /usr/kafka_2xxx      ./kafka-server-start.sh -daemon  ../config/server.properties

 二、topic 创建说明
 在使用kafka发送消息和消费消息之前，必须先要创建topic，在kafka中创建topic的方式有以下3种：
 1、如果kafka broker中的config/server.properties配置文件中配置了auto.create.topics.enable参数为true（默认值就是true），那么当生产者向一个尚未创建的topic发送消息时，
    会自动创建一个num.partitions（默认值为1）个分区和default.replication.factor（默认值为1）个副本的对应topic。不过我们一般不建议将auto.create.topics.enable参数设置为true，因为这个参数会影响topic的管理与维护。
    [allow.auto.create.topics = true]该参数自动创建topic,当生产者发送ProducerRecord 时服务没有该topic时
 2、通过kafka提供的kafka-topics.sh脚本来创建，并且我们也建议通过这种方式（或者相关的变种方式）来创建topic。
 3、kafka的0.10版本之前，可以使用内置的kafka-admin包，后续提供了专门的类AdminClient API来进行API层面上的topic创建。

三、如属性说明
  3.1 partions 分区
   创建topic时会指定partion 分区
   如： ./bin/kafka-topics.sh --bootstrap-server CentOS:9092 --create --topic topic01 --partitions 2  --replication-factor 1
    1） 创建的partion 会平分给每个组的消费者 如 4个分区，两个消费者那每个消费者会收到两个分区的信息。
    2） 生产者生产发送消息时会根据Key hash分配给某个分区，也可以指定一个分区;
    3）如果没有key会根据轮询方式给每个分区;
    4） #自动创建主题 auto.create.topics.enable=true # 默认主题的分区数 num.partitions=8 # 默认分区副本 default.replication.factor=3

      Record的持久化时间是通过配置文件指定,默认是168小时。log.retention.hours=168

  3.2 offset 偏移量
    Kafka消费者默认对于未订阅的topic的offset的时候，也就是系统并没有存储该消费者的消费分区的记录信息，
    默认Kafka消费者的默认首次消费策略：latest，auto.offset.reset=latest
    earliest - 自动将偏移量重置为最早的偏移量
    latest - 自动将偏移量重置为最新的偏移量
    none - 如果未找到消费者组的先前偏移量，则向消费者抛出异常
    3.2.1 自动设置偏移量
         * 设置自动提交偏移量
         *   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);【enable.auto.commit=true】
         *   properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,500000);[auto.commit.interval.ms=500000]
         *   如果时间间隔较长，当消费者在提交前异常停止，再重新启动时会重复消费未提交偏移量的数据
    3.2.2 手动设置偏移量
       properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false); 【enable.auto.commit=false】
       可以对每个已经消费的记录进行 立刻提交偏移量使记录很少被重复消费
       提交offset需要注意：
                Map<TopicPartition, OffsetAndMetadata> offsets=new HashMap<TopicPartition, OffsetAndMetadata>();
                // 提交的偏移量需要 offset+1，因为提交的偏移量是下一次读取的位为了避免重复启动后消费需要 offset+1
                offsets.put(new TopicPartition(consumerRecord.topic(),partition),new OffsetAndMetadata(offset+1));

  3.3 ack 应答确认机制
     3.3.1 如果生产者在规定的时间内，并没有得到Kafka的Leader的Ack应答，Kafka可以开启reties机制。
      request.timeout.ms = 30000  默认
      retries = 2147483647 默认
     3.3.2  有哪几种应答机制 acks=1 默认
     acks=1 - Leader会将Record写到其本地日志中，但会在不等待所有Follower的完全确认的情况下做出响应。在这种情况下，如果Leader在确认记录后立即失败，但在Follower复制记录之前失败，则记录将丢失。
     acks=0 - 生产者根本不会等待服务器的任何确认。该记录将立即添加到套接字缓冲区中并视为已发送。在这种情况下，不能保证服务器已收到记录。
     acks=all - 这意味着Leader将等待全套同步副本确认记录。这保证了只要至少一个同步副本仍处于活动状态，记录就不会丢失。这是最有力的保证。这等效于acks = -1设置。
   在生者代码位置配置
         // 应达机制 ---超时后会被重试发3次，3次后将不在发送
           props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,1);
           props.put(ProducerConfig.ACKS_CONFIG,"-1");
           props.put(ProducerConfig.RETRIES_CONFIG,3);// 重试次数不包括第一次

           我就抛砖引玉聊一聊第一条吧：Kafka 消费端自己虽然没有重试队列这个概念，
           但是其实在客户端上构建异常消息逻辑并不是非常困难的，举例来说在Kafka Streams客户端就有不少选择，
           比如下面这个条目：https://docs.confluent.io/current/streams/faq.html#failure-and-exception-handling （很遗憾没有中文翻译...）。
           总的来说在考虑重试之前先要考虑什么样的异常应该重试，什么样的不应该，
           1、比如：如果一条消息本身有报文格式异常。
           2、如果消费之后的处理逻辑出错（最简单的例子，比如说 divide by 0）。
           3、如果只是字符串验证出错，比如可能传输过程中有几个字符corrupt。那么只有第三种情况值得重试，而其余两种不应该重试而是应该直接向监测系统报错，
           之于是否关闭整个消费客户端就看是否有顺序要求了，等等。总而言之，Kafka 需要用户自行根据异常条件来进行异常处理，
           而不会大包大揽提供out of the box的异常机制比如重试队列。
           链接：https://www.zhihu.com/question/307757429/answer/1212963174


   3.4   幂等性
      Kafka在0.11.0.0版本支持增加了对幂等的支持。幂等是针对生产者角度的特性。幂等可以保证上生产者发送的消息，不会丢失，
         而且不会重复。实现幂等的关键点就是服务端可以区分请求是否重复，过滤掉重复的请求。要区分请求是否重复的有两点：
      幂等又称为exactly once。要停止多次处理消息，必须仅将其持久化到Kafka Topic中仅仅一次。在初始化期间，kafka会给生产者生成一个唯一的ID称为Producer ID或PID。

      PID和序列号与消息捆绑在一起，然后发送给Broker。由于序列号从零开始并且单调递增，因此，仅当消息的序列号比该PID / TopicPartition对中最后提交的消息正好大1时
        ，Broker才会接受该消息。如果不是这种情况，则Broker认定是生产者重新发送该消息。
      enable.idempotence= false 默认
      注意:在使用幂等性的时候，要求必须开启retries=true和acks=all

  3.3 kafka 可以开启事务处理

  3.4 kafka 为什么这么快
  磁盘顺序读写、稀疏索引、批量文件压缩、零拷贝技术
  磁盘分为：内核空间和用户空间 ，可以不经过用户空间发送到网卡

4、集成springboot 遇到的问题 application.yml 文件consumer
需要这样（通过看报错源码）
报错信息：kafka Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'
 key:
    deserializer: org.apache.kafka.common.serialization.StringSerializer
 value:
    deserializer: org.apache.kafka.common.serialization.StringDeserializer
不应该是
 key-deserializer: org.apache.kafka.common.serialization.StringSerializer
 value-deserializer: org.apache.kafka.common.serialization.StringDeserializer


 四、遇到问题
1、org.apache.kafka.common.errors.TimeoutException: Topic topic03 not present in metadata after 60000 ms.
   尝试按问题2 解决

2、  kafka Connection to node -1 (CentOS1/192.168.88.129:9092) could not be established. Broker may not b
    解决:
    关闭防火墙
    如果是winodws 修改本机hosts   192.168.88.129 CentOS
    使与 kafaka主机  etc/sysconfig/network  中HOSTNAME=CentOS 一致
    使与 vi /etc/hosts   192.168.88.129  CentOS 一致
    使与 kafkaxxx/config/server.properties 中 listeners=PLAINTEXT://CentOS:9092
    相关网址：https://www.yisu.com/zixun/160374.html