springboot 整合kafka最强指南_springboot kafkatemplate

首先部署好kafka和Zookeeper 可以参考我的其他博文 在Windows上搭建Kafka 或者Windows环境下Zookeeper安装配置

一、kafka与Spring版本对应的关系表

先看一下我的基础环境

springboot version：springboot2.1.3.RELEASE

kafka version ：2.2.4.RELEASE

集成之前先放上一张官网的kafka与Spring版本对应的关系表，后续如遇到消息不能正常发送接收就请排查一下是不是版本问题

官网链接：https://spring.io/projects/spring-kafka

二、新建SpringBoot项目

创建springboot项目我就略了。。

1.引入pom依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.1.3.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.kafkaDemo</groupId>
	<artifactId>kafkaDemo</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>kafkaDemo</name>
	<description>kafkaDemoproject for Spring Boot</description>
	<properties>
		<java.version>1.8</java.version>
	</properties>
	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>

		<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>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.junit.jupiter</groupId>
			<artifactId>junit-jupiter-api</artifactId>
			<version>RELEASE</version>
			<scope>test</scope>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

2.application.yml配置

spring:
  kafka:
    ###########【Kafka集群】###########(kafka 默认的端口号为9092)
    bootstrap-servers: 127.0.0.1:9092
    producer:
      # 重试次数
      retries: 0
      # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
      acks: 1
      # 批量大小
      batch-size: 16384
          # 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
          # linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了

      # 生产端缓冲区大小
      buffer-memory: 33554432
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      # 默认的消费组ID
      group-id: defaultConsumerGroup
      # 是否自动提交offset
      enable-auto-commit: true
      auto-commit-interval: 100ms
      # 当kafka中没有初始offset或offset超出范围时将自动重置offset
      # earliest:重置为分区中最小的offset;
      # latest:重置为分区中最新的offset(消费分区中新产生的数据);
      # none:只要有一个分区不存在已提交的offset,就抛出异常;
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

server:
  port: 8085
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

三、开始愉快的玩耍kafka

1、初始化topic

我们可以手动用命令创建topic，也可以不手动，因为在执行代码kafkaTemplate.send(“topic1”, normalMessage)发送消息时，kafka会帮我们自动完成topic的创建工作，但这种情况下创建的topic默认只有一个分区，分区也没有副本。所以，我们可以在项目中新建一个配置类专门用来初始化topic

代码如下：

@Configuration
public class KafkaConfiguration {
    // 创建一个名为testtopic的Topic并设置分区数为8，分区副本数为2
    @Bean
    public NewTopic createTopic() {
        return new NewTopic("testTopic",8, (short) 2 );
    }
​
     // 如果要修改分区数，只需修改配置值重启项目即可
    // 修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小
    @Bean
    public NewTopic updateTopic() {
        return new NewTopic("testTopic",10, (short) 2 );
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2.普通消息的生产和消费

2.1消息生产者

@RestController
@RequestMapping("/kafka")
public class KafkaProducer {

        @Autowired
        private KafkaTemplate<String, Object> kafkaTemplate;
        // 发送消息
        @GetMapping("/normal/{message}")
        public void sendMessage1(@PathVariable("message") String normalMessage) {
            kafkaTemplate.send("topic1", normalMessage);
        }
}
1
2
3
4
5
6
7
8
9
10
11
12

2.2消息消费者

@Component
public class KafkaConsumer {
    // 消费监听
    @KafkaListener(topics = {"topic1"})
    public void onMessage1(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("普通消息消费："+record.topic()+"-"+record.partition()+"-"+record.value());
    }
}
1
2
3
4
5
6
7
8
9

2.3测试

上面示例创建了一个生产者，发送消息到topic1，消费者监听topic1消费消息。监听器用@KafkaListener注解，topics表示监听的topic，支持同时监听多个，用英文逗号分隔。

启动项目，postman访问方法生产消息 ：http://192.168.100.9:8085/kafka/normal/11111

消费者消费结果，可以看到监听器消费成功！

3.生产者进阶

上面的例子我们实现了普通消息的生产和消费，我们再来点高级的

3.1带回调的消息生产者

kafkaTemplate提供了一个回调方法addCallback，我们可以在回调方法中监控消息是否发送成功 或 失败时做补偿处理，有两种写法

第一种

        @GetMapping("/callbackOne/{message}")
        public void sendMessage2(@PathVariable("message") String callbackMessage) {
                kafkaTemplate.send("topic1", callbackMessage).addCallback(success -> {
                        // 消息发送到的topic
                        String topic = success.getRecordMetadata().topic();
                        // 消息发送到的分区
                        int partition = success.getRecordMetadata().partition();
                        // 消息在分区内的offset
                        long offset = success.getRecordMetadata().offset();
                        System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
                }, failure -> {
                        System.out.println("发送消息失败:" + failure.getMessage());
                });
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

第二种

        @GetMapping("/callbackTwo/{message}")
        public void sendMessage3(@PathVariable("message") String callbackMessage) {
                kafkaTemplate.send("topic1", callbackMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                        @Override
                        public void onFailure(Throwable ex) {
                                System.out.println("发送消息失败："+ex.getMessage());
                        }

                        @Override
                        public void onSuccess(SendResult<String, Object> result) {
                                System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                                        + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
                        }
                });
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3.2事务消息

我们在本地事务处理成功后发送发送消息

        @GetMapping("/transaction")
        public void sendMessage4(){
                // 声明事务：后面报错消息不会发出去
                kafkaTemplate.executeInTransaction(operations -> {
                        operations.send("topic1","test , this is Transaction");
                        throw new RuntimeException("fail");
                });
​
                // 不声明事务：后面报错但前面消息已经发送成功了
                kafkaTemplate.send("topic1","test,this is not Transaction");
                throw new RuntimeException("fail");
        }
1
2
3
4
5
6
7
8
9
10
11
12

测试看看

我们看到事务消息直接报错

无法发送

非事务消息发送也报错 但是消息发出去了