Kafka 之事务消息

前言：

在分布式消息系统中，事务消息也是一个热门课题，在项目的实际业务场景中，如果用到事务消息的场景也不少见，那 Kafka 作为一个高性能的分布式消息中间件，同样也支持事务消息，本篇我们将对 Kafka 的事务消息展开讨论。

Kafka 系列文章传送门

Kafka 简介及核心概念讲解

Spring Boot 整合 Kafka 详解

Kafka @KafkaListener 注解的详解及使用

Kafka 客户端工具使用分享【offsetexplorer】

Kafka 之消息同步/异步发送

Kafka 之批量消息发送消费

Kafka 之消息广播消费

Kafka 之消息并发消费

Kafka 之顺序消息

事务消息的使用场景

事务消息的使用场景众多，这里我简单列举几个如下：

• 金融交易处理：在金融领域，每笔交易都必须具备原子性，确保不发生不一致或重复的交易，事务性消息可用于金融交易的场景，来保证交易的完整性。
• 订单处理：用户订单处理必须是完整的，需要保订单的创建、支付和发货不会出现问题，事务性消息可以用于订单场景，来保证整个流程的完整性。
• 。。。。等等等场景。

什么是 Kafka 的事务消息？

Kafka 事务性消息是一种机制，用于确保消息的可靠性传递和处理，与非事务性消息相比，它们在数据处理中提供了额外的保证，一旦消息被写入Kafka 集群，它们将被认为是已经处理，无论发生了什么，Kafka 的事务不同于 RocketMQ，RocketMQ 是保障本地事务与 RocketMQ 消息发送的事务一致性，而 Kafka 的事务消息主要是保障一次发送多条消息的事务一致性，发送的消息要么同时成功要么同时失败。

Kafka 事务消息的特性？

• 原子性：Kafka 事务性消息要么完全成功，要么完全失败，保障了消息不会被部分处理。
• 可靠性：Kafka 事务性消息一旦写入Kafka，它们将被视为已经处理，即使发生了应用程序或系统故障。
• 幂等性：Kafka 生产者可以配置为幂等，确保相同的消息不会被重复发送，对于 Kafka 事务消息通常会配置为幂等。
• Exactly Once 语义：仅一次，Kafka 事务性消息支持仅一次，即消息要么完全到达一次，要么不到达。

Kafka的消息传输保障
Kafka 的消息传输保障分为 3个层级：
at most once：至多一次，消息可能丢失，但绝不会重复。
at least once：最少一次，消息绝不丢失，但可能重复传输。
exactly once：恰好一次，每条消息肯定会被传输一次且仅传输一次。

Kafka 事务消息的使用

Kafka 事务消息的配置

Kafka 的事务消息需要对生产者和消费者进行一些配置来启用 Kafka 对事务消息的支持。

生产者配置

• acks：这个参数我们前面有见到过，是有关生产者接收到确认之后才认为消息发送成功的设置，对于事务性消息，通常将其设置为acks=all（表示需要等到消息写入到所有 ISR 同步副本中，设置为 -1 也是一样的效果），以确保消息在事务完全提交后才被视为成功发送。
• transactional.id：事务id，这是用于标识生产者实例的唯一ID，在配置文件中设置 transactional.id 是启用事务性消息的核心步骤。

消费者配置

• isolation.level：Kafka 消费者隔离级别的设置，对于事务性消息，通常将其设置为isolation.level=read_committed，确保只读取已经提交的事务消息。
• auto.offset.reset：这是消费者启动时从哪里开始读取消息的设置，通常将其设置为auto.offset.reset=earliest（如果分区下有提交的 offset 从提交的 offset 开始消费，如果没有提交的 offset，从头开始消费），以确保不会错过任何已提交的消息。

具体配置如下：

#消息应答
spring.kafka.producer.acks = -1
#事务id配置
spring.kafka.producer.transaction-id-prefix=my-transaction-
#读取已提交的消息
spring.kafka.consumer.isolation-level=read_committed

Kafka 事务消息案例演示

Producer 代码案例

对于生产者我们两种类型可以选择，使用 KafkaTemplate 进行事务消息发送和使用原生 API 进行事务消息发送。

使用 KafkaTemplate 完成事务消息发送代码如下：

package com.order.service.kafka.producer;import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;/*** @ClassName： TransactionalProducer* @Author: Author* @Date: 2024/10/22 19:22* @Description: 事务消息生产者*/
@Slf4j
@Component
public class TransactionalProducer {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;//发送事务消息（编程式）public void sendTransactionalMessage(int number) {try {kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, String, Object>() {@Overridepublic Object doInOperations(KafkaOperations<String, String> kafkaOperations) {kafkaOperations.send("transactional-topic", "我是第一条事务消息");if (number == 0) {//模拟异常int a = 1 / 0;}kafkaOperations.send("transactional-topic", "我是第二条事务消息");return true;}});} catch (Exception e) {e.printStackTrace();}}//发送事务消息（注解方式）@Transactional(rollbackFor = RuntimeException.class)public void sendTransactionalMessageByAnnotation(int number) {kafkaTemplate.send("transactional-topic", "我是第一条事务消息");if (number == 0) {//模拟异常int a = 1 / 0;}kafkaTemplate.send("transactional-topic", "我是第二条事务消息");}}

上述消息生产者代码中我们使用了两种方式实现，分别是编程式事务和注解式事务消息，大家根据自己的喜好来选择，两种方式都是可以的。

Consumer 代码案例

事务消息的 Consumer 代码没有什么特殊之处，我们使用 Manual ACK 即可。

package com.order.service.kafka.consumer;import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;import java.text.SimpleDateFormat;
import java.util.Date;/*** @ClassName： TransactionalConsumer* @Author: Author* @Date: 2024/10/22 19:22* @Description: 事务消息消费者*/
@Slf4j
@Component
public class TransactionalConsumer {@KafkaListener(id = "transactional-consumer",groupId = "transactional-consumer-groupId",topics = "transactional-topic",containerFactory = "myContainerFactory")public void listen(String message, Acknowledgment acknowledgment) {log.info("事务消息消费成功,消息内容:{}", message);//手动提交 ACKacknowledgment.acknowledge();}
}

事务消息结果验证

我们触发注解式事务消息发送，numer 传值 1，得到如下结果：

2024-11-05 19:48:32.649  INFO 20652 --- [-consumer-2-C-1] c.o.s.k.consumer.TransactionalConsumer   : 事务消息消费成功,消息内容:我是注解式事务第一条事务消息
2024-11-05 19:48:32.655  INFO 20652 --- [-consumer-2-C-1] c.o.s.k.consumer.TransactionalConsumer   : 事务消息消费成功,消息内容:我是注解式事务第二条事务消息

结果符合预期

2024-11-05 19:52:27.444  INFO 20652 --- [nio-8086-exec-8] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-my-transaction-0, transactionalId=my-transaction-0] Aborting incomplete transaction
2024-11-05 19:52:27.451 ERROR 20652 --- [y-transaction-0] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='我是注解式事务第一条事务消息' to topic transactional-topic:org.apache.kafka.common.KafkaException: Failing batch since transaction was abortedat org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:423) [kafka-clients-2.6.0.jar:na]at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:313) [kafka-clients-2.6.0.jar:na]at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240) [kafka-clients-2.6.0.jar:na]at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_121]2024-11-05 19:52:27.456 ERROR 20652 --- [nio-8086-exec-8] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is java.lang.ArithmeticException: / by zero] with root causejava.lang.ArithmeticException: / by zeroat com.order.service.kafka.producer.TransactionalProducer.sendTransactionalMessageByAnnotation(TransactionalProducer.java:49) ~[classes/:na]at com.order.service.kafka.producer.TransactionalProducer$$FastClassBySpringCGLIB$$dbc0d44d.invoke(<generated>) ~[classes/:na]at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) ~[spring-core-5.3.6.jar:5.3.6]at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:779) ~[spring-aop-5.3.6.jar:5.3.6]at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.6.jar:5.3.6]at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:750) ~[spring-aop-5.3.6.jar:5.3.6]at org.springframework.transaction.interceptor.TransactionInterceptor$1.proceedWithInvocation(TransactionInterceptor.java:123) ~[spring-tx-5.3.6.jar:5.3.6]at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:388) ~[spring-tx-5.3.6.jar:5.3.6]at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119) ~[spring-tx-5.3.6.jar:5.3.6]at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.6.jar:5.3.6]at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:750) ~[spring-aop-5.3.6.jar:5.3.6]at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:692) ~[spring-aop-5.3.6.jar:5.3.6]at com.order.service.kafka.producer.TransactionalProducer$$EnhancerBySpringCGLIB$$e4350198.sendTransactionalMessageByAnnotation(<generated>) ~[classes/:na]at com.order.service.controller.KafkaController.sendTransactionalMessageByAnnotation(KafkaController.java:96) ~[classes/:na]at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_121]at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_121]at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_121]at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_121]at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:197) ~[spring-web-5.3.6.jar:5.3.6]at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:141) ~[spring-web-5.3.6.jar:5.3.6]at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:106) ~[spring-webmvc-5.3.6.jar:5.3.6]at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:894) ~[spring-webmvc-5.3.6.jar:5.3.6]at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:808) ~[spring-webmvc-5.3.6.jar:5.3.6]at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87) ~[spring-webmvc-5.3.6.jar:5.3.6]at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1060) ~[spring-webmvc-5.3.6.jar:5.3.6]at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:962) ~[spring-webmvc-5.3.6.jar:5.3.6]at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006) ~[spring-webmvc-5.3.6.jar:5.3.6]at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898) ~[spring-webmvc-5.3.6.jar:5.3.6]at javax.servlet.http.HttpServlet.service(HttpServlet.java:626) ~[tomcat-embed-core-9.0.45.jar:4.0.FR]at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883) ~[spring-webmvc-5.3.6.jar:5.3.6]at javax.servlet.http.HttpServlet.service(HttpServlet.java:733) ~[tomcat-embed-core-9.0.45.jar:4.0.FR]at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:227) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53) ~[tomcat-embed-websocket-9.0.45.jar:9.0.45]at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100) ~[spring-web-5.3.6.jar:5.3.6]at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) ~[spring-web-5.3.6.jar:5.3.6]at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93) ~[spring-web-5.3.6.jar:5.3.6]at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) ~[spring-web-5.3.6.jar:5.3.6]at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.springframework.boot.actuate.metrics.web.servlet.WebMvcMetricsFilter.doFilterInternal(WebMvcMetricsFilter.java:93) ~[spring-boot-actuator-2.4.5.jar:2.4.5]at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) ~[spring-web-5.3.6.jar:5.3.6]at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201) ~[spring-web-5.3.6.jar:5.3.6]at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) ~[spring-web-5.3.6.jar:5.3.6]at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:202) ~[tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:97) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:542) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:143) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:78) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:357) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:374) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:65) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:893) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1707) [tomcat-embed-core-9.0.45.jar:9.0.45]at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49) [tomcat-embed-core-9.0.45.jar:9.0.45]at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_121]at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_121]at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61) [tomcat-embed-core-9.0.45.jar:9.0.45]at java.lang.Thread.run(Thread.java:745) [na:1.8.0_121]

从日志结果我们看到消费者没有消费到任何一条消息（这里我们是在第一条消息发送结束后模拟的异常），结果符合预期。

编程式事务消息这里就不在验证了，有兴趣的朋友可以自己去测试一下。

Producer 使用原生 API 发送事务消息

使用原生 API 发送事务消息，需要有一个 KafkaProducer 对象，我这里使用注入 Bean 的方式，具体如下：

@Bean
public KafkaProducer<String, String> kafkaProducer() {return new KafkaProducer<>(getMyKafkaProps());
}

有了 KafkaProducer 只有我们调用其 API 即可完成事务消息发送，KafkaProducer 有几个重要的 API 如下：

//初始化事务
public void initTransactions();//开启事务
public void beginTransaction() throws ProducerFencedException ;//在事务内提交已经消费的偏移量
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException ;//提交事务
public void commitTransaction() throws ProducerFencedException;//丢弃事务
public void abortTransaction() throws ProducerFencedException ;

使用原生 API 进行事务消息发送代码如下：

package com.order.service.kafka.producer;import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;/*** @ClassName： NativeTransactionalProducer* @Author: Author* @Date: 2024/10/22 19:22* @Description: 事务消息生产者 原生API*/
@Slf4j
@Component
public class NativeTransactionalProducer {@Autowiredprivate KafkaProducer<String, String> kafkaProducer;//发送事务消息（原生API）public void sendTransactionalMessageByNativeApi(int number) {try {//初始化一个事务kafkaProducer.initTransactions();//开启事务kafkaProducer.beginTransaction();//发送消息kafkaProducer.send(new ProducerRecord<>("transactional-topic", "我是原生API发送出的第一条事务消息"));if (number == 0) {//模拟异常int a = 1 / 0;}kafkaProducer.send(new ProducerRecord<>("transactional-topic", "我是原生API发送出的第二条事务消息"));//提交事务消息kafkaProducer.commitTransaction();} catch (ProducerFencedException e) {//关闭资源kafkaProducer.close();} finally {//关闭资源kafkaProducer.close();}}}

原生 API 发送事务消息结果验证

我们触发原生 API 发送事务消息发送，numer 传值 1，得到如下结果：

2024-11-05 19:48:32.649  INFO 20652 --- [-consumer-2-C-1] c.o.s.k.consumer.TransactionalConsumer   : 事务消息消费成功,消息内容:我是注解式事务第一条事务消息
2024-11-05 19:48:32.655  INFO 20652 --- [-consumer-2-C-1] c.o.s.k.consumer.TransactionalConsumer   : 事务消息消费成功,消息内容:我是注解式事务第二条事务消息

结果符合预期，异常的情况我这里就不演示了，有兴趣的朋友可以自己去验证。

事务消息注意事项

只要开启了事务消息功能，不管是 KafkaProducer 还是 KafkaTemplate 发送的所有消息，都必须处于 Kafka 事务之中，否则会抛出如下异常：

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 recordat org.springframework.util.Assert.state(Assert.java:76)at org.springframework.kafka.core.KafkaTemplate.getTheProducer(KafkaTemplate.java:640)at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:552)at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:363)at com.order.service.kafka.producer.ManualKafkaProducer.sendManualMessage(ManualKafkaProducer.java:34)at com.order.service.controller.KafkaController.sendManualMessage(KafkaController.java:87)at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)at java.lang.reflect.Method.invoke(Method.java:498)at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:197)at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:141)at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:106)at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:894)at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:808)at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1060)at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:962)at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006)at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898)at javax.servlet.http.HttpServlet.service(HttpServlet.java:626)at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883)at javax.servlet.http.HttpServlet.service(HttpServlet.java:733)at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:227)at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53)at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100)at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93)at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)at org.springframework.boot.actuate.metrics.web.servlet.WebMvcMetricsFilter.doFilterInternal(WebMvcMetricsFilter.java:93)at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201)at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:202)at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:97)at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:542)at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:143)at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92)at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:78)at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:357)at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:374)at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:65)at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:893)at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1707)at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)at java.lang.Thread.run(Thread.java:745)

如果业务中，存在需要事务的情况，也存在不需要事务的情况，那么则需要分别定义两个 KafkaTemplate（Kafka Producer），按需使用。

总结：本篇分享了 Kafka 的事务消息的使用，可以感受到 Kafka 的事务消息和 RocketMQ 是完全不一样的，希望可以帮助到需要用到的伙伴们。

如有不正确的地方欢迎各位指出纠正。