分布式服务分布式@架构师

Kafka-生产者数据问题及解决方案

2023-01-23  本文已影响0人  我可能是个假开发

一、数据可靠性

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， 一般用于传输普通日志， 允许丢个别数据； acks=-1， 一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

二、数据幂等性

acks=-1时，可能存在生产者重复发送数据：
数据发送完毕，落盘完毕，同步完毕，但是在应答的时候，Leader宕机，选举Follower成为新Leader，因为producer未收到应答，所以再次发送数据，导致重复发送。


数据重复.png

1.数据传递语义

总结：

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

2.幂等性

Kafka 幂等性原理
幂等性就是指Producer不论向Broker发送多少次重复数据， Broker端都只会持久化一条， 保证了不重复。
精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2） 。

幂等性.png

重复数据的判断标准： 具有<PID, Partition, SeqNumber>相同主键的消息提交时， Broker只会持久化一条。

所以幂等性只能保证的是在单分区单会话内不重复。

开启幂等性：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：

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 参数的值即可。当前这个参数有两个取值：

三、数据有序性

数据有序性.png

生产者端对每个broker节点最多可以缓存5个没有应答的请求。
当开启了幂等性后，因为有了对应的单调递增的序列号Sequence Number，当request2发送失败了，进行重试，这时候request3过来，1->3并不是单调递增的，此时request3就缓存在内存中，不能落盘，直到收到了2的请求，再对内存中的所有数据进行重新排序，再进行落盘，保证了单分区内的有序。

kafka在1.x版本之前保证数据单分区有序条件：
max.in.flight.requests.per.connection=1（ 不需要考虑是否开启幂等性） 。

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

在kafka1.x以后，启用幂等后， kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。

极客时间《Kafka 核心技术与实战》学习笔记Day9 - http://gk.link/a/11UOV

上一篇下一篇

猜你喜欢

热点阅读