Apache Kafka 入门 - Spring Boot 集成 Kafka

深藏阁楼爱情的钟 2022-06-12 21:07 151阅读 0赞

# Apache Kafka 入门 #

*  Kafka的基本配置和运行
 *  Kafka命令详细介绍
 *  Kafka-manager的基本配置和运行
 *  Kafka API 简单用法
 *  Spring Boot 集成Kafka

本篇为第五篇。

> 第四篇和第五篇源码下载：  
> 链接：[http://pan.baidu.com/s/1dE4vpBj][http_pan.baidu.com_s_1dE4vpBj] 密码：j74j

## Spring Boot 集成 Kafka ##

> 参考文档
> 
> [https://kafka.apache.org][https_kafka.apache.org]
> 
> [https://projects.spring.io/spring-kafka/][https_projects.spring.io_spring-kafka]
> 
> [http://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-messaging.html\#boot-features-kafka][http_docs.spring.io_spring-boot_docs_current_reference_html_boot-features-messaging.html_boot-features-kafka]
> 
> [http://docs.spring.io/spring-kafka/docs/1.2.2.RELEASE/reference/htmlsingle/][http_docs.spring.io_spring-kafka_docs_1.2.2.RELEASE_reference_htmlsingle]

Spring Boot 集成 Kafka 相当容易，只需要在基本的依赖中添加下面的依赖。

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

然后在 application.properties 中增加相应的配置即可。

# kafka
    spring.kafka.bootstrap-servers=192.168.16.150:9092
    spring.kafka.consumer.group-id=springboot-group1
    spring.kafka.consumer.auto-offset-reset=earliest

更多可以配置的属性参考 [`KafkaProperties`][KafkaProperties] 。

然后在 Spring Boot 中就可以使用 `KafkaTemplate` 发送消息，使用 `@KafkaListener` 消费指定主题的消息。简单演示代码如下：

package com.github.abel533.boot;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    /** * @author liuzh * @since 2017/6/17. */
    @Controller
    @EnableAutoConfiguration
    public class SampleController { 
    
        public static Logger logger = LoggerFactory.getLogger(SampleController.class);
    
        @Autowired
        private KafkaTemplate<String, String> template;
    
        @RequestMapping("/send")
        @ResponseBody
        String send(String topic, String key, String data) {
            template.send(topic, key, data);
            return "success";
        }
    
        public static void main(String[] args) throws Exception {
            SpringApplication.run(SampleController.class, args);
        }
    
        @KafkaListener(id = "t1", topics = "t1")
        public void listenT1(ConsumerRecord<?, ?> cr) throws Exception {
            logger.info("{} - {} : {}", cr.topic(), cr.key(), cr.value());
        }
    
        @KafkaListener(id = "t2", topics = "t2")
        public void listenT2(ConsumerRecord<?, ?> cr) throws Exception {
            logger.info("{} - {} : {}", cr.topic(), cr.key(), cr.value());
        }
    
    }

通过`http://localhost:8080/send?topic=t2&key=test1&data=hello122`方法可以产生指定主题的消息。或者可以通过前面使用命令行方式或者API方式创建消息。

上面代码输出的日志（和生产者有关）如下：

[t1-0-C-1] t1 - 0 : 0
    [t2-0-C-1] t2 - 0 : 0
    [t2-0-C-1] t2 - 1 : 1
    [t1-0-C-1] t1 - 1 : 1
    [t1-0-C-1] t1 - 2 : 2
    [t2-0-C-1] t2 - 2 : 2
    [t1-0-C-1] t1 - 3 : 3
    [t2-0-C-1] t2 - 3 : 3
    [t1-0-C-1] t1 - 4 : 4
    [t2-0-C-1] t2 - 4 : 4
    [t2-0-C-1] t2 - 5 : 5
    [t1-0-C-1] t1 - 5 : 5
    [t1-0-C-1] t1 - 6 : 6
    [t2-0-C-1] t2 - 6 : 6
    [t2-0-C-1] t2 - 7 : 7
    [t1-0-C-1] t1 - 7 : 7
    [t1-0-C-1] t1 - 8 : 8
    [t2-0-C-1] t2 - 8 : 8
    [t2-0-C-1] t2 - 9 : 9
    [t1-0-C-1] t1 - 9 : 9

上面只是在Spring Boot中最基本的用法，想要定制更多的功能可以参考 [Spring for Apache Kafka][http_docs.spring.io_spring-kafka_docs_1.2.2.RELEASE_reference_htmlsingle] 的完整文档。

[http_pan.baidu.com_s_1dE4vpBj]: http://pan.baidu.com/s/1dE4vpBj
[https_kafka.apache.org]: https://kafka.apache.org
[https_projects.spring.io_spring-kafka]: https://projects.spring.io/spring-kafka/
[http_docs.spring.io_spring-boot_docs_current_reference_html_boot-features-messaging.html_boot-features-kafka]: http://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-messaging.html#boot-features-kafka
[http_docs.spring.io_spring-kafka_docs_1.2.2.RELEASE_reference_htmlsingle]: http://docs.spring.io/spring-kafka/docs/1.2.2.RELEASE/reference/htmlsingle/
[KafkaProperties]: https://github.com/spring-projects/spring-boot/tree/v1.5.4.RELEASE/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java