摘要

在这篇文章中，我们已经看到了Kafka系统中使用的基本术语。我们还看到使用Spring Boot配置Kafka是多么容易。大多数工作都是由Spring Boot在幕后完成的。一种简单快捷的方法是在application.yml文件中配置与Kafka相关的详细信息，如果我们更改Kafka集群并且必须将服务器指向新的Kafka集群地址，这非常的便捷!

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服务器启动 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

现在，如果我们运行应用程序并按如下所示到达端点，那么我们已经向主题发布了一条消息。

现在，如果我们从控制台检查日志，它应该打印发送到发布端点的消息，如下所示。

点关注，不迷路；持续更新Java架构相关技术及资讯热文！！！