java - 使用 KafkaTransactionManager 在事务 KafkaTemplate 中进行基于事件的提交

Question

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() 上添加 @Transaction 或

template.executeInTransaction(...

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

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

Answer 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<?> .