概述
Kafka的事务不同于Rocketmq，Rocketmq是保障本地事务(比如数据库)与mq消息发送的事务一致性，Kafka的事务主要是保障一次发送多条消息的事务一致性(要么同时成功要么同时失败)。
一般在kafka的流式计算场景用得多一点，比如，kafka需要对一个topic里的消息做不同的流式计算处理，处理完分别发到不同的topic里，这些topic分别被不同的下游系统消费(比如hbase，redis，es等)，这种我们肯定希望系统发送到多个topic的数据保持事务一致性。
Kafka要实现类似Rocketmq的分布式事务需要额外开发功能。
官方文档： http://kafka.apache.org/24/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html
这个功能比较鸡肋，大家看着用哈 ,它保证不了不同介质的数据一致性。
官方示例
From Kafka 0.11, the KafkaProducer supports two additional modes: the idempotent producer and the transactional producer.
原生的API操作，请查看文档，这里我们来看下使用Spring kafka如何实现事务消息。
Code （原生API）
@Test
public void testT(){ // 正常的
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.126.140:9092");
props.put("transactional.id", "my-transactional-id");
Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
producer.initTransactions();
try {
producer.beginTransaction();
for (int i = 0; i < 100; i++)
producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
// We can't recover from these exceptions, so our only option is to close the producer and exit.
producer.close();
} catch (KafkaException e) {
// For all other exceptions, just abort the transaction and try again.
producer.abortTransaction();
}
producer.close();
}
@Test
public void testT2(){ // 测试异常情况
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.126.140:9092");
props.put("transactional.id", "my-transactional-id");
Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
producer.initTransactions();
try {
producer.beginTransaction();
for (int i = 0; i < 100; i++){
producer.send(new ProducerRecord<>("my-topic2", Integer.toString(i), Integer.toString(i)));
if (i == 50) {
throw new RuntimeException("MOCK Exception");
}
}
producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
// We can't recover from these exceptions, so our only option is to close the producer and exit.
producer.close();
} catch (KafkaException e) {
// For all other exceptions, just abort the transaction and try again.
producer.abortTransaction();
}
producer.close();
}
看看数据
可以看到 入了一部分，只是这里进入的数据也无法被消费。
Code （Spring Kafka）
POM依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- 引入 Spring-Kafka 依赖 -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
配置文件
spring:
# Kafka 配置项，对应 KafkaProperties 配置类
kafka:
bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔
# Kafka Producer 配置项
producer:
acks: all # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。
retries: 3 # 发送失败时，重试发送的次数
key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化
transaction-id-prefix: artisan. # 事务编号前缀
# Kafka Consumer 配置项
consumer:
auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
spring:
json:
trusted:
packages: com.artisan.springkafka.domain
isolation-level: read_committed # 读取已提交的消息
# Kafka Consumer Listener 监听器配置
listener:
missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错
logging:
level:
org:
springframework:
kafka: ERROR # spring-kafka
apache:
kafka: ERROR # kafka
-
spring.kafka.producer.acks配置为all，Kafka 的事务消息需要基于幂等性来实现，必须保证所有节点都写入成功，否则的话启动时会抛出Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence
-
事务编号前缀属性设置
transaction-id-prefix， 需要保证相同应用配置相同，不同应用配置不同。 How to choose Kafka transaction id for several applications, hosted in Kubernetes? -
spring.kafka.consumer.properties.isolation.level设置为 read_committed ，Consumer 仅读取已提交的消息， 否则不生效
生产者
package com.artisan.springkafka.producer;
import com.artisan.springkafka.constants.TOPIC;
import com.artisan.springkafka.domain.MessageMock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import java.util.Random;
import java.util.concurrent.ExecutionException;
/**
* @author 小工匠
* @version 1.0
* @description: TODO
* @date 2021/2/17 22:25
* @mark: show me the code , change the world
*/
@Component
public class ArtisanProducerMock {
private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private KafkaTemplate<Object,Object> kafkaTemplate ;
public String testTransaction(Runnable runnable){
return kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<Object, Object, String>() {
@Override
public String doInOperations(KafkaOperations<Object, Object> operations) throws RuntimeException {
for (int i = 1; i <= 10; i++) {
// 用于测试 消息是否在同一个事务中
if (i == 7 ) {
throw new RuntimeException("MOCK ERROR , TEST Tranasction");
}
// 模拟发送的消息
Integer id = new Random().nextInt(100);
MessageMock messageMock = new MessageMock(id,"messageSendByAsync-" + id);
SendResult<Object, Object> sendResult = null;
try {
sendResult = operations.send(TOPIC.TOPIC, messageMock).get();
} catch ( Exception e) {
logger.error("Error {}", e);
}
logger.info( i+ "-[doInOperations][发送数据：[{}] 发送结果：[{}]]", messageMock, sendResult);
// 本地业务逻辑...
runnable.run();
}
// 返回结果
return "OJ8K";
}
});
}
}
我们模拟发送10条消息，第7条的时候抛出异常，观察消费者是否能消费前面已经发送的6条 ，如果能消费，那肯定不符合和预期。 因为Kafka的事务主要是保障一次发送多条消息的事务一致性(要么同时成功要么同时失败)。
调用 kafkaTemplate#executeInTransaction(OperationsCallback<K, V, T> callback) 模板方法，实现在 Kafka 事务中，执行自定义 KafkaOperations.OperationsCallback 操作。
-
executeInTransaction(...)方法中，可以通过 KafkaOperations 来执行发送消息等 Kafka 相关的操作，当然了也可以执行自己的业务逻辑，比如 runnable参数，用于表示本地业务逻辑 -
executeInTransaction(...)方法的开始，会自动动创建 Kafka 的事务，然后执行KafkaOperations 的逻辑。成功，则提交 Kafka 事务；失败，则回滚 Kafka 事务。
注意事项
如果 Kafka Producer 开启了事务的功能，则所有发送的消息，都必须处于 Kafka 事务之中，否则会抛出 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
假设业务中，即存在需要事务的情况，也存在不需要事务的情况，那么则需要分别定义两个 KafkaTemplate（Kafka Producer）
消费者
package com.artisan.springkafka.consumer;
import com.artisan.springkafka.constants.TOPIC;
import com.artisan.springkafka.domain.MessageMock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
/**
* @author 小工匠
* @version 1.0
* @description: TODO
* @date 2021/2/17 22:33
* @mark: show me the code , change the world
*/
@Component
public class ArtisanCosumerMock {
private Logger logger = LoggerFactory.getLogger(getClass());
private static final String CONSUMER_GROUP_PREFIX = "MANUAL_ACK_" ;
@KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)
public void onMessage(MessageMock messageMock) {
logger.info("【接受到消息][线程:{} 消息内容：{}]", Thread.currentThread().getName(), messageMock);
}
}
单元测试
package com.artisan.springkafka.produceTest;
import com.artisan.springkafka.SpringkafkaApplication;
import com.artisan.springkafka.producer.ArtisanProducerMock;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.support.SendResult;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
/**
* @author 小工匠
* * @version 1.0
* @description: TODO
* @date 2021/2/17 22:40
* @mark: show me the code , change the world
*/
@RunWith(SpringRunner.class)
@SpringBootTest(classes = SpringkafkaApplication.class)
public class ProduceMockTest {
private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private ArtisanProducerMock artisanProducerMock;
@Autowired
ArtisanProducerMock producerMock;
@Test
public void testAsynSend() throws ExecutionException, InterruptedException {
logger.info("开始发送");
producerMock.testTransaction(() -> {
logger.info(" mock doing bussiness ");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.info("bussiness over ");
});
// 阻塞等待，保证消费
new CountDownLatch(1).await();
}
}
测试结果
....
....
....
. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v2.4.1)
2021-02-20 01:35:44.452 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : Starting ProduceMockTest using Java 1.8.0_261 on LAPTOP-JF3RBRRJ with PID 12108 (started by artisan in D:\IdeaProjects\boot2\springkafkaTransaction)
2021-02-20 01:35:44.456 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : No active profile set, falling back to default profiles: default
2021-02-20 01:35:45.832 INFO 12108 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'applicationTaskExecutor'
2021-02-20 01:35:46.811 INFO 12108 --- [ main] o.s.s.c.ThreadPoolTaskScheduler : Initializing ExecutorService
2021-02-20 01:35:46.827 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : Started ProduceMockTest in 2.77 seconds (JVM running for 3.55)
2021-02-20 01:35:47.021 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : 开始发送
2021-02-20 01:35:47.298 INFO 12108 --- [ main] c.a.s.producer.ArtisanProducerMock : 1-[doInOperations][发送数据：[MessageMock{id=2, name='messageSendByAsync-2'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=AC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=2, name='messageSendByAsync-2'}, timestamp=null), recordMetadata=AC-0@16]]]
2021-02-20 01:35:47.298 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 01:35:48.302 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 01:35:48.305 INFO 12108 --- [ main] c.a.s.producer.ArtisanProducerMock : 2-[doInOperations][发送数据：[MessageMock{id=2, name='messageSendByAsync-2'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=AC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=2, name='messageSendByAsync-2'}, timestamp=null), recordMetadata=AC-0@17]]]
2021-02-20 01:35:48.305 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 01:35:49.308 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 01:35:49.308 INFO 12108 --- [ main] c.a.s.producer.ArtisanProducerMock : 3-[doInOperations][发送数据：[MessageMock{id=36, name='messageSendByAsync-36'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=AC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=36, name='messageSendByAsync-36'}, timestamp=null), recordMetadata=AC-0@18]]]
2021-02-20 01:35:49.308 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 01:35:50.314 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 01:35:50.314 INFO 12108 --- [ main] c.a.s.producer.ArtisanProducerMock : 4-[doInOperations][发送数据：[MessageMock{id=19, name='messageSendByAsync-19'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=AC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=19, name='messageSendByAsync-19'}, timestamp=null), recordMetadata=AC-0@19]]]
2021-02-20 01:35:50.318 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 01:35:51.321 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 01:35:51.321 INFO 12108 --- [ main] c.a.s.producer.ArtisanProducerMock : 5-[doInOperations][发送数据：[MessageMock{id=29, name='messageSendByAsync-29'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=AC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=29, name='messageSendByAsync-29'}, timestamp=null), recordMetadata=AC-0@20]]]
2021-02-20 01:35:51.325 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 01:35:52.326 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 01:35:52.326 INFO 12108 --- [ main] c.a.s.producer.ArtisanProducerMock : 6-[doInOperations][发送数据：[MessageMock{id=45, name='messageSendByAsync-45'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=AC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=45, name='messageSendByAsync-45'}, timestamp=null), recordMetadata=AC-0@21]]]
2021-02-20 01:35:52.326 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 01:35:53.326 INFO 12108 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
java.lang.RuntimeException: MOCK ERROR , TEST Tranasction
at com.artisan.springkafka.producer.ArtisanProducerMock$1.doInOperations(ArtisanProducerMock.java:42)
at com.artisan.springkafka.producer.ArtisanProducerMock$1.doInOperations(ArtisanProducerMock.java:34)
at org.springframework.kafka.core.KafkaTemplate.executeInTransaction(KafkaTemplate.java:467)
at com.artisan.springkafka.producer.ArtisanProducerMock.testTransaction(ArtisanProducerMock.java:34)
at com.artisan.springkafka.produceTest.ProduceMockTest.testAsynSend(ProduceMockTest.java:45)
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.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:74)
at org.springframework.test.context.junit4.statements.RunAfterTestExecutionCallbacks.evaluate(RunAfterTestExecutionCallbacks.java:84)
at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75)
at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86)
at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:251)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33)
at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:220)
at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:53)
2021-02-20 01:35:53.346 INFO 12108 --- [ntainer#0-0-C-1] o.s.s.c.ThreadPoolTaskScheduler : Shutting down ExecutorService
2021-02-20 01:35:53.357 INFO 12108 --- [extShutdownHook] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'applicationTaskExecutor'
Process finished with exit code -1
可以看到，有异常了，消费者未的消费到消息。
那我们来个成功的看看嘛
2021-02-20 10:10:46.103 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : 开始发送
2021-02-20 10:10:46.429 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 1-[doInOperations][发送数据：[MessageMock{id=44, name='messageSendByAsync-44'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=44, name='messageSendByAsync-44'}, timestamp=null), recordMetadata=OOO_TOIPC-0@7]]]
2021-02-20 10:10:46.429 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:47.430 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:47.430 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 2-[doInOperations][发送数据：[MessageMock{id=76, name='messageSendByAsync-76'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=76, name='messageSendByAsync-76'}, timestamp=null), recordMetadata=OOO_TOIPC-0@8]]]
2021-02-20 10:10:47.430 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:48.430 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:48.431 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 3-[doInOperations][发送数据：[MessageMock{id=2, name='messageSendByAsync-2'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=2, name='messageSendByAsync-2'}, timestamp=null), recordMetadata=OOO_TOIPC-0@9]]]
2021-02-20 10:10:48.431 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:49.434 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:49.438 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 4-[doInOperations][发送数据：[MessageMock{id=34, name='messageSendByAsync-34'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=34, name='messageSendByAsync-34'}, timestamp=null), recordMetadata=OOO_TOIPC-0@10]]]
2021-02-20 10:10:49.438 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:50.440 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:50.440 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 5-[doInOperations][发送数据：[MessageMock{id=41, name='messageSendByAsync-41'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=41, name='messageSendByAsync-41'}, timestamp=null), recordMetadata=OOO_TOIPC-0@11]]]
2021-02-20 10:10:50.444 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:51.446 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:51.446 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 6-[doInOperations][发送数据：[MessageMock{id=29, name='messageSendByAsync-29'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=29, name='messageSendByAsync-29'}, timestamp=null), recordMetadata=OOO_TOIPC-0@12]]]
2021-02-20 10:10:51.446 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:52.447 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:52.447 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 7-[doInOperations][发送数据：[MessageMock{id=49, name='messageSendByAsync-49'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=49, name='messageSendByAsync-49'}, timestamp=null), recordMetadata=OOO_TOIPC-0@13]]]
2021-02-20 10:10:52.447 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:53.450 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:53.450 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 8-[doInOperations][发送数据：[MessageMock{id=12, name='messageSendByAsync-12'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=12, name='messageSendByAsync-12'}, timestamp=null), recordMetadata=OOO_TOIPC-0@14]]]
2021-02-20 10:10:53.450 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:54.450 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:54.450 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 9-[doInOperations][发送数据：[MessageMock{id=15, name='messageSendByAsync-15'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=15, name='messageSendByAsync-15'}, timestamp=null), recordMetadata=OOO_TOIPC-0@15]]]
2021-02-20 10:10:54.450 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:55.454 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:55.454 INFO 25272 --- [ main] c.a.s.producer.ArtisanProducerMock : 10-[doInOperations][发送数据：[MessageMock{id=25, name='messageSendByAsync-25'}] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=OOO_TOIPC, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 114, 116, 105, 115, 97, 110, 46, 115, 112, 114, 105, 110, 103, 107, 97, 102, 107, 97, 46, 100, 111, 109, 97, 105, 110, 46, 77, 101, 115, 115, 97, 103, 101, 77, 111, 99, 107])], isReadOnly = true), key=null, value=MessageMock{id=25, name='messageSendByAsync-25'}, timestamp=null), recordMetadata=OOO_TOIPC-0@16]]]
2021-02-20 10:10:55.458 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : mock doing bussiness
2021-02-20 10:10:56.460 INFO 25272 --- [ main] c.a.s.produceTest.ProduceMockTest : bussiness over
2021-02-20 10:10:56.625 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=44, name='messageSendByAsync-44'}]
2021-02-20 10:10:56.737 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=76, name='messageSendByAsync-76'}]
2021-02-20 10:10:56.846 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=2, name='messageSendByAsync-2'}]
2021-02-20 10:10:56.962 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=34, name='messageSendByAsync-34'}]
2021-02-20 10:10:56.970 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=41, name='messageSendByAsync-41'}]
2021-02-20 10:10:57.074 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=29, name='messageSendByAsync-29'}]
2021-02-20 10:10:57.082 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=49, name='messageSendByAsync-49'}]
2021-02-20 10:10:57.090 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=12, name='messageSendByAsync-12'}]
2021-02-20 10:10:57.094 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=15, name='messageSendByAsync-15'}]
2021-02-20 10:10:57.101 INFO 25272 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=25, name='messageSendByAsync-25'}]
懂了么，老兄 ~
我们继续看下数据 （新跑的数据，和日志有出入）
源码地址
https://github.com/yangshangwei/boot2/tree/master/springkafkaTransaction
本文标题：Apache Kafka-事务消息的支持与实现（本地事务）
本文链接：https://blog.quwenai.cn/post/6883.html
版权声明：本文不使用任何协议授权，您可以任何形式自由转载或使用。
还没有评论，来说两句吧...