springboot入门14 – Kafka应用 - JOYK Joy of Geek, Geek News, Link all geek

这几天优化了一下之前写的一个springboot kafka组件。比较起原生的spring-kafka来，我希望能够简化kafka的使用，可以更聚焦于具体的消息处理逻辑。

接下来的内容是这个组件的用法。

这个组件已经提交到了maven中央仓库，可以直接通过依赖的形式引入：

<groupId>org.chobit.spring</groupId>

<artifactId>kafka-spring-boot-starter</artifactId>

</dependency>

0.2.2是这两天刚发布的一个版本。

消费者Processor

kafka-spring-boot-starter这个组件已经完成了kafka消费者的主要功能。

对于开发者来说，可以不必关注 KafkaConsumer的创建，只需要实现 Processor接口并注入到容器中即可。

下面是一个简单的示例：

@Component("zhyyy")

public class MyProcessor implements Processor<String, String> {

@Override

public void process(ConsumerRecords<String, String> records) {

for (ConsumerRecord<String, String> r : records) {

String json = r.value();

System.out.println(json);

如示例中，通过 @Component注解完成了 Processor实现类的实例的注入，并为注入的Bean提供了一个名称：zhyyy。记住这个名称，在之后的配置文件中会用到。

使用生产者

kafka-spring-boot-starter会根据配置主动创建 KafkaProducer。开发使用时可以直接从容器中获取 ProducerTemplate实例来发送消息：

@Autowired

private ProducerTemplate<?, ?> producerTemplate;

如果写入kafka的消息的key和value的序列化方案采用的都是默认的字符串（反）序列化方案（ StringDeserializer和 StringSerializer），可以使用 StringProducerTemplate实例：

@Autowired

private StringProducerTemplate producerTemplate;

发送消息时酌情调用不同的 send()方法：

void send(String topic, V value){...}

void send(String topic, K key, V value){...}

void send(String topic, V value, Callback callback){...}

void send(String topic, K key, V value, Callback callback) {...}

下面是一个最简单的配置：

kafka:

config:

test-group00:

bootstrap-servers: kafka1,kafka2,kafka3

topics: test-topic1

consumer:

processor: zhyyy

count: 4

如上配置中：

test-group00既是配置项的ID，也是消费组ID
bootstrap-servers我想不需要多做解释。
topics对应的是一个数组结构，也可以写作 [test-topic1]或 [test-topic1,test-topic2]，即支持同一个kafka集群上多个类似topic的统一处理
consumer是消费者相关配置，processor对应的是 Processor实现类的Bean名称，count标识的是应用内消费线程的数量

默认的序列化方案采用的是字符串序列化方案。

虽然在配置中没有体现，但kafka-spring-boot-starter组件会基于已有的信息创建 KafkaProducer，使用时可以通过 ProducerTemplate执行消息发送。

比较完整的配置是这样子的：

kafka:

common:

consumer:

prop1: value1

prop2: value2

producer:

prop3: value3

prop4: value4

config:

test-group00:

bootstrapServers: kafka1,kafka2,kafka3

topics: test-topic

consumer:

autoOffsetReset: latest

processor: zhyyy

count: 4

keyDeserializer: org.apache.kafka.common.serialization.StringDeserializer

valueDeserializer: org.apache.kafka.common.serialization.StringDeserializer

props:

prop1: value1

prop2: value2

producer:

keySerializer: org.apache.kafka.common.serialization.StringSerializer

valueSerializer: org.apache.kafka.common.serialization.StringSerializer

props:

prop3: value3

prop4: value4

test-group02:

bootstrapServers: kafka4,kafka5,kafka6

topics: test-topic2

consumer:

autoOffsetReset: latest

processor: zhyyy

其中common模块下是一些通用的配置，config模块下则是一或多组具体的配置（这里是两组）。common下的配置会被config下的配置覆盖。

此外还独立出来了一些常用的配置项，如autoOffsetReset，keyDeserializer等，以便在使用时进行配置。

示例应用在 github / spring-boot-kafka。

kafka-spring-boot-starter这个组件的源码也在 github 。如果有定制化的需求可以据此进行调整。

springboot入门14 – Kafka应用

消费者Processor

使用生产者

Recommend

【Kafka】《Kafka权威指南》入门原荐

全网最通俗易懂的Kafka入门

全网最通俗易懂的Kafka入门！

Kafka入门（2）：消费与位移

Akka入门系列(八)：akka kafka Consumer

Akka入门系列(七)：akka kafka Producer

springboot入门13 – 多CacheManager应用

SpringBoot使用Testcontainers+Avro消息测试Kafka消费者

SpringBoot 整合 Kafka 实现数据高吞吐

Kafka Tutorial - Producer and Consumer using SpringBoot

About Joyk