十年技术总监分享：Kafka Spring Boot的教程
Apache Kafka是一个分布式数据流平台，具有发布和订阅数据流，以容错方式存储记录以及处理该数据流等功能。
它用于构建实时流数据管道，可以执行功能，例如将数据流从一个应用程序可靠地传递到另一个应用程序，以及处理数据并将其传输到目标应用程序。
主题
Kafka作为一个或多个服务器中的集群运行，集群存储/检索名为Topics的Feed /类别中的记录。主题中的每条记录都存储有键，值和时间戳。
主题可以包含零个，一个或多个使用者，他们将订阅写入该主题的数据。在Kafka术语中，主题始终是多用户馈送的一部分。
分区
Kafka集群为每个主题使用分区日志。
分区维护数据的插入顺序，一旦将记录发布到主题，它将保留在那里，具体取决于保留期（可配置）。记录始终附加在分区的末尾。它维护一个名为 “offsets” 的标志，用于唯一标识分区中的每条记录。
偏移量由消费应用程序控制。使用偏移量，消费者可能会回溯到较旧的偏移量并在需要时重新处理记录。
生产者
记录流，即数据，由生产者发布到主题。它们还可以在分区向主题发布数据时分配分区。生产者可以循环发送数据，也可以根据记录的优先级将记录发送到某些分区，从而实现优先级系统。
消费者
消费者使用该主题的记录。它们基于消费者群体的概念，其中一些消费者被分配在群组中。发布到主题的记录仅从一个消费者组传递给消费者的一个实例。Kafka内部使用消费者组内消费记录的机制。消费者的每个实例将获得特定分区日志，使得在消费者组内，记录可以由每个消费者并行处理。
Spring Boot Kafka
Spring为Kafka提供了很好的支持，并提供了与原生Kafka Java客户端一起使用的抽象层。
我们可以添加以下依赖项来开始使用Spring Boot和Kafka。
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>2.2.3.RELEASE</version> </dependency>
要下载并安装Kafka，请参阅官方指南 https://kafka.apache.org/quickstart 。
下载Kafka后，您可以发出命令来启动ZooKeeper，Kafka使用它来存储元数据。
zookeeper-server-start.bat .configzookeeper.properties
接下来，我们需要通过发出以下命令在本地启动Kafka集群。
kafka-server-start.bat .configserver.properties
现在，默认情况下，Kafka服务器启动 localhost:9092。
编写一个简单的REST控制器并公开一个端点/publish，如下所示。它用于将消息发布到主题。
package com.rahul.kafkaspringboot.controllers;
import com.rahul.kafkaspringboot.services.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping(value = "/kafka")
public class KafkaController {
private final Producer producer;
@Autowired
public KafkaController(Producer producer) {
this.producer = producer;
}
@PostMapping(value = "/publish")
public void sendMessageToKafkaTopic(@RequestParam("message") String message){
this.producer.sendMessage(message);
}
}
然后，我们可以编写使用Spring的kafkaTemplate将消息发送到名为users的主题的生产者，如下所示。
package com.rahul.kafkaspringboot.services;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class Producer {
private static final Logger logger = LoggerFactory.getLogger(Producer.class);
private static final String TOPIC = "users";
@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
public void sendMessage(String message){
logger.info(String.format("$$ -> Producing message --> %s",message));
this.kafkaTemplate.send(TOPIC,message);
}
}我们还可以编写如下所示的消费者，它使用来自主题用户的消息并将日志输出到控制台。
package com.rahul.kafkaspringboot.services;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class Consumer {
private final Logger logger = LoggerFactory.getLogger(Consumer.class);
@KafkaListener(topics = "users", groupId = "group_id")
public void consume(String message){
logger.info(String.format("$$ -> Consumed Message -> %s",message));
}
}
现在，我们需要一种方法告诉我们的应用程序在哪里找到Kafka服务器并创建一个主题并发布到它。我们可以使用 application.yaml 如下所示的方法。
server: port: 9000 spring: kafka: consumer: bootstrap-servers: localhost:9092 group-id: group-id auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer producer: bootstrap-servers: localhost:9092 key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
现在，如果我们运行应用程序并按如下所示到达端点，那么我们已经向主题发布了一条消息。
现在，如果我们从控制台检查日志，它应该打印发送到发布端点的消息，如下所示。
摘要
在这篇文章中，我们已经看到了Kafka系统中使用的基本术语。我们还看到使用Spring Boot配置Kafka是多么容易。大多数工作都是由Spring Boot在幕后完成的。一种简单快捷的方法是在application.yml文件中配置与Kafka相关的详细信息，如果我们更改Kafka集群并且必须将服务器指向新的Kafka集群地址，这非常的便捷!
end：如果你觉得本文对你有帮助的话，记得点赞转发，你的支持就是我更新动力。