这篇博客文章展示了如何配置Spring Kafka和Spring Boot以使用JSON发送消息并以多种格式接收它们：JSON，纯字符串或字节数组。基于此配置，您还可以将Kafka生成器从发送JSON切换到其他序列化方法。
此示例应用程序还演示了同一消费组中三个Kafka消费者的使用情况，因此消息在三者之间进行负载平衡。每个消费者实现不同的反序列化方法。
您可以了解一些Kafka概念，如Consumer Group和Topic分区。
多个消费者
要更好地理解配置，请查看下图。如您所见，我们创建了一个包含三个分区的Kafka主题。在消费者方面，只有一个应用程序，但它实现了具有相同group.id 属性的三个Kafka消费者。
当我们启动应用程序时，Kafka会为每个消费者分配一个不同的分区。消费者组将以负载平衡的方式接收消息。在这篇文章的后面，如果我们让它们具有不同的组标识符，你会看到有什么区别(如果你熟悉Kafka，你可能知道结果)。
示例用例
我们要构建的逻辑很简单。每次我们调用指定REST端点hello，应用程序将生成可配置数量的消息，并使用序列号作为Kafka密钥将它们发送到同一主题，等待消费所有消息后返回Hello Kafka!
设置Kafka和Spring Boot
首先，您需要有一个正在运行的Kafka集群才能连接。对于这个应用程序，我将在单个节点中使用docker-compose和Kafka。这显然远不是一个生产配置，但它足以满足这篇文章的目标。
以下是docker-compose.yml配置
version: '2'
services:
zookeeper:
image: wurstmeister/zookeeper
ports:
- "2181:2181"kafka:
image: wurstmeister/kafka
ports:
-"9092:9092"environment:
KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
请注意，我将Kafka配置为不自动创建主题(最后一行配置)。我们将在Spring Boot应用程序创建我们的主题，因为我们想要传递一些自定义配置。如果你想玩玩这些Docker图像(例如使用多个节点)，请查看wurstmeister/zookeeper图像文档
要启动Kafka和Zookeeper容器，只要在上述配置目录下运行 docker-compose up
获取SpringBoot应用程序骨架的最简单方法是到start.spring.io，使用使用YAML进行配置配置application.yml：
spring:
kafka:
consumer:
group-id: tpd-loggers
auto-offset-reset: earliest
# change this property if you are using your own
# Kafka cluster or your Docker IP is different
bootstrap-servers: localhost:9092
tpd:
topic-name: advice-topic
messages-per-request: 10
第一部分属性是Spring Kafka配置：
Kafka的组标识 group-id
auto-offset-reset 属性设置为earliest，这意味着当消费者没有发现偏移量(指针)时，消费者将开始从最早的消息中读取消息。
第三行用于连接Kafka的服务器，在这种情况下，如果您使用单节点配置，则是唯一可用的服务器。请注意，如果使用默认值 localhost:9092，则此属性是多余的 。
第二部分是特定于应用程序的自定义配置。我们定义Kafka主题名称以及每次执行HTTP REST请求时要发送的消息数。
Message类
这是我们将用作Kafka消息的Java类。这里没有什么复杂的，只是@JsonProperty 在构造函数参数中带有注释的不可变类， 因此Jackson可以正确地反序列化它。
import com.fasterxml.jackson.annotation.JsonProperty;
public class PracticalAdvice {
private final String message;
private final int identifier;
public PracticalAdvice(@JsonProperty("message") final String message,
@JsonProperty("identifier") final int identifier) {
this.message = message;
this.identifier = identifier;
}
public String getMessage() {
return message;
}
public int getIdentifier() {
return identifier;
}
@Override
public String toString() {
return"PracticalAdvice::toString() {"+"message='"+ message + '\'' +", identifier="+ identifier +
'}';
}
}
Spring Boot中的Kafka Producer配置
为了简化应用程序，我们将在Spring Boot类中添加配置。最后，我们希望在此处包含生产者和消费者配置，并使用三种不同的变体进行反序列化。请记住，您可以在GitHub存储库中找到完整的源代码。
首先，让我们关注Producer配置：
@SpringBootApplication
public class KafkaExampleApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaExampleApplication.class, args);
}
@Autowired
private KafkaProperties kafkaProperties;
@Value("${tpd.topic-name}")
