springboot整合Kafka生产者

### Spring Boot 整合 Kafka 生产者示例 #### 1. 添加 Maven 或 Gradle 依赖 为了在 Spring Boot 应用程序中使用 Kafka，需要引入 `spring-kafka` 和其他必要的库到构建工具配置文件中。 对于 Maven 用户，在 pom.xml 文件内添加如下依赖： ```xml <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> ``` Gradle 用户则应在 build.gradle 文件里加入这行代码[^1]： ```groovy implementation 'org.springframework.kafka:spring-kafka' ``` #### 2. 配置 application.properties 或 application.yml 接下来设置应用程序属性来指定 Kafka broker 地址以及其他消费者参数。这里给出的是 properties 格式的例子；如果偏好 YAML，则可以相应调整格式[^2]。 ```properties spring.kafka.bootstrap-servers=localhost:9092 ``` 此配置项定义了用于连接至 Kafka cluster 的服务器地址列表。 #### 3. 创建 KafkaProducerConfig 类 创建一个新的 Java 类用来封装生产者的配置选项。此类通常会包含 @Configuration 注解以便被自动扫描并注册为 Bean 定义的一部分。 ```java import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import java.util.HashMap; import java.util.Map; @Configuration public class KafkaProducerConfig { private final String bootstrapServers = "localhost:9092"; public Map<String, Object> producerConfigs() { var props = new HashMap<String, Object>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); return props; } @Bean public ProducerFactory<String, String> producerFactory() { return new DefaultKafkaProducerFactory<>(producerConfigs()); } @Bean public KafkaTemplate<String, String> kafkaTemplate() { return new KafkaTemplate<>(producerFactory()); } } ``` 这段代码设置了基本的序列化器以及指定了要使用的 Kafka brokers[^3]。 #### 4. 编写消息发送服务类 最后一步是编写实际负责向主题发布消息的服务组件。通过注入前面声明好的 `KafkaTemplate` 实现这一点非常容易做到。 ```java import lombok.RequiredArgsConstructor; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Service; @Service @RequiredArgsConstructor public class MessageSenderService { private final KafkaTemplate<String, String> kafkaTemplate; public void sendMessage(String topicName, String message) { System.out.println("Sending message=" + message + " to the topic=" + topicName); this.kafkaTemplate.send(topicName, message); } } ``` 上述方法接收两个参数——目标主题名称和待传输的消息字符串，并调用了模板对象上的 send 方法完成数据传递操作。