事务生产者

前面我们学习了幂等生产者，知道了幂等生产者存在的两个局限不支持跨回话和多分区以及它的优点就是使用简单，针对这两个不足我们看一下kafka 的另外一个解决方案——事务

Kafka 自 0.11 版本开始也提供了对事务的支持，目前主要是在 read committed 隔离级别上做事情。
它能保证多条消息原子性地写入到目标分区，同时也能保证 Consumer 只能看到事务成功提交的消息。
事务型 Producer 能够保证将消息原子性地写入到多个分区中。这批消息要么全部写入成功，要么全部失败。
事务型 Producer 也不惧进程的重启。Producer 重启回来后，Kafka 依然保证它们发送消息的精确一次处理。

其实看到这里我们大致知道了，全部失败的情况下不能保证数据不写入kafka,因为kafka 并没有回滚机制，所以失败的情况下需要consumer 不可见，就是consumer 不去读取失败的数据。

使用方法

producer 端

和幂等性 Producer 一样，开启 enable.idempotence = true。
设置 Producer 端参数 transactional. id 最好为其设置一个有意义的名字。
需要多代码进行一定的改造(可以和幂等的时候对比，事务型 Producer 的显著特点是调用了一些事务 API)

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

这段代码能够保证 Record1 和 Record2 被当作一个事务统一提交到 Kafka，要么它们全部提交成功，要么全部写入失败。实际上即使写入失败，Kafka 也会把它们写入到底层的日志中，也就是说 Consumer 还是会看到这些消息。

consumer 端配合

实际上即使写入失败，Kafka 也会把它们写入到底层的日志中，也就是说 Consumer 还是会看到这些消息。
因此在 Consumer 端，读取事务型 Producer 发送的消息也是需要一些变更的。修改起来也很简单，设置 isolation.level 参数的值即可

read_uncommitted

这是默认值，表明 Consumer 能够读取到 Kafka 写入的任何消息，不论事务型 Producer 提交事务还是终止事务，其写入的消息都可以读取。很显然，如果你用了事务型 Producer，那么对应的 Consumer 就不要使用这个值。

read_committed

表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。当然了，它也能看到非事务型 Producer 写入的所有消息。

实现原理

事务型 Producer 可以实现一组消息要么全部写入成，要么全部失败，但是事务型Producer是具体怎么实现多分区以及多会话上的消息无重复的呢，主要的机制是两阶段提交（2PC），引入了事务协调器的组件帮助完成分布式事务
Kafka引入了一个新的组件Transaction Coordinator，它管理了一个全局唯一的事务ID（Transaction ID），并将生产者的PID和事务ID进行绑定，当生产者重启时虽然PID会变，但仍然可以和Transaction Coordinator交互，通过事务ID可以找回原来的PID，这样就保证了重启后的生产者也能保证Exactly Once 了。
同时，Transaction Coordinator 将事务信息写入 Kafka 的一个内部 Topic，即使整个kafka服务重启，由于事务状态已持久化到topic，进行中的事务状态也可以得到恢复，然后继续进行。

代码演示

生产者代码

这里我们的生产者偶数值，奇数值是不提交的

public class TransactionProducer {
    private static final Logger logger = LoggerFactory.getLogger(IdempotenceProducer.class);
    private static KafkaProducer<String, String> producer = null;

    /* 初始化生产者 */
    static {
        Properties configs = initConfig();
        producer = new KafkaProducer<String, String>(configs);
    }

    /* 初始化配置 */
    private static Properties initConfig() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        // 必须设置的
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "1v1_bu_stu_reg");
        props.put("retries", 3);
        return props;
    }

    public static void main(String[] args) throws InterruptedException {
        //消息实体
        ProducerRecord<String, String> record = null;
        producer.initTransactions();
        for (int i = 0; i < 10000; i++) {
            record = new ProducerRecord<String, String>("test", "value" + i);

            producer.beginTransaction();
            //发送消息
            producer.send(record);
            producer.send(record);
            if (i %2 ==0){
                producer.commitTransaction();
            }else {
                producer.abortTransaction();
            }
            TimeUnit.MILLISECONDS.sleep(1000);
        }
        producer.close();
    }
}

这里注意下，我们除了配置开启事务和事务ID 之外，还需要设置重试次数，否则会得到如下错误

Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka producer
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:430)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:298)
	at com.kingcall.clients.producer.transaction.TransactionProducer.<clinit>(TransactionProducer.java:22)
Caused by: org.apache.kafka.common.config.ConfigException: Must set retries to non-zero when using the idempotent producer.
	at org.apache.kafka.clients.producer.KafkaProducer.configureRetries(KafkaProducer.java:545)
	at org.apache.kafka.clients.producer.KafkaProducer.newSender(KafkaProducer.java:458)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:419)
	... 2 more

消费者命令行演示(消费失败数据)

我们先用命令行来演示一下，我们如果开启消费者相关配置，我们是会消费到事务失败的消息的，也就是失败的消息已经写入了kafka 了

从上面的消费信息来看，好像我们的消费者确实只消费到了偶数值，那意味着奇数值没提交吗，也就是失败的消息没有提交到kafka 吗，其实不是这是和kafka 的批量提交有关的,也就是说是因为producer 根本没有提交，然后事务abort 的话，缓存里的消息就没有发送了，这里我们在发送消息的时候执行一下等待，等待producer 将数据发送出去之后我们再abort事务，生产者和事务提交之间加一行代码TimeUnit.MILLISECONDS.sleep(1000);,然后我们看一下效果

producer.beginTransaction();
 //发送消息
 producer.send(record);
 TimeUnit.MILLISECONDS.sleep(1000);
 if (i %2 ==0){
     producer.commitTransaction();
 }else {
     producer.abortTransaction();
 }

这次我们就看到了事务失败的时候提交的数据了

消费者代码演示(不消费失败数据)

我们保持生产者代码不变，然后通过在消费者端配置不消费事务失败的数据

public class TransactionConsumer {
    private static KafkaConsumer<String,String> consumer;
    /** * 初始化消费者 */
    static {
        Properties configs = initConfig();
        consumer = new KafkaConsumer<String, String>(configs);
        consumer.subscribe(Arrays.asList("test"));

    }
    /** * 初始化配置 */
    private static Properties initConfig(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit",true);
        props.put("auto.commit.interval.ms", 1000);
        props.put("session.timeout.ms", 30000);
        props.put("max.poll.records", 1000);
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        // 只消费 read_committed数据
        props.put("isolation.level","read_committed");

        return props;
    }

    public static void main(String[] args) {

        while (true) {
            // 这里的参数指的是轮询的时间间隔，也就是多长时间去拉一次数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(3000));
            records.forEach((ConsumerRecord<String, String> record)->{
                System.out.println(record.value());
            });
        }
    }
}

我们看一下消费结果

我们看到值消费了事务提交的数据

总结

幂等的意思是多次重试是对数据不影响，因为broker 已经帮我们做了去重，所以消费者不需要做变动
事务的意思是读取数据的时候不要读取事务失败的消息，但是因为数据已经写进去去了，所有consumer 要设置不能读取这些数据，从而保证对业务对下游没有影响
不论是幂等producer 还是事务producer都可能发送重复消息，而且同样broker端有一套机制来去重(幂等性依赖seq number机制，事务依赖各种marker来标记),区别就是这个去重是不是broker来做的
等性 Producer 和事务型 Producer 都是 Kafka 社区力图为 Kafka 实现精确一次处理语义所提供的工具，只是它们的作用范围是不同的。
幂等性 Producer 只能保证单分区、单会话上的消息幂等性；而事务能够保证跨分区、跨会话间的幂等性。从交付语义上来看，自然是事务型 Producer 能做的更多。
比起幂等性 Producer，事务型 Producer 的性能要更差，在实际使用过程中，我们需要仔细评估引入事务的开销，切不可无脑地启用事务。
对于consumer端，由于偏移量的提交和消息处理的顺序有前有后，依然可能导致重复消费或者消息丢失消费，如果要实现消费者消费的精确一次，还需要通过额外机制在消费端实现偏移量提交和消息消费的事务处理
kafka 的事务不只支持跨回话还支持跨topic的，也就是支持在一个producer 中同时写多个topic

kafka系列之事务生产者(12)