3. 引入 (3. Introduction) | Spring for Apache Kafkaversion 2.9.13中文文档|Spring官方文档|SpringBoot 教程|Spring中文网

3. 引入

此参考文档的这一部分是对 Spring for Apache Kafka 的高层次概述，涵盖了底层概念以及一些代码片段，帮助您尽可能快地开始使用。spring-doc.cadn.net.cn

3.1. 快速巡游

前提条件：您必须安装并运行Apache Kafka。然后您必须将Spring for Apache Kafka（spring-kafka）JAR及其所有依赖项放入您的类路径中。最简单的方法是在构建工具中声明一个依赖。spring-doc.cadn.net.cn

如果你不使用 Spring Boot，请在你的项目中将 spring-kafka jar 作为依赖进行声明。spring-doc.cadn.net.cn

Maven

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>2.9.13</version>
</dependency>

Gradle

compile 'org.springframework.kafka:spring-kafka:2.9.13'

在使用 Spring Boot 时（如果没有使用 start.spring.io 来创建项目），可以省略版本信息，Boot 会自动引入与您的 Boot 版本兼容的正确版本：

Maven

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

Gradle

compile 'org.springframework.kafka:spring-kafka'

但是，最快捷的入门方式是使用start.spring.io（或Spring Tool Suits和Intellij IDEA中的向导）创建一个项目，并选择'Spring for Apache Kafka'作为依赖项。spring-doc.cadn.net.cn

3.1.1. 兼容性

本快速入门适用于以下版本：spring-doc.cadn.net.cn

Apache Kafka 客户端 3.2.xspring-doc.cadn.net.cn
Spring Framework 5.3.xspring-doc.cadn.net.cn
最低Java版本: 8spring-doc.cadn.net.cn

3.1.2. Getting Started

最简单的开始方式是使用start.spring.io（或Spring Tool Suits和Intellij IDEA中的向导）创建项目，并选择'Spring for Apache Kafka'作为依赖项。
有关其对基础架构bean的意见化自动配置的更多信息，请参阅Spring Boot文档。spring-doc.cadn.net.cn

这是一个简单的消费者应用程序。spring-doc.cadn.net.cn

Spring Boot 消费者应用

Example 1. 应用程序

Java

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("topic1")
                .partitions(10)
                .replicas(1)
                .build();
    }

    @KafkaListener(id = "myId", topics = "topic1")
    public void listen(String in) {
        System.out.println(in);
    }

}

Kotlin

@SpringBootApplication
class Application {

    @Bean
    fun topic() = NewTopic("topic1", 10, 1)

    @KafkaListener(id = "myId", topics = ["topic1"])
    fun listen(value: String?) {
        println(value)
    }

}

fun main(args: Array<String>) = runApplication<Application>(*args)

Example 2. application.properties

spring.kafka.consumer.auto-offset-reset=earliest

零号（NewTopic）bean 导致在代理上创建主题；如果主题已经存在，则不需要此步骤。spring-doc.cadn.net.cn

Spring Boot Producer App

例 3. 应用程序

Java

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("topic1")
                .partitions(10)
                .replicas(1)
                .build();
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> {
            template.send("topic1", "test");
        };
    }

}

Kotlin

@SpringBootApplication
class Application {

    @Bean
    fun topic() = NewTopic("topic1", 10, 1)

    @Bean
    fun runner(template: KafkaTemplate<String?, String?>) =
        ApplicationRunner { template.send("topic1", "test") }

    companion object {
        @JvmStatic
        fun main(args: Array<String>) = runApplication<Application>(*args)
    }

}

使用Java配置（无Spring Boot）

Spring for Apache Kafka 的设计目的是用于 Spring 应用上下文中。

spring-doc.cadn.net.cn

例如，如果您在 Spring 上下文之外自行创建侦听器容器，则除非满足容器实现的所有 …Aware 接口，否则不会所有功能都能正常工作。spring-doc.cadn.net.cn

这里有一个不使用 Spring Boot 的应用程序示例；它同时具有 Consumer 和 Producer。spring-doc.cadn.net.cn

Example 4. Without Boot

Java

public class Sender {

	public static void main(String[] args) {
		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class);
		context.getBean(Sender.class).send("test", 42);
	}

	private final KafkaTemplate<Integer, String> template;

	public Sender(KafkaTemplate<Integer, String> template) {
		this.template = template;
	}

	public void send(String toSend, int key) {
		this.template.send("topic1", key, toSend);
	}

}

public class Listener {

    @KafkaListener(id = "listen1", topics = "topic1")
    public void listen1(String in) {
        System.out.println(in);
    }

}

@Configuration
@EnableKafka
public class Config {

    @Bean
    ConcurrentKafkaListenerContainerFactory<Integer, String>
                        kafkaListenerContainerFactory(ConsumerFactory<Integer, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerProps());
    }

    private Map<String, Object> consumerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // ...
        return props;
    }

    @Bean
    public Sender sender(KafkaTemplate<Integer, String> template) {
        return new Sender(template);
    }

    @Bean
    public Listener listener() {
        return new Listener();
    }

    @Bean
    public ProducerFactory<Integer, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(senderProps());
    }

    private Map<String, Object> senderProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        //...
        return props;
    }

    @Bean
    public KafkaTemplate<Integer, String> kafkaTemplate(ProducerFactory<Integer, String> producerFactory) {
        return new KafkaTemplate<Integer, String>(producerFactory);
    }

}

Kotlin

class Sender(private val template: KafkaTemplate<Int, String>) {

    fun send(toSend: String, key: Int) {
        template.send("topic1", key, toSend)
    }

}

class Listener {

    @KafkaListener(id = "listen1", topics = ["topic1"])
    fun listen1(`in`: String) {
        println(`in`)
    }

}

@Configuration
@EnableKafka
class Config {

    @Bean
    fun kafkaListenerContainerFactory(consumerFactory: ConsumerFactory<Int, String>) =
        ConcurrentKafkaListenerContainerFactory<Int, String>().also { it.consumerFactory = consumerFactory }


    @Bean
    fun consumerFactory() = DefaultKafkaConsumerFactory<Int, String>(consumerProps)

    val consumerProps = mapOf(
        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
        ConsumerConfig.GROUP_ID_CONFIG to "group",
        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to IntegerDeserializer::class.java,
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest"
    )

    @Bean
    fun sender(template: KafkaTemplate<Int, String>) = Sender(template)

    @Bean
    fun listener() = Listener()

    @Bean
    fun producerFactory() = DefaultKafkaProducerFactory<Int, String>(senderProps)

    val senderProps = mapOf(
        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
        ProducerConfig.LINGER_MS_CONFIG to 10,
        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to IntegerSerializer::class.java,
        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java
    )

    @Bean
    fun kafkaTemplate(producerFactory: ProducerFactory<Int, String>) = KafkaTemplate(producerFactory)

}

如你所见，当不使用Spring Boot时，你必须定义几个基础架构bean。spring-doc.cadn.net.cn