springboot使用kafka事务的示例代码

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

@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);
    // 配置分区策略
    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.example.springbootkafka.config.CustomizePartitioner");
    // 配置生产者拦截器
    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());
    //设置事务Id前缀 开启事务
    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);
}
} 

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

@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");
        }
    });
}

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