【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 发送消息。