一 kafka发送数据问题-丢数据

1.1 丢数据问题抛出

根据发送端Ack的发送机制，存在以下问题，怎么保证其可靠性。

情况一： ack=0，生产者发送过来的数据，不需要等数据落盘应答

问题：存在数据丢数据问题

情况二： ack=1，生产者发送过来数据，leader收到数据后就应答。

 问题： leader收到应答后，正好某种原因挂掉了，新的leader被选举出来，但是没有收到hello的信息，因为之前的旧leader发送的ack应答机制，生产者已经认为发送成功了。此时的问题还是丢数据。

情况三：ack=-1，生产者发送过来数据，leader和副本都收到数据后才进行应答。

 问题：Leader收到数据，所有Follower都开始同步数据， 但有一个Follower，因为某种故障，迟迟不能与Leader进行 同步，那这个问题怎么解决呢？

方案为：

1.Leader维护了一个动态的in-sync replica set（ISR），意为和Leader保持同步Follower+Leader集合(leader：0，isr:0,1,2)。
2.如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。例如follower2超时，将其提出isr，即(leader:0, isr:0,1)。
这样就把哪些长期联系不上，发生故障的节点，排除再外。不再等待这些提出isr中的节点是否接收到数据。

1.2 解决方案

1.3 代码实现

package com.ljf.spring.boot.demo.producer;import com.ljf.spring.boot.demo.utils.DateUtils;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Date;
import java.util.Properties;/*** @ClassName: TutuProducer* @Description: TODO* @Author: liujianfu* @Date: 2022/04/07 11:05:50* @Version: V1.0**/
public class TutuProducer {public static void main(String[] args) throws InterruptedException {// 1. 创建 kafka 生产者的配置对象Properties properties = new Properties();// 2. 给 kafka 配置对象添加配置信息properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.152.136:9092,192.168.152.138:9092,192.168.152.140:9092");// key,value 序列化（必须）：key.serializer，value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());//优化参数，提高吞吐量// batch.size：批次大小，默认 16Kproperties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);// linger.ms：等待时间，默认 0properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);// RecordAccumulator：缓冲区大小，默认 32M：buffer.memoryproperties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);// compression.type：压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstdproperties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");//设置可靠性// 设置 acksproperties.put(ProducerConfig.ACKS_CONFIG, "all");// 重试次数 retries，默认是 int 最大值，2147483647properties.put(ProducerConfig.RETRIES_CONFIG, 3);// 3. 创建 kafka 生产者对象KafkaProducer<String, String> kafkaProducer = newKafkaProducer<String, String>(properties);// 4. 调用 send 方法,发送消息for (int i = 0; i < 5; i++) {// 添加回调String info="xiawu-xx"+i+" 时间为:"+ DateUtils.dateToStr(new Date(), "yyyy-MM-dd HH:mm:ss");kafkaProducer.send(new ProducerRecord<>("kafka-ljf",info), new Callback() {// 该方法在 Producer 收到 ack 时调用，为异步调用@Overridepublic void onCompletion(RecordMetadata metadata,Exception exception) {if (exception == null) {// 没有异常,输出信息到控制台System.out.println(" 主题： " +metadata.topic() + "->" + "分区：" + metadata.partition());} else {// 出现异常打印exception.printStackTrace();}}});// 延迟一会会看到数据发往不同分区Thread.sleep(2000);}// 5. 关闭资源kafkaProducer.close();}
}

结果：

 二 kafka发送数据问题-重复数据

2.1 数据语义

1.至少一次（At Least Once）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

2.最多一次（At Most Once）= ACK级别设置为0

At Most Once可以保证数据不重复，但是不能保证数据不丢失。 
At Least Once可以保证数据不丢失，但是不能保证数据不重复

如果要保证既不重复数据，又不丢失数据，那么需要这样设置：

3.精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。

Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。

2.2 解决数据重复

1.幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。
2.精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2）
3.重复数据的判断标准：具有<PID, Partition, SeqNumber>相同主键的消息提交时，Broker只会持久化一条。
其 中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。
4.所以幂等性只能保证的是在单分区单会话内不重复。

 2.3 开启幂等性设置

开启参数 enable.idempotence 默认为 true，false 关闭。

 2.4 生产事务

1.开启事务，必须开启幂等性

2.producer在使用事务功能前，必须先自定义一个唯一的transactional.id,有了transactional.id，即使客户端挂掉了，它重启后能继续处理未完成的事务。 

2.5 kafka使用事务的API

2.6 代码案例

public class TransactionProducer {public static void main(String[] args) {// 1. 创建 kafka 生产者的配置对象Properties properties = new Properties();// 2. 给 kafka 配置对象添加配置信息：bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.152.136:9092,192.168.152.138:9092,192.168.152.140:9092");// key,value 序列化（必须）：key.serializer，value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");// 设置事务 id（必须），事务 id 任意起名properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");// 3. 创建 kafka 生产者对象KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);// 初始化事务kafkaProducer.initTransactions();// 开启事务kafkaProducer.beginTransaction();try {// 4. 调用 send 方法,发送消息for (int i = 100; i < 103; i++) {String v = "kfdemo-haha" + i + "时间:" + DateUtils.dateToStr(new Date(), "yyyy-MM-dd HH:mm:ss");kafkaProducer.send(new ProducerRecord<>("kafka-ljf", v));}// 提交事务kafkaProducer.commitTransaction();} catch (Exception e) {// 终止事务kafkaProducer.abortTransaction();} finally {// 5. 关闭资源kafkaProducer.close();}System.out.println("执行完毕！！！");}
}

消费者：

 三 数据的有序

单分区内，有序（有条件的，详见下节）； 多分区，分区与分区间无序；

四 数据的乱序

1）kafka在1.x版本之前保证数据单分区有序，条件如下：

max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）。

2）kafka在1.x及以后版本保证数据单分区有序，条件如下：

（2）开启幂等性

max.in.flight.requests.per.connection需要设置小于等于5。

（1）未开启幂等性

max.in.flight.requests.per.connection需要设置为1。

原因说明：因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，

故无论如何，都可以保证最近5个request的数据都是有序的。