@Configuration
public class KafkaConfig {

    public final static String TEST_TOPIC = "testTopic";

    // 创建一个名为testTopic的Topic并设置分区数为4，分区副本数为1
    // 如果要修改分区数，只需修改配置值重启项目即可
    // 修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic(TEST_TOPIC, 4, (short) 1);
    }

}

@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;

@GetMapping("/send/{message}")
public void send(@PathVariable String message) {
    ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send(KafkaConfig.TEST_TOPIC, message);
    listenableFuture.addCallback(
        sendResult -> logger.info("success: partition is {}", sendResult.getRecordMetadata().partition()),
        throwable -> logger.info("failure : message is {}", throwable.getMessage())
    );
}

@Slf4j
@Component
public class OneTopicReceiver{

    @KafkaListener(topics = {"twoTopic"}, errorHandler = "consumerAwareErrorHandler", concurrency = "4", containerFactory = "filterContainerFactory")
    public void onReceive(Message message, Acknowledgment ack){
        log.info("{}:{}",message.getHeaders().get("kafka_receivedPartitionId"), message.getPayload());
        ack.acknowledge();
    }

}

//通过带代码可以看出，只要transaction-id-prefix配置不为null，就会开启事务
public boolean transactionCapable() {
    return this.transactionIdPrefix != null;
}

kafkaTemplate.executeInTransaction(operations -> {
    for (int i = 0; i < 100; i++) {
        ListenableFuture<SendResult<String, Object>> listenableFuture = operations.send("twoTopic", String.valueOf(i));
        listenableFuture.addCallback(sendResult -> logger.info("success: partition is {}", sendResult.getRecordMetadata().partition()), throwable -> logger.info("failure : message is {}",throwable.getMessage()));
    }
    return "";
});

public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
    Assert.notNull(callback, "'callback' cannot be null");
    Assert.state(this.transactional, "Producer factory does not support transactions");
    Producer<K, V> producer = (Producer)this.producers.get();
    Assert.state(producer == null, "Nested calls to 'executeInTransaction' are not allowed");
    String transactionIdSuffix;
    if (this.producerFactory.isProducerPerConsumerPartition()) {
        transactionIdSuffix = TransactionSupport.getTransactionIdSuffix();
        TransactionSupport.clearTransactionIdSuffix();
    } else {
        transactionIdSuffix = null;
    }

    producer = this.producerFactory.createProducer();

    try {
        //开始事务
        producer.beginTransaction();
    } catch (Exception var15) {
        this.closeProducer(producer, false);
        throw var15;
    }

    this.producers.set(producer);

    Object var5;
    try {
        //真正进行的操作，即上面的发送100条消息
        Object result = callback.doInOperations(this);

        try {
            //提交事务
            producer.commitTransaction();
        } catch (Exception var12) {
            throw new KafkaTemplate.SkipAbortException(var12);
        }

        var5 = result;
    } catch (KafkaTemplate.SkipAbortException var13) {
        throw (RuntimeException)var13.getCause();
    } catch (Exception var14) {
        //放弃事务，类似于回滚
        producer.abortTransaction();
        throw var14;
    } finally {
        if (transactionIdSuffix != null) {
            TransactionSupport.setTransactionIdSuffix(transactionIdSuffix);
        }

        this.producers.remove();
        this.closeProducer(producer, false);
    }

    return var5;
}

@GetMapping("/sendTransaction/{message}")
@Transactional
public void sendTransaction(@PathVariable String message) {
    for (int i = 0; i < 100; i++) {
        ListenableFuture<SendResult<String, Object>> listenableFuture = kafkaTemplate.send("twoTopic", String.valueOf(i));
        listenableFuture.addCallback(sendResult -> logger.info("success: partition is {}", sendResult.getRecordMetadata().partition()), throwable -> logger.info("failure : message is {}", throwable.getMessage()));
    }
}

protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, TransactionAspectSupport.InvocationCallback invocation) throws Throwable {
    TransactionAttributeSource tas = this.getTransactionAttributeSource();
    TransactionAttribute txAttr = tas != null ? tas.getTransactionAttribute(method, targetClass) : null;
    PlatformTransactionManager tm = this.determineTransactionManager(txAttr);
    String joinpointIdentification = this.methodIdentification(method, targetClass, txAttr);
    Object result;
    if (txAttr != null && tm instanceof CallbackPreferringPlatformTransactionManager) {
        //该部分代码比较复杂，没有贴出，可以自行查看，其实就是我没看懂，所以没贴
    } else {
        TransactionAspectSupport.TransactionInfo txInfo = this.createTransactionIfNecessary(tm, txAttr, joinpointIdentification);

        try {
            //这里代表我们实际的方法，即加上@Transactional注解的方法，spring在注册bean时，会扫描包含@Transactional注解的方法，然后生成代理方法，即aop拦截
            result = invocation.proceedWithInvocation();
        } catch (Throwable var17) {
            //如果发生了报错，则回滚，调用PlatformTransactionManager的rollback()，最终调用producer.abortTransaction();
            this.completeTransactionAfterThrowing(txInfo, var17);
            throw var17;
        } finally {
            this.cleanupTransactionInfo(txInfo);
        }
        //如果执行成功，则提交，调用PlatformTransactionManager的commit()，最终调用producer.commitTransaction();
        this.commitTransactionAfterReturning(txInfo);
        return result;
    }
}