@TOCKafka学习笔记
1. 基本概念：Producer：消息生产者，向Kafka中发布消息的角色。 Consumer：消息消费者，即从Kafka中拉取消息消费的客户端。 Consumer Group：消费者组，消费者组则是一组中存在多个消费者，消费者消费Broker中当前Topic的不同分区中的消息，消费者组之间互不影响，所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。某一个分区中的消息只能够一个消费者组中的一个消费者所消费 Broker：经纪人，一台Kafka服务器就是一个Broker，一个集群由多个Broker组成，一个Broker可以容纳多个Topic。 Topic：主题，可以理解为一个队列，生产者和消费者都是面向一个Topic Partition：分区，为了实现扩展性，一个非常大的Topic可以分布到多个Broker上，一个Topic可以分为多个Partition，每个Partition是一个有序的队列(分区有序，不能保证全局有序) Replica：副本Replication，为保证集群中某个节点发生故障，节点上的Partition数据不丢失，Kafka可以正常的工作，Kafka提供了副本机制，一个Topic的每个分区有若干个副本，一个Leader和多个Follower Leader：每个分区多个副本的主角色，生产者发送数据的对象，以及消费者消费数据的对象都是Leader。 Follower：每个分区多个副本的从角色，实时的从Leader中同步数据，保持和Leader数据的同步，Leader发生故障的时候，某个Follower会成为新的Leader。2. 实操：
1）、docker拉取zookeeper和kafka
2）、启动zookeeper
3）、启动三台kafaka容器
4）、模拟生产者/消费者
a、进入kafka容器下的opt/kafka_2.13-2.8.1/bin路径，启用kafka-topics.sh创建一个3分区的topic(队列)
b、调用生产者创建消息
c、调用消费者查看消息（顺序不一致）
d、删除topic（逻辑删除）
3. 文件存储：
kafka的文件存储是落地存储的，通过对应的log和index文件。一个Topic有多个分区，一个分区有多个segment，一个segment有两个文件，一个log，一个index。 生产者不断向log文件追加消息。4. ACK机制：
发送ack的时机：
确保有follower与leader同步完成，leader在发送ack，这样可以保证在leader挂掉之后，follower中可以选出新的leader（主要是确保follower中数据不丢失）
follower同步完成多少才发送ack：
a、半数以上的follower同步完成，即可发送ack
优点：延迟低
缺点：缺点是选举新的leader的时候，容忍n台节点的故障，需要2n+1个副本（因为需要半数同意，所以故障的时候，能够选举的前提是剩下的副本超过半数），容错率为1/2
b、全部的follower同步完成，才可以发送ack
优点：容错率高
缺点：延迟高
5. ISR：
leader中维护了一个动态的ISR（in-sync replica set），即与leader保持同步的follower集合，当ISR中的follower完成数据的同步之后，给leader发送ack，如果follower长时间没有向leader同步数据，则该follower将从ISR中被踢出，该之间阈值由replica.lag.time.max.ms参数设定。当leader发生故障之后，会从ISR中选举出新的leader。6. ExactlyOnce：
将服务器的ACK级别设置为-1（all），可以保证producer到Server之间不会丢失数据，即At Least Once至少一次语义。将服务器ACK级别设置为0，可以保证生产者每条消息只会被发送一次，即At Most Once至多一次。 At Least Once可以保证数据不丢失，但是不能保证数据不重复，而At Most Once可以保证数据不重复，但是不能保证数据不丢失，对于重要的数据，则要求数据不重复也不丢失，即Exactly Once即精确的一次。7. 幂等性：
幂等性指代Producer不论向Server发送了多少次重复数据，Server端都只会持久化一条数据。幂等性结合At Least Once语义就构成了Kafka的Exactly Once语义。8. offset是什么？
对于每一个topic， Kafka集群都会维持一个分区日志，每个分区都是有序且顺序不可变的记录集，并且不断地追加到结构化的log文件。分区中的每一个记录都会分配一个id号来表示顺序，我们称之为offset，offset用来唯一的标识分区中每一条记录。
9. offset有什么用？消费者在消费数据时,发生宕机后,再次重新启动后,消费的数据需要从宕机位置开始读取
如果从头读取,有一部分消息一定出现了重复消费
如果从宕机时的消费位置读取,就不会出现重复消费
因此kafka设计了offset可以用于处理这种情况。
有两种方式,
自动提交，设置enable.auto.commit=true，更新的频率根据参数【auto.commit.interval.ms】来定。这种方式也被称为【at most once】，fetch到消息后就可以更新offset，无论是否消费成功。默认就是true
手动提交，设置enable.auto.commit=false，这种方式称为【at least once】。fetch到消息后，等消费完成再调用方法【consumer.commitSync()】，手动更新offset；如果消费失败，则offset也不会更新，此条消息会被重复消费一次
a、分区分配策略：
以parition进行分区。方便扩展和提高并发。
b、分区的原则：
①、指明partition（这里的指明是指第几个分区）的情况下，直接将指明的值作为partition的值
②、没有指明partition的情况下，但是存在值key，此时将key的hash值与topic的partition总数进行取余得到partition值
③、值与partition均无的情况下，第一次调用时随机生成一个整数，后面每次调用在这个整数上自增，将这个值与topic可用的partition总数取余得到partition值，即round-robin算法。
c、生产者ack机制：
producer在收到partition的ACK后，会发送新一轮的数据，否则会重新发送数据。
kafka为用户提供三种可靠级别：
0：producer不等待broker的ack，这一操作提供了最低的延迟，broker接收到还没有写入磁盘就已经返回，当broker故障时有可能丢失数据(保证数据只发送一次，最多一次)
1：producer等待broker的ack，partition的leader落盘成功后返回ack，如果在follower同步成功之前leader故障，那么将丢失数据。（只是leader落盘）
-1(all)：producer等待broker的ack，partition的leader和ISR的follower全部落盘成功才返回ack，但是如果在follower同步完成后，broker发送ack之前，如果leader发生故障，会造成数据重复。(这里的数据重复是因为没有收到，所以继续重发导致的数据重复)(保证数据不会丢失，最少一次)
一句话总结：producer返ack，0无落盘直接返，1只leader落盘然后返，-1全部落盘然后返
c、数据一致性：
幂等性：
幂等性指代Producer不论向Server发送了多少次重复数据，Server端都只会持久化一条数据。
kafka发送消息的三种方式：
①、发后既忘。
②、同步发送。
③、异步发送+回调函数。回调函数用来监控消息是否发送成功或发送失败做处理。
内存映射+零复制技术+顺序落盘。
消费者分区：
消费者采取pull（拉）的方式进行信息读取，这可以适应不同消费速率的消费者。
a、分区分配策略
1）round-robin循环：轮询方式
2）range：重分配方式
b、offset存储
由于Consumer在消费过程中可能会出现断电宕机等故障，Consumer恢复以后，需要从故障前的位置继续消费，所以Consumer需要实时记录自己消费到了那个offset，以便故障恢复后继续消费。
c、消费者组
①、当消费者组中的消费者大于topic分区数量时，同时只能有一个消费者获取数据，其余的消费者空闲；
②、当消费者组中的消费者小于topic分区数量时，有的消费者对应多个分区。
③、当消费者组中的消费者等于topic分区数量时，一个消费者对应一个分区。
④、当启动多个消费者组时，相同的数据会被不同的消费组中的消费者消费多次。
使用消费者组的优势：高性能，灵活，故障容灾。
Kafka集群中有一个broker会被选举为Controller，负责管理集群broker的上下线、所有topic的分区副本分配和leader的选举等工作。Controller的工作管理是依赖于zookeeper的。
顺序写和零复制技术 零复制技术：只用将磁盘文件的数据复制到页面缓冲区一次，然后将数据直接发送到网络中，不同的消费者可以使用同一个页面缓存，从而避免了复制操作。 zookeeper的作用：Kafka集群中有一个broker会被选举为Controller，负责管理集群broker的上下线、所有topic的分区副本分配和leader的选举等工作。Controller的工作管理是依赖于zookeeper的。14. 事务：
producer事务：
Kafka 的事务处理，主要是允许应用可以把消费和生产的 batch 处理（涉及多个 Partition）在一个原子单元内完成，操作要么全部完成、要么全部失败。这需要一个transaction_id和producer_id唯一匹配。
当用户使用 Kafka 的事务性时，Kafka 可以做到的保证：
①、跨会话的幂等性写入：即使中间故障，恢复后依然可以保持幂等性；
②、跨会话的事务恢复：如果一个应用实例挂了，启动的下一个实例依然可以保证上一个事务完成（commit 或者 abort）；
③、跨多个 Topic-Partition 的幂等性写入，Kafka 可以保证跨多个 Topic-Partition 的数据要么全部写入成功，要么全部失败，不会出现中间状态。
Kafka 事务性的使用：
用户只需要在 Producer 的配置中配置 transactional.id，通过 initTransactions() 初始化事务状态信息，再通过 beginTransaction() 标识一个事务的开始，然后通过 commitTransaction() 或 abortTransaction() 对事务进行 commit 或 abort，示例如下：
Properties props = new Properties();
props.put(“key.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);
props.put(“value.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);
props.put(“client.id”, “ProducerTranscationnalExample”);
props.put(“bootstrap.servers”, “localhost:9092”);
props.put(“transactional.id”, “test-transactional”);
props.put(“acks”, “all”);
KafkaProducer producer = new KafkaProducer(props);
producer.initTransactions();
try {
String msg = “matt test”;
producer.beginTransaction();
producer.send(new ProducerRecord(topic, “0”, msg.toString()));
producer.send(new ProducerRecord(topic, “1”, msg.toString()));
producer.send(new ProducerRecord(topic, “2”, msg.toString()));
producer.commitTransaction();
} catch (ProducerFencedException e1) {
e1.printStackTrace();
producer.close();
} catch (KafkaException e2) {
e2.printStackTrace();
producer.abortTransaction();
}
producer.close();
1、在写多个 Topic-Partition 时，执行的一批写入操作，有可能出现部分 Topic-Partition 写入成功，部分写入失败（比如达到重试次数），这相当于出现了中间的状态，这并不是我们期望的结果。 解决：如果启用事务性的话，涉及到多个 Topic-Partition 的写入时，这个事务操作要么会全部成功，要么会全部失败，不会出现上面的情况（部分成功、部分失败），如果有 Topic-Partition 无法写入，那么当前这个事务操作会直接 abort。
事务的实现原理：
Kafka Server 端的 TransactionCoordinator角色，充当协调者，在producer进行多parition写入的时候，会告诉coordinator事务开始，当所有commit完成后，事务才真正开始提交。
1、Finding a TransactionCoordinator ： 第一步首先需要做的就是找到这个事务 txn.id 对应的 TransactionCoordinator，Transaction Producer 会向 Broker （随机选择一台 broker，一般选择本地连接最少的这台 broker）发送 FindCoordinatorRequest 请求，获取其 TransactionCoordinator。
2、Getting a PID ：
3、Starting a Transaction ：
4、Consume-Porcess-Produce Loop ：
4.1 AddPartitionsToTxnRequest
Producer 在调用 send() 方法时，Producer 会将这个对应的 Topic—Partition 添加到 TransactionManager 的记录中，如果这个 Topic-Partition 之前不存在，那么就添加到 newPartitionsInTransaction 集合。
4.2. ProduceRequest
这一步与正常 Producer 写入基本上一样，就是相应的 Leader 在持久化数据时会在头信息中标识这条数据是不是来自事务 Producer 的写入（主要是数据协议有变动，Server 处理并不需要做额外的处理）。
4.3. AddOffsetsToTxnRequest
Producer 在调用 sendOffsetsToTransaction() 方法时，第一步首先向TransactionCoordinator 发送相应的 AddOffsetsToTxnRequest 请求。TransactionCoordinator 在收到这个请求时，处理方法与 4.1 中的一样，把这个 group.id 对应的 __consumer_offsets 的 Partition （与写入涉及的 Topic-Partition 一样）保存到事务对应的 meta 中，之后会持久化相应的事务日志。
4.4. TxnOffsetsCommitRequest
Producer 在收到 TransactionCoordinator 关于 AddOffsetsToTxnRequest 请求的结果后，后再次发送 TxnOffsetsCommitRequest 请求给对应的 GroupCoordinator。GroupCoordinator 在收到相应的请求后，会将 offset 信息持久化到 consumer offsets log 中（包含对应的 PID 信息），但是不会更新到缓存中，除非这个事务 commit 了，这样的话就可以保证这个 offset 信息对 consumer 是不可见的（没有更新到缓存中的数据是不可见的，通过接口是获取的，这是 GroupCoordinator 本身来保证的）。
5、Committing or Aborting a Transaction ：
5.1. EndTxnRequest
无论是 Commit 还是 Abort，对于 Producer 而言，都是向 TransactionCoordinator 发送 EndTxnRequest 请求，这个请求的内容里会标识是 commit 操作还是 abort 操作。
TransactionCoordinator 在收到 EndTxnRequest 请求后，会做以下处理：
①.更新事务的 meta 信息，状态转移成 PREPARE_COMMIT 或 PREPARE_ABORT，并将事务状态信息持久化到事务日志中；
②.根据事务 meta 信息，向其涉及到的所有 Topic-Partition 的 leader 发送 Transaction Marker 信息（也就是 WriteTxnMarkerRquest 请求，见下面的 5.2 分析）；
③.最后将事务状态更新为 COMMIT 或者 ABORT，并将事务的 meta 持久化到事务日志中，也就是 5.3 步骤。
5.2. WriteTxnMarkerRquest
WriteTxnMarkerRquest 是 TransactionCoordinator 收到 Producer 的 EndTxnRequest 请求后向其他 Broker 发送的请求，主要是告诉它们事务已经完成。不论是普通的 Topic-Partition 还是 __consumer_offsets，在收到这个请求后，都会把事务结果（Transaction Marker 的格数据式见前面）持久化到对应的日志文件中，这样下游 Consumer 在消费这个数据时，就知道这个事务是 commit 还是 abort。
5.3. Writing the Final Commit or Abort Message
当这个事务涉及到所有 Topic-Partition 都已经把这个 marker 信息持久化到日志文件之后，TransactionCoordinator 会将这个事务的状态置为 COMMIT 或 ABORT，并持久化到事务日志文件中，到这里，这个事务操作就算真正完成了，TransactionCoordinator 缓存的很多关于这个事务的数据可以被清除了。
事务demo实操：
springboot整合kafka：
1）、创建springboot项目，引入依赖和配置配置文件
2）、linux启动kafka容器，使用springboot整合kafka，调用kafkaTemplate的send方法发送消息给队列“topic-spring-01”，分区为0，key为0，消息为message。发送后并返回topic，分区，偏移量等。
3）、发送请求
控制台显示消费者收到消息
