SpringBoot集成Kafka流处理

本文目录导读：

1. 项目依赖配置
2. 基础配置
3. 流处理配置类
4. 高级流处理示例
5. 生产者和消费者示例
6. 完整流处理应用示例
7. 测试配置
8. 最佳实践与注意事项

我来详细介绍Spring Boot集成Kafka流处理的完整实现方案。

项目依赖配置

Maven依赖

<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>
<!-- Kafka Streams -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
</dependency>
<!-- 可选：用于JSON处理 -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

Gradle依赖

implementation 'org.springframework.boot:spring-boot-starter-web'
implementation 'org.springframework.kafka:spring-kafka'
implementation 'org.apache.kafka:kafka-streams'
implementation 'com.fasterxml.jackson.core:jackson-databind'

基础配置

application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-stream-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    # Kafka Streams 配置
    streams:
      application-id: my-stream-app
      properties:
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        # 如果使用JSON，可以配置自定义Serde
        spring.json.trusted.packages: "*"

流处理配置类

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
import java.util.HashMap;
import java.util.Map;
@Configuration
@EnableKafka
@EnableKafkaStreams
public class KafkaStreamsConfig {
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        // 使用Avro或JSON Serde的配置
        // props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class);
        return new KafkaStreamsConfiguration(props);
    }
    @Bean
    public KStream<String, String> kStream(StreamsBuilder builder) {
        // 创建输入流
        KStream<String, String> stream = builder.stream("input-topic");
        // 处理逻辑
        stream
            .filter((key, value) -> value != null && !value.isEmpty())
            .mapValues(value -> value.toUpperCase())
            .to("output-topic");
        return stream;
    }
}

高级流处理示例

1 单词计数示例

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class WordCountStream {
    @Bean
    public KStream<String, String> wordCountStream(StreamsBuilder builder) {
        // 创建输入流
        KStream<String, String> textLines = builder.stream("text-input-topic");
        // 单词计数处理
        KTable<String, Long> wordCounts = textLines
            .flatMapValues(value -> 
                java.util.Arrays.asList(value.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word)
            .count(Materialized.as("word-counts-store"));
        // 输出结果
        wordCounts
            .toStream()
            .to("word-count-output-topic", 
                Produced.with(Serdes.String(), Serdes.Long()));
        return textLines;
    }
}

2 连接操作示例

@Configuration
public class JoinStreamsConfig {
    @Bean
    public KStream<String, String> joinStream(StreamsBuilder builder) {
        // 两个输入流
        KStream<String, String> leftStream = 
            builder.stream("left-topic");
        KStream<String, String> rightStream = 
            builder.stream("right-topic");
        // 基于key进行内连接，使用5分钟窗口
        KStream<String, String> joinedStream = leftStream
            .join(rightStream,
                (leftValue, rightValue) -> "Left: " + leftValue + ", Right: " + rightValue,
                JoinWindows.of(Duration.ofMinutes(5)),
                StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()));
        // 输出结果
        joinedStream.to("joined-output-topic");
        return joinedStream;
    }
}

3 自定义Serde

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.kafka.support.serializer.JsonSerde;
import java.io.IOException;
// 自定义对象
public class UserEvent {
    private String userId;
    private String eventType;
    private long timestamp;
    // getters and setters
}
// 使用Spring的JsonSerde
@Bean
public KStream<String, UserEvent> userEventStream(StreamsBuilder builder) {
    JsonSerde<UserEvent> userSerde = new JsonSerde<>(UserEvent.class);
    KStream<String, UserEvent> stream = 
        builder.stream("user-events-topic", 
            Consumed.with(Serdes.String(), userSerde));
    // 处理用户事件
    stream
        .filter((key, event) -> event.getEventType() != null)
        .to("processed-user-events", 
            Produced.with(Serdes.String(), userSerde));
    return stream;
}

生产者和消费者示例

生产者服务

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducerService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
    public void sendMessage(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message);
    }
}

消费者服务

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class KafkaConsumerService {
    @KafkaListener(topics = "output-topic", groupId = "my-stream-group")
    public void consume(String message) {
        System.out.println("Received message: " + message);
    }
    @KafkaListener(topics = "word-count-output-topic")
    public void consumeWordCount(String message) {
        System.out.println("Word count result: " + message);
    }
}

完整流处理应用示例

应用程序类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class KafkaStreamsApplication {
    public static void main(String[] args) {
        SpringApplication.run(KafkaStreamsApplication.class, args);
    }
}

REST控制器

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/api/kafka")
public class KafkaController {
    @Autowired
    private KafkaProducerService producerService;
    @PostMapping("/publish")
    public String publishMessage(@RequestParam String message) {
        producerService.sendMessage("input-topic", message);
        return "Message published: " + message;
    }
    @GetMapping("/health")
    public String health() {
        return "Kafka Streams Application is running!";
    }
}

测试配置

单元测试

import org.apache.kafka.streams.StreamsBuilder;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"input-topic", "output-topic"})
public class KafkaStreamsTest {
    @Autowired
    private StreamsBuilder streamsBuilder;
    @Test
    public void testStreamProcessing() {
        // 测试流处理逻辑
        // 可以使用TestInputTopic和TestOutputTopic进行测试
    }
}

最佳实践与注意事项

状态管理

@Bean
public KStream<String, String> statefulStream(StreamsBuilder builder) {
    // 使用状态存储
    StoreBuilder<KeyValueStore<String, Long>> storeBuilder = 
        Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("my-state-store"),
            Serdes.String(),
            Serdes.Long());
    builder.addStateStore(storeBuilder);
    return builder.stream("input-topic");
}

错误处理

@Bean
public KStream<String, String> errorHandlingStream(StreamsBuilder builder) {
    return builder
        .stream("input-topic")
        .transform(() -> new Transformer<String, String, KeyValue<String, String>>() {
            @Override
            public void init(ProcessorContext context) {
                // 初始化
            }
            @Override
            public KeyValue<String, String> transform(String key, String value) {
                try {
                    // 处理逻辑
                    return KeyValue.pair(key, processValue(value));
                } catch (Exception e) {
                    // 发送到死信队列
                    context.forward(key, "ERROR: " + e.getMessage());
                    return null;
                }
            }
            @Override
            public void close() {
                // 清理
            }
        });
}

这种集成方案提供了完整的Kafka流处理能力,可以根据实际业务需求进行调整和扩展。