【kafka实战】06 kafkaTemplate java代码使用示例

在 Spring Boot 中使用 KafkaTemplate 可以方便地向 Kafka 发送消息。下面为你详细介绍使用步骤和示例代码。

1. 创建 Spring Boot 项目

你可以使用 Spring Initializr（https://start.spring.io/ ）来创建一个新的 Spring Boot 项目，添加以下依赖：

• Spring for Apache Kafka

或者在 pom.xml 中手动添加依赖：

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>
</dependencies>

2. 配置 Kafka

在 application.properties 或 application.yml 中配置 Kafka 的相关信息，示例如下：

application.properties

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

application.yml

spring:kafka:bootstrap-servers: localhost:9092producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer

3. 创建消息生产者服务

创建一个服务类，使用 KafkaTemplate 发送消息。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;@Service
public class KafkaProducerService {private static final String TOPIC_NAME = "test_topic";@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;public void sendMessage(String message) {kafkaTemplate.send(TOPIC_NAME, message);System.out.println("Message sent: " + message);}public void sendMessageWithKey(String key, String message) {kafkaTemplate.send(TOPIC_NAME, key, message);System.out.println("Message sent with key " + key + ": " + message);}
}

4. 创建控制器（可选）

如果你想通过 RESTful API 触发消息发送，可以创建一个控制器。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("/kafka")
public class KafkaController {@Autowiredprivate KafkaProducerService kafkaProducerService;@GetMapping("/send/{message}")public String sendMessage(@PathVariable String message) {kafkaProducerService.sendMessage(message);return "Message sent successfully";}@GetMapping("/send/{key}/{message}")public String sendMessageWithKey(@PathVariable String key, @PathVariable String message) {kafkaProducerService.sendMessageWithKey(key, message);return "Message with key sent successfully";}
}

5. 异步发送消息并处理结果

KafkaTemplate 的 send 方法返回一个 ListenableFuture，可以用来异步处理消息发送的结果。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;@Service
public class AsyncKafkaProducerService {private static final String TOPIC_NAME = "test_topic";@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;public void sendAsyncMessage(String message) {ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC_NAME, message);future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {@Overridepublic void onSuccess(SendResult<String, String> result) {System.out.println("Message sent successfully: " + message + ", Offset: " + result.getRecordMetadata().offset());}@Overridepublic void onFailure(Throwable ex) {System.err.println("Failed to send message: " + message + ", Error: " + ex.getMessage());}});}
}

6. 运行 Spring Boot 应用

启动 Spring Boot 应用程序，你可以通过以下方式测试消息发送：

• 如果使用了控制器，可以通过访问 http://localhost:8080/kafka/send/your_message 或 http://localhost:8080/kafka/send/your_key/your_message 来发送消息。
• 也可以在代码中直接调用 KafkaProducerService 或 AsyncKafkaProducerService 的方法来发送消息。

7. 注意事项

• 确保你的 Kafka 服务正在运行，并且 bootstrap-servers 配置正确。
• 对于不同的数据类型，你可能需要调整 key-serializer 和 value-serializer 的配置。例如，如果要发送 JSON 数据，可以使用 org.springframework.kafka.support.serializer.JsonSerializer。

通过以上步骤，你可以在 Spring Boot 项目中使用 KafkaTemplate 方便地向 Kafka 发送消息。