Kafka幂等Producer和事务Producer

た入场券 2021-11-11 13:44 341阅读 0赞

### 一、Kafka 消息交付可靠性 ###

Kafka 对Producer和Consumer 需要处理的消息提供三种承诺：

*  最多一次（at most once）：消息可能丢失，但绝不会被重发；
 *  至少一次（at least once）：消息不会丢失，但有可能被重复发送（默认此可靠性）；
 *  精确一次（at most once）：消息不会丢失，也不会被重复发送。

目前，Kafka默认提供的交付可靠性保障是第二种，即至少一次。Producer 发送消息后，没有收到 Broker的callback，就会发起重试，发送相同的消息到Broker。

### 二、幂等性 Producer ###

在Kafka中，Producer 默认不是幂等性的，但我们可以闯进幂等性Producer。在0.11 版本之后，添加一个参数即可实现幂等性，即 props.put(“enable.idempotence”,true)或props.put(ProducerConfig.ENABLE\_IDEMPOTENCE\_CONFIG,true)。  
设置参数后，Kafka 自动帮你做消息的重复去重。底层实现原理很简单，就是典型的空间去换时间的优化思路，即在Broker端多保存一些字段。当Producer 发送相同字段值的消息后，Broker能够自动知晓这消息已经去重复了， 于是可以在后台默默的把它们"丢弃"掉。  
Producer 幂等性只能保证单分区上的幂等性，即保证某一个分区上不出现重复消息，它无法实现多分区的幂等性。你可以理解为Producer 进程的一次运行，但你重启了Producer 进程以后，这种幂等性保证就是丧失了。

### 三、事务型Producer ###

事务型Producer能够保证消息原子性地写入到多个分区中。这批消息要么全部成功，要么全部失败。另外，事务型Producer也不惧进程重启，Producer重启后，Kafka依然保证它们发送消息的精确一次处理。  
设置事务型Producer 需要满足两要求：

*  和幂等性Producer一样，开启 enable.idempotence = true;
 *  设置Producer 端参数 [transaction.id][] ,最好为其设置一个有意义的名字。

在Producer 代码中也需要做调整，代码示例如下：

producer.initTransactions();
    try {
                producer.beginTransaction();
                producer.send(record1);
                producer.send(record2);
                producer.commitTransaction();
    } catch (KafkaException e) {
                producer.abortTransaction();
    }

和普通的Producer 代码相比，事务型Producer 的显著特点是调用了一些事务API 如：initTransaction（事务初始化）、beginTransaction（事务开始）、commitTranscation（事务提交）、abortTransaction（事务终止）。  
这段代码中能够保证record 1 和record 2 被当做一个事务统一提交到Kafka，要么全部提交成功，要么全部提交失败。  
事务型Producer 是基于两阶段提交，时间上写入失败时，Kafka也会把消息写入到日志中，也就是说Consumer 有可能会看到，写入失败数据，所以读取事务消息Consumer也需要做一些变更，即 设置 isoIation.level 参数，当前这个参数有两个取值：

*  read\_uncommitted:这是默认值，表明Consumer  
    能够读取到Kafka写入的任何消息，在用事务型Producer时，对应的consumer 就不要使用这个值。
 *  read\_committed:表明Consumer只会读取事务型Producer成功提交事务写入的消息。当然它也能看到非事务型Producer 写入的消息。

[transaction.id]: http://transaction.id