Spring Integration for Apache Kafka 教程

2024-08-07 11:55:51作者：翟萌耘Ralph

1. 项目介绍

Spring Integration for Apache Kafka 是一个扩展项目，它为 Apache Kafka 提供了 Spring Integration 的支持。该项目基于 Spring for Apache Kafka，允许开发者利用核心 Spring 概念来构建基于 Kafka 的消息解决方案。它提供了入站和出站通道适配器以及网关，方便应用程序与 Kafka 进行交互。

2. 项目快速启动

准备工作

确保已安装以下依赖项：

Java Development Kit（JDK）8 或更高版本
Apache Maven
Apache Kafka 集群

添加依赖

在 pom.xml 文件中添加 Spring Integration for Apache Kafka 的依赖：

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-kafka</artifactId>
    <version>最新版本号</version> <!-- 替换为你所需的版本 -->
</dependency>

创建配置

创建一个名为 application.yml 的配置文件，设置 Kafka 相关属性：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: my-consumer-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

编写示例代码

创建一个 Spring Boot 应用，包含一个简单的生产者和消费者：

Producer.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class Producer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void send(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

Consumer.java

import org.springframework.boot.CommandLineRunner;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer implements CommandLineRunner {

    @KafkaListener(topics = "test")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }

    @Override
    public void run(String... args) throws Exception {
        // 当应用启动时，监听器会自动开始运行
    }
}