快捷導(dǎo)航

springboot使用kafka事務(wù)的示例代碼

更新時(shí)間：2024年06月14日 09:38:31 作者：????????五敷有你

Kafka?同數(shù)據(jù)庫一樣支持事務(wù),當(dāng)發(fā)生異常的時(shí)候可以進(jìn)行回滾,確保消息監(jiān)聽器不會(huì)接收到一些錯(cuò)誤的或者不需要的消息,本文就來介紹一下springboot使用kafka事務(wù)的示例代碼,具有一定的參考價(jià)值,感興趣的可以了解一下

先看下下面這種情況，程序都出錯(cuò)了，按理說消息也不應(yīng)該成功

@GetMapping("/send")
public void test9(String message) {
    kafkaTemplate.send(topic, message);
    throw new RuntimeException("fail");
}

但是執(zhí)行結(jié)果是發(fā)生了異常并且消息發(fā)送成功了：

Kafka 同數(shù)據(jù)庫一樣支持事務(wù)，當(dāng)發(fā)生異常的時(shí)候可以進(jìn)行回滾，確保消息監(jiān)聽器不會(huì)接收到一些錯(cuò)誤的或者不需要的消息。

kafka事務(wù)屬性是指一系列的生產(chǎn)者生產(chǎn)消息和消費(fèi)者提交偏移量的操作在一個(gè)事務(wù)，或者說是是一個(gè)原子操作），同時(shí)成功或者失敗。使用事務(wù)也很簡單，需要先開啟事務(wù)支持，然后再使用。

如何開啟事務(wù)

如果使用默認(rèn)配置只需要在yml添加spring.kafka.producer.transaction-id-prefix配置來開啟事務(wù)，之前沒有使用默認(rèn)的配置，自定義的kafkaTemplate，那么需要在ProducerFactory中設(shè)置事務(wù)Id前綴開啟事務(wù)并將KafkaTransactionManager注入到spring中，看下KafkaProducerConfig完整代碼：

@Configuration
@EnableKafka
public class KafkaProducerConfig {
@Value("${kafka.producer.servers}")
private String servers;
@Value("${kafka.producer.retries}")
private int retries;
 
public Map<String,Object> producerConfigs(){
    Map<String,Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
    props.put(ProducerConfig.RETRIES_CONFIG,retries);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // 配置分區(qū)策略
    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.example.springbootkafka.config.CustomizePartitioner");
    // 配置生產(chǎn)者攔截器
    props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.example.springbootkafka.interceptor.CustomProducerInterceptor");
    // 配置攔截器消息處理類
    SendMessageInterceptorUtil sendMessageInterceptorUtil = new SendMessageInterceptorUtil();
    props.put("interceptorUtil",sendMessageInterceptorUtil);
    return props;
}
 
@Bean
public ProducerFactory<String,String> producerFactory(){
    DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory(producerConfigs());
    //設(shè)置事務(wù)Id前綴 開啟事務(wù)
    producerFactory.setTransactionIdPrefix("tx-");
    return producerFactory;
}
 
@Bean
public KafkaTemplate<String,String> kafkaTemplate(){
    return new KafkaTemplate<>(producerFactory());
}
 
@Bean
public KafkaTransactionManager<Integer, String> kafkaTransactionManager(ProducerFactory<String, String> producerFactory) {
    return new KafkaTransactionManager(producerFactory);
}
}

配置開啟事務(wù)后，使用大體有兩種方式，先記錄下第一種使用事務(wù)方式：使用 executeInTransaction 方法

直接看下代碼：

@GetMapping("/send11")
public void test11(String message) {
    kafkaTemplate.executeInTransaction(operations ->{
        operations.send(topic,message);
        throw new RuntimeException("fail");
    });
}

當(dāng)然你可以這么寫：

@GetMapping("/send11")
public void test11(String message) {
    kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback(){
        @Override
        public Object doInOperations(KafkaOperations operations) {
            operations.send(topic,message);
            throw new RuntimeException("fail");
        }
    });
}

啟動(dòng)項(xiàng)目，訪問http://localhost:8080/send10?message=test10 結(jié)果如下：

如上：消費(fèi)者沒打印消息，說明消息沒發(fā)送成功，并且前面會(huì)報(bào)錯(cuò)org.apache.kafka.common.KafkaException: Failing batch since transaction was aborted 的錯(cuò)誤，說明事務(wù)生效了。

第一種使用事務(wù)方式：使用 @Transactional 注解方式直接在方法上加上@Transactional注解即可，看下代碼：

@GetMapping("/send12")
@Transactional
public void test12(String message) {
    kafkaTemplate.send(topic, message);
    throw new RuntimeException("fail");
}

如果開啟的事務(wù)，則后續(xù)發(fā)送消息必須使用@Transactional注解或者使用kafkaTemplate.executeInTransaction() ,否則拋出異常，異常信息如下：

貼下完整的異常吧：java.lang.IllegalStateException: No transaction is in process; possible solutions: run the template operation within the scope of a template.executeInTransaction() operation, start a transaction with @Transactional before invoking the template method, run in a transaction started by a listener container when consuming a record

到此這篇關(guān)于springboot使用kafka事務(wù)的文章就介紹到這了,更多相關(guān)springboot kafka事務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: