SpringCloud使用Kafka Streams实现实时数据处理

<dependencies>
    <!-- Spring Boot -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
 
    <!-- Spring Kafka -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
 
    <!-- Kafka Streams -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
    </dependency>
</dependencies>

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.group-id=my-group

@Configuration
@EnableKafkaStreams
public class KafkaStreamsProcessor implements KafkaStreamsDSL {
    
    private static final String INPUT_TOPIC = "my-input-topic";
    private static final String OUTPUT_TOPIC = "my-output-topic";
 
    @Override
    public void buildStreams(StreamsBuilder builder) {
        KStream<String, String> inputTopic = builder.stream(INPUT_TOPIC);
        
        // 在这里添加数据处理逻辑
        KStream<String, String> outputTopic = inputTopic
            .mapValues(value -> value.toUpperCase())
            .filter((key, value) -> value.length() > 5);
            
        outputTopic.to(OUTPUT_TOPIC);
    }
}

@SpringBootApplication
public class Application {
    
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
        
        KafkaStreamsProcessor kafkaStreamsProcessor = 
            new KafkaStreamsProcessor();
            
        kafkaStreamsProcessor.start();
    }
}

@RestController
public class MessageController {
 
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
 
    @PostMapping("/send")
    public ResponseEntity<String> sendMessage(@RequestBody String message) {
        kafkaTemplate.send("my-input-topic", message);
        return ResponseEntity.ok("Message sent successfully");
    }
    
    @GetMapping("/receive")
    public ResponseEntity<List<String>> receiveMessages() {
        List<String> messages = // 从输出主题读取消息
        return ResponseEntity.ok(messages);
    }
}

mvn spring-boot:run