此版本仍在开发中，尚未被视为稳定版本。如需最新稳定版本，请使用 Spring Boot 4.0.4!spring-doc.cadn.net.cn

Apache Kafka 支持

Apache Kafka 通过提供 spring-kafka 项目的自动配置来支持。spring-doc.cadn.net.cn

Kafka 配置由 spring.kafka.* 中的外部配置属性控制。 例如，您可以在 application.properties 中声明以下部分：spring-doc.cadn.net.cn

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=myGroup
spring:
  kafka:
    bootstrap-servers: "localhost:9092"
    consumer:
      group-id: "myGroup"
要在启动时创建主题，请添加一个类型为 NewTopic 的 bean。 如果该主题已存在，则会忽略此 bean。

有关更多支持的选项，请参见 KafkaProperties。spring-doc.cadn.net.cn

发送消息

Spring 的 KafkaTemplate 是自动配置的，你可以直接在自己的 Bean 中自动装配它，如下例所示：spring-doc.cadn.net.cn

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

	private final KafkaTemplate<String, String> kafkaTemplate;

	public MyBean(KafkaTemplate<String, String> kafkaTemplate) {
		this.kafkaTemplate = kafkaTemplate;
	}

	// ...

	public void someMethod() {
		this.kafkaTemplate.send("someTopic", "Hello");
	}

}
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.stereotype.Component

@Component
class MyBean(private val kafkaTemplate: KafkaTemplate<String, String>) {

	// ...

	fun someMethod() {
		kafkaTemplate.send("someTopic", "Hello")
	}

}
如果定义了属性 spring.kafka.producer.transaction-id-prefix，则会自动配置一个 KafkaTransactionManager。 此外，如果定义了一个 RecordMessageConverter bean，它会自动关联到自动配置的 KafkaTemplate。

如果上下文中存在类型为 KafkaTemplateObservationConvention 的 bean，它将自动注册到 KafkaTemplate 上。spring-doc.cadn.net.cn

接收消息

当存在 Apache Kafka 基础架构时，可以使用 @KafkaListener 注解任何 bean 以创建监听器端点。 如果未定义 KafkaListenerContainerFactory，则会自动配置一个默认实例，其键在 spring.kafka.listener.* 中定义。spring-doc.cadn.net.cn

以下组件在 someTopic 主题上创建一个监听器端点：spring-doc.cadn.net.cn

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MyBean {

	@KafkaListener(topics = "someTopic")
	public void processMessage(String content) {
		// ...
	}

}
import org.springframework.kafka.annotation.KafkaListener
import org.springframework.stereotype.Component

@Component
class MyBean {

	@KafkaListener(topics = ["someTopic"])
	fun processMessage(content: String?) {
		// ...
	}

}

如果定义了 KafkaTransactionManager Bean，它会自动关联到容器工厂。 同样地，如果定义了 RecordFilterStrategy、CommonErrorHandler、AfterRollbackProcessor 或 ConsumerAwareRebalanceListener Bean，它会自动关联到默认工厂。spring-doc.cadn.net.cn

根据监听器类型，一个 RecordMessageConverter 或 BatchMessageConverter Bean 会与默认工厂关联。 如果批处理监听器仅存在一个 RecordMessageConverter Bean，它将被包装在 BatchMessageConverter 中。spring-doc.cadn.net.cn

自定义的 ChainedKafkaTransactionManager 必须标记为 @Primary，因为它通常引用自动配置的 KafkaTransactionManager Bean。

如果上下文中存在类型为 KafkaListenerObservationConvention 的 bean，它将被自动注册到容器工厂中。spring-doc.cadn.net.cn

Kafka Streams

Spring for Apache Kafka 提供了一个工厂 bean 来创建一个 StreamsBuilder 对象并管理其流的生命周期。 Spring Boot 会在 kafka-streams 在类路径上并且通过 @EnableKafkaStreams 注解启用 Kafka Streams 时自动配置所需的 KafkaStreamsConfiguration bean。spring-doc.cadn.net.cn

启用 Kafka Streams 意味着必须设置应用程序 ID 和引导服务器。 前者可以使用 spring.kafka.streams.application-id 进行配置，如果未设置，则默认为 spring.application.name。 后者可以全局设置，或者仅针对 streams 进行特定覆盖。spring-doc.cadn.net.cn

使用专用属性可以使用其他附加属性；其他任意的Kafka属性可以使用spring.kafka.streams.properties命名空间进行设置。 有关更多信息，请参见附加的Kafka属性。spring-doc.cadn.net.cn

要使用工厂 Bean，请按照以下示例所示，将 StreamsBuilder 注入到你的 @Bean 中：spring-doc.cadn.net.cn

import java.util.Locale;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;

@Configuration(proxyBeanMethods = false)
@EnableKafkaStreams
public class MyKafkaStreamsConfiguration {

	@Bean
	public KStream<Integer, String> kStream(StreamsBuilder streamsBuilder) {
		KStream<Integer, String> stream = streamsBuilder.stream("ks1In");
		stream.map(this::uppercaseValue)
			.to("ks1Out",
					Produced.with(Serdes.Integer(), new org.springframework.kafka.support.serializer.JsonSerde<>()));
		return stream;
	}

	private KeyValue<Integer, String> uppercaseValue(Integer key, String value) {
		return new KeyValue<>(key, value.toUpperCase(Locale.getDefault()));
	}

}
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.streams.KeyValue
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.kstream.KStream
import org.apache.kafka.streams.kstream.Produced
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.kafka.annotation.EnableKafkaStreams

@Configuration(proxyBeanMethods = false)
@EnableKafkaStreams
class MyKafkaStreamsConfiguration {

	@Bean
	fun kStream(streamsBuilder: StreamsBuilder): KStream<Int, String> {
		val stream = streamsBuilder.stream<Int, String>("ks1In")
		stream.map(this::uppercaseValue).to("ks1Out", Produced.with(Serdes.Integer(),
			org.springframework.kafka.support.serializer.JsonSerde()))
		return stream
	}

	private fun uppercaseValue(key: Int, value: String): KeyValue<Int, String> {
		return KeyValue(key, value.uppercase())
	}

}

默认情况下，由 StreamsBuilder 对象管理的流会自动启动。 您可以使用 spring.kafka.streams.auto-startup 属性自定义此行为。spring-doc.cadn.net.cn

您还可以注册任意数量的实现 StreamsBuilderFactoryBeanConfigurer 的 bean，以进行更高级的自定义。

其他 Kafka 属性

自动配置支持的属性在附录的 集成属性 部分中列出。 请注意，这些属性（连字符或camelCase）通常直接映射到Apache Kafka的点分属性。 有关详细信息，请参阅Apache Kafka文档。spring-doc.cadn.net.cn

名称中不包含客户端类型（producer、consumer、admin 或 streams）的属性被视为通用属性，适用于所有客户端。 如果需要，可以为一种或多种客户端类型覆盖这些通用属性中的大多数。spring-doc.cadn.net.cn

Apache Kafka 将属性的重要性分为 HIGH、MEDIUM 或 LOW。 Spring Boot 自动配置支持所有 HIGH 重要性的属性，部分选定的 MEDIUM 和 LOW 属性，以及任何没有默认值的属性。spring-doc.cadn.net.cn

只有 Kafka 支持的属性子集可以通过 KafkaProperties 类直接访问。 如果您希望使用其他不受直接支持的属性来配置各个客户端类型，请使用以下属性：spring-doc.cadn.net.cn

spring.kafka.properties[prop.one]=first
spring.kafka.admin.properties[prop.two]=second
spring.kafka.consumer.properties[prop.three]=third
spring.kafka.producer.properties[prop.four]=fourth
spring.kafka.streams.properties[prop.five]=fifth
spring:
  kafka:
    properties:
      "[prop.one]": "first"
    admin:
      properties:
        "[prop.two]": "second"
    consumer:
      properties:
        "[prop.three]": "third"
    producer:
      properties:
        "[prop.four]": "fourth"
    streams:
      properties:
        "[prop.five]": "fifth"

这会将通用的 prop.one Kafka 属性设置为 first（适用于生产者、消费者、管理员和流），将 prop.two 管理员属性设置为 second，将 prop.three 消费者属性设置为 third，将 prop.four 生产者属性设置为 fourth，并将 prop.five 流属性设置为 fifth。spring-doc.cadn.net.cn

您还可以按如下方式配置 Spring Kafka JsonDeserializer：spring-doc.cadn.net.cn

spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties[spring.json.value.default.type]=com.example.Invoice
spring.kafka.consumer.properties[spring.json.trusted.packages]=com.example.main,com.example.another
spring:
  kafka:
    consumer:
      value-deserializer: "org.springframework.kafka.support.serializer.JsonDeserializer"
      properties:
        "[spring.json.value.default.type]": "com.example.Invoice"
        "[spring.json.trusted.packages]": "com.example.main,com.example.another"

同样，您可以禁用 JsonSerializer 在请求头中发送类型信息的默认行为：spring-doc.cadn.net.cn

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.properties[spring.json.add.type.headers]=false
spring:
  kafka:
    producer:
      value-serializer: "org.springframework.kafka.support.serializer.JsonSerializer"
      properties:
        "[spring.json.add.type.headers]": false
以这种方式设置的属性会覆盖 Spring Boot 明确支持的任何配置项。

使用嵌入式 Kafka 进行测试

Spring for Apache Kafka 为使用嵌入式 Apache Kafka 代理测试项目提供了一种便捷的方式。 要使用此功能，请将测试类标注为 @EmbeddedKafka 来自 spring-kafka-test 模块。 有关更多信息，请参阅 Spring for Apache Kafka 参考手册。spring-doc.cadn.net.cn

要使 Spring Boot 自动配置与上述嵌入式 Apache Kafka 代理协同工作，您需要将嵌入式代理地址的系统属性（由 EmbeddedKafkaBroker 填充）重新映射到 Apache Kafka 的 Spring Boot 配置属性。 有几种方法可以实现这一点：spring-doc.cadn.net.cn

  • 提供一个系统属性，将嵌入式代理地址映射为测试类中的 spring.kafka.bootstrap-servers：spring-doc.cadn.net.cn

	static {
		System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers");
	}
	init {
		System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers")
	}
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;

@SpringBootTest
@EmbeddedKafka(topics = "someTopic", bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class MyTest {

	// ...

}
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.kafka.test.context.EmbeddedKafka

@SpringBootTest
@EmbeddedKafka(topics = ["someTopic"], bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class MyTest {

	// ...

}
spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}
spring:
  kafka:
    bootstrap-servers: "${spring.embedded.kafka.brokers}"