微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦!

Apache Kafka-事务消息的支持与实现本地事务

文章目录


在这里插入图片描述


概述

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.SpringJUnit4Cla***unner.runchild(SpringJUnit4Cla***unner.java:251)
	at org.springframework.test.context.junit4.SpringJUnit4Cla***unner.runchild(SpringJUnit4Cla***unner.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.SpringJUnit4Cla***unner.run(SpringJUnit4Cla***unner.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

在这里插入图片描述


版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐