一、数据可靠性
1.Kafka数据应答级别
①acks=0
生产者发送过来的数据， 不需要等数据落盘应答。（可能丢数据）
acks=0.png
Leader宕机，数据未与Follower同步，数据还在内存中，全部丢失
②acks=1
生产者发送过来的数据， Leader收到数据后应答。（可能丢数据）
acks=1.png
数据应答完毕，此时Leader宕机，数据还没有同步给Follower，重新选举新的Leader，新的Leader不会再收到"hello"信息，因为生产者认为已经发送成功了。
③acks=-1
生产者发送过来的数据， Leader和ISR队列里面的所有节点收齐数据后应答。
acks=-1.png
如果Follower故障，不能及时与Leader进行同步时：
Leader维护了一个动态的 in-sync replica set（ ISR），意为和Leader保持同步的Follower+Leader集合(leader:0; isr:0,1,2)。
如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。
该时间阈值由replica.lag.time.max.ms参数设定，默认30s。
例如2超时，(leader:0, isr:0,1)。这样就不用等长期联系不上或者已经故障的节点。
存在的问题：
如果分区副本设置为1个（只有Leader）， 或者ISR里应答的最小副本数量
（ min.insync.replicas 默认为1） 设置为1， 和ack=1的效果是一
样的， 仍然有丢数的风险（leader： 0， isr:0） 。
数据完全可靠条件：
ACK级别设置为-1 + 分区副本大于等于2（至少一个副本） + ISR里应答的最小副本数量大于等于2。
package kafka;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
* @Title: MyProducer.java
* @Package kafka
* @Description: 生产者
* @Author: hongcaixia
* @Date: 2023/1/20 21:24
* @Version V1.0
*/
public class MyProducerAcks {
public static void main(String[] args) {
// 1. 创建kafka生产者的配置对象
Properties properties = new Properties();
// 2. 给kafka配置对象添加配置信息：bootstrap.servers
// 连接集群
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.1:9092,192.168.0.2:9092");
// 指定序列化类型 key,value 序列化（必须）：key.serializer，value.serializer
properties.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");
// 设置应答级别
properties.put(ProducerConfig.ACKS_CONFIG, "1");
// 设置重试次数 默认为int最大值
properties.put(ProducerConfig.RETRIES_CONFIG, 3);
// 3. 创建kafka生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
// 4. 调用send方法,发送消息
for (int i = 0; i < 5; i++) {
kafkaProducer.send(new ProducerRecord<>("first", "record" + i));
}
// 5. 关闭资源
kafkaProducer.close();
}
}
2.可靠性总结
- acks=0， 生产者发送过来数据就不管了， 可靠性差， 效率高；
- acks=1， 生产者发送过来数据Leader应答， 可靠性中等， 效率中等；
- acks=-1， 生产者发送过来数据Leader和ISR队列里面所有Follwer应答， 可靠性高， 效率低；
在生产环境中， acks=0很少使用； acks=1， 一般用于传输普通日志， 允许丢个别数据； acks=-1， 一般用于传输和钱相关的数据，对可靠性要求比较高的场景。
二、数据幂等性
acks=-1时，可能存在生产者重复发送数据：
数据发送完毕，落盘完毕，同步完毕，但是在应答的时候，Leader宕机，选举Follower成为新Leader，因为producer未收到应答，所以再次发送数据，导致重复发送。
数据重复.png
1.数据传递语义
- 至少一次（At Least Once）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2
- 最多一次（At Most Once） = ACK级别设置为0
总结：
- At Least Once可以保证数据不丢失， 但是不能保证数据不重复；
- At Most Once可以保证数据不重复， 但是不能保证数据不丢失。
- 精确一次（Exactly Once）：对于一些非常重要的信息， 比如和钱相关的数据， 要求数据既不能重复也不丢失。
Kafka 0.11版本以后， 引入了一项重大特性： 幂等性和事务。
2.幂等性
Kafka 幂等性原理
幂等性就是指Producer不论向Broker发送多少次重复数据， Broker端都只会持久化一条， 保证了不重复。
精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2） 。
幂等性.png
重复数据的判断标准： 具有<PID, Partition, SeqNumber>相同主键的消息提交时， Broker只会持久化一条。
- PID是Kafka每次重启都会分配一个新的；
- Partition 表示分区号；
- Sequence Number是单调自增的。
所以幂等性只能保证的是在单分区单会话内不重复。
开启幂等性：enable.idempotence 默认为 true， false 关闭。
props.put("enable.idempotence", ture)，或
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG， true)。
3.生产者事务
Kafka 事务原理
说明：开启事务， 必须开启幂等性。
Kafka 事务原理.png
transactional.id：
Producer在使用事务前，必须先自定义一个唯一的transactional.id。有了transactional.id，即使客户端挂掉了，重启后也能继续处理未完成的事务。
_transaction_state-分区-Leader：
默认50个分区，每个分区负责一部分事务。
事务划分是根据transactional.id的hashcode值%50计算出该事务属于哪个分区，该分区Leader副本所在的borker节点即为这个transactional.id对应的Transaction Coordinator节点。
kafka事务API
// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,String consumerGroupId) throws ProducerFencedException;
// 4 提交事务
void commitTransaction() throws ProducerFencedException;
// 5 放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;
demo：
- 开启 enable.idempotence = true。
- 设置 Producer 端参数 transactional. id。
package kafka;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
* @Title: MyProducer.java
* @Package kafka
* @Description: 生产者
* @Author: hongcaixia
* @Date: 2023/1/20 21:24
* @Version V1.0
*/
public class MyProducerTransaction {
public static void main(String[] args) {
// 1. 创建kafka生产者的配置对象
Properties properties = new Properties();
// 2. 给kafka配置对象添加配置信息：bootstrap.servers
// 连接集群
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.1:9092,192.168.0.2:9092");
// 指定序列化类型 key,value 序列化（必须）：key.serializer，value.serializer
properties.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");
// 指定 Producer 幂等性
properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// 指定事务id(保证全局唯一)
properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional_id_01");
// 3. 创建kafka生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
//初始化事务
kafkaProducer.initTransactions();
//开启事务
kafkaProducer.beginTransaction();
try {
// 4. 调用send方法,发送消息
for (int i = 0; i < 5; i++) {
kafkaProducer.send(new ProducerRecord<>("first", "record" + i));
}
//提交事务
kafkaProducer.commitTransaction();
}catch (Exception e){
//终止事务
kafkaProducer.abortTransaction();
}finally {
// 5. 关闭资源
kafkaProducer.close();
}
}
}
这段代码能够保证 Record1～Record5 被当作一个事务统一提交到 Kafka，要么它们全部提交成功，要么全部写入失败。
实际上即使写入失败，Kafka 也会把它们写入到底层的日志中，也就是说 Consumer 还是会看到这些消息。
因此在 Consumer 端，读取事务型 Producer 发送的消息也是需要一些变更。修改起来也很简单，设置 isolation.level 参数的值即可。当前这个参数有两个取值：
- read_uncommitted：这是默认值，表明 Consumer 能够读取到 Kafka 写入的任何消息，不论事务型 Producer 提交事务还是终止事务，其写入的消息都可以读取。很显然，如果你用了事务型 Producer，那么对应的 Consumer 就不要使用这个值。
- read_committed：表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。当然了，它也能看到非事务型 Producer 写入的所有消息。
三、数据有序性
数据有序性.png
生产者端对每个broker节点最多可以缓存5个没有应答的请求。
当开启了幂等性后，因为有了对应的单调递增的序列号Sequence Number，当request2发送失败了，进行重试，这时候request3过来，1->3并不是单调递增的，此时request3就缓存在内存中，不能落盘，直到收到了2的请求，再对内存中的所有数据进行重新排序，再进行落盘，保证了单分区内的有序。
kafka在1.x版本之前保证数据单分区有序条件：
max.in.flight.requests.per.connection=1（ 不需要考虑是否开启幂等性） 。
kafka在1.x及以后版本保证数据单分区有序条件：
- 未开启幂等性：
max.in.flight.requests.per.connection需要设置为1 - 开启幂等性：
max.in.flight.requests.per.connection需要设置小于等于5。
在kafka1.x以后，启用幂等后， kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。
极客时间《Kafka 核心技术与实战》学习笔记Day9 - http://gk.link/a/11UOV
网友评论