kafka生产者，数据的可靠性总结与数据重复问题

生产者发送数据的可靠性主要是依赖应答acks的等级，不同的等级对应的可靠性也不尽相同。

acks=0：

安全性很低，如果在落盘数据时leader挂掉，那么数据丢失。

acks=1：

acks= -1（all）

 类似于datanode和namenode的心跳机制。

可靠性：

 在代码中配置应答acks：

        //配置acks
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        //配置应答失败重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG,3);

数据的重复问题：

精确一次 ：

幂等性和事务：

幂等性：无论Prducer不论向Broker发送多少重复数据，Broker端都只会持久化一条，保证数据的不重复。

精确一次=幂等性+至少一次

那么幂等性如何判断数据是相同的呢？

具有相同主键的消息提交时，Broker只会持久化一条。其 中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。

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

生产者事务：开启事务，必须先开启幂等性

说白了,事务主要是弥补幂等性多会话中Pid会不同的情况。

// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map offsets,
 String consumerGroupId) throws 
ProducerFencedException;
// 4 提交事务
void commitTransaction() throws ProducerFencedException;
// 5 放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;


public class CustomProducerTran {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //创建配置对象
        Properties properties = new Properties();
        //TODO 连接kafka与集群
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");
        //TODO 指定对应的key和value的序列化类型
        //StringSerializer  : 传入的类型的序列化，需要看你传入的是什么类型，我们这里是String
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//配置事务id  ***************很重要
properties.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,"tran");
        //TODO 创建kafka对象
        KafkaProducer kp = new KafkaProducer(properties);

        //初始化事务
        kp.initTransactions();
        //开启事务
        kp.beginTransaction();

        //TODO 发送消息
        try {
            for (int i = 0; i < 5; i++) {
                kp.send(new ProducerRecord("third", "atguigu")).get();
            }
            //提交事务
            kp.commitTransaction();
        } catch (InterruptedException | ExecutionException e) {
            //回滚事务
            kp.abortTransaction();
        }finally {
            //TODO 关闭资源
            kp.close();
        }
    }
}