使用 KafkaTransactionManager 在事务 KafkaTemplate 中基于事件的提交答案

【问题标题】：Event based commit in transaction KafkaTemplate using KafkaTransactionManager使用 KafkaTransactionManager 在事务 KafkaTemplate 中基于事件的提交
【发布时间】：2018-09-15 04:29:28
【问题描述】：

Spring 托管的 KafkaTemplate 提供

template.send(record).addCallback(...
template.executeInTransaction(...

现在假设我有一个方法 doWork()，它在事件（比如 TCP/IP 消息）上触发。

@Autowired
KafkaTemplate template;

// This method is triggered on a event
doWork(EventType event){
    switch(event){
        case Events.Type1 :
            template.send(record); break;
        case Events.Type2 :
            // Question : How do I achieve a commit of all my previous sends here?
         default : break;
    }
}

基本上，我需要通过在 doWork() 或 a 上添加 @Transaction 来实现事务

template.executeInTransaction(...

在代码中。但是我想批处理几个 [template.send()] 并在几次调用 doWork() 方法后进行提交，我该如何实现？

我的生产者配置启用了事务，并将 KafkaTransactionManager 连接到生产者工厂。

【问题讨论】：

标签： java spring kafka-producer-api spring-kafka

【解决方案1】：

kafkaTemplate.executeInTransaction(t -> {
    boolean stayIntransaction = true;
    while (stayInTransaction) {
        Event event = readTcp()
        doWork(event);
        stayInTransaction = transactionDone(event);
    }
}

只要doWork()方法使用相同的模板，并且在回调范围内运行，工作就会在事务中运行。

或者

@Transactional
public void doIt() {
    boolean stayIntransaction = true;
    while (stayInTransaction) {
        Event event = readTcp()
        doWork(event);
        stayInTransaction = transactionDone(event);
    }
}

使用声明式事务时。

如果 TCP 事件是异步的，您将需要以某种方式将它们交给运行事务的线程，例如使用 BlockingQueue<?>。

【讨论】：