kafka(二)--springboot+kafka整合(环境：IntelliJ IDEA)

Bertha 。 2022-10-09 04:52 199阅读 0赞

**目录**

（一）kafka

（二）kafka本机测试(在虚拟机Ubuntu(服务器))

（三）用Java创建生产者和消费者：

--------------------

原文链接：[https://blog.csdn.net/u014453898/article/details/84870761][https_blog.csdn.net_u014453898_article_details_84870761]

**项目代码下载：[https://github.com/zhujunwen/springboot-kafka][https_github.com_zhujunwen_springboot-kafka]**

# （一）kafka #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70][]

producer：生产者，负责发布消息到kafka cluster(kafka集群)中。生产者可以是web前端产生的page view，或者是服务器日志，系统CPU、memory等。

consumer：消费者，每个consumer属于一个特定的consuer group（可为每个consumer指定group name，若不指定group name则属于默认的group）。创建消费者时，要指定消费者接受的消息的topic，该消费者只会接受该topic的消息。

topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为topic。（物理上不同topic的消息分开存储，逻辑上一个topic的消息虽然保存于一个或多个broker上但用户只需指定消息的topic即可生产或消费数据而不必关心数据存于何处）。

broker：kafka集群包含一个或多个服务器，这些服务器就叫做broker。

# （二）kafka本机测试(在虚拟机Ubuntu(服务器)) #

在用IntelliJ 编写Java代码前，先要在服务器启动kafka和自带的zookeeper，本文在虚拟机上启动。

kafka下载：从官网下载 kafka\_2.12-2.1.0.tgz，解压在虚拟机(服务器)上。

进入到kafka的解压目录：

> 输入命令启动zookeeper：
> 
> bin/zookeeper-server-start.sh config/zookeeper.properties &

>   
> 打开另一个终端输入命令启动kafka：
> 
> bin/kafka-server-start.sh config/server.properties

接下来我们就可以用尝试用命令创建生产者和消费者了（当然是用来给我们看一下效果用的）：

再打开另一个终端输入命令创建生产者：

>  bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test\_topic

broker-list 参数指定生产者所使用的broker

localhost：9092 参数表示broker，这个broker为本机(127.0.0.1)，且使用的端口是kafka的默认端口号是9092

topic 参数表示生产者生产的消息的topic 为 “test\_topic”

**最后再打开另一个终端创建消费者：**

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test\_topic --from-beginning

bootstrap-server 是指定consumer从哪里(broker)取出消息

topic 指定消费者consumer取出的 topic 为“test\_topic”的消息。

from-beginning：Kafka实际环境有可能会出现Consumer全部宕机，虽然基于Kafka的高可用特性，消费者群组中的消费者可以实现再均衡，所有Consumer不处理数据的情况很少，但是还是有可能会出现，此时就要求Consumer重启的时候能够读取在宕机期间Producer发送的数据。基于消费者订阅模式默认是无法实现的，因为只能订阅最新发送的数据。通过消费者命令行可以实现，只要在命令行中加上--from-beginning即可

效果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 1][]  
创建生产者和消费者：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 2][]

通过输入者输入消息，生产者会显示消息：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 3][]

# （三）用Java创建生产者和消费者： #

在虚拟机(服务器)上 运行了kafka和kafka自带的zookeeper之后，就可以编写Java代码来创建生产者和消费者了：

若是在windows上使用intelli编程，首先要保证本机和虚拟机之间能ping通(确保网段一样192.168.x.x)，同理kafka要是部署在服务器的话，本机就 要和服务器之间能ping通。

**创建springboot项目：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 4][]

**选择web 和 kafka：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 5][]

**最终的项目目录树：（红色框的是要的动手做的部分）**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 6][]

**配置文件application.yml：(一般项目自动生成的是applicaiton.properties，但为了书写简便，改成yml)**

> spring:  
>   kafka:  
>     bootstrap-servers: 192.168.153.131:9092 \#服务器的ip及端口，可以写多个，服务器之间用“：”间隔  
>     producer: \#生产者配置  
>       key-serializer: org.apache.kafka.common.serialization.StringSerializer  
>       value-serializer: org.apache.kafka.common.serialization.StringSerializer  
>     consumer: \#消费者配置  
>       group-id: test \#设置消费者的组id  
>       enable-auto-commit: true  
>       auto-commit-interval: 1000  
>       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer  
>       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

**DemoApplication.java：（java的入口程序，最后是运行这个程序）**

package com.example.demo;
     
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
     
    @SpringBootApplication
    public class DemoApplication {
     
        public static void main(String[] args) {
            SpringApplication.run(DemoApplication.class, args);
        }
    }

**TestKafkaProducerController.java:(生产者)**

package com.example.demo;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
     
     
    @RestController     //定义这是一个控制器，可以通过浏览器访问
    @RequestMapping("kafka")
    public class TestKafkaProducerController {
     
        @Autowired
    //    定义一个kafka模板
        private KafkaTemplate<String, String> kafkaTemplate; //这里会报错，但不影响
    //当在浏览器上输入http://localhost:8080/kafka/send?msg=abc，就会发送abc到服务器上去让消费者接收，msg对应下面的String msg
        @RequestMapping("send")
        public String send(String msg){
            kafkaTemplate.send("test_topic", msg); //使用kafka模板发送信息
            return "success";
        }
    }

**TestConsumer.java:**

package com.example.demo;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
     
     
    @Component
    public class TestConsumer {
     
        @KafkaListener(topics = "test_topic") //定义此消费者接收topic为“test_topic”的消息
    //    监听服务器上的kafka是否有相关的消息发过来
    //    record变量代表消息本身，可以通过ConsumerRecord<?,?>类型的record变量来打印接收的消息的各种信息
        public void listen (ConsumerRecord<?, ?> record) throws Exception {
            System.out.printf("topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
        }
    }

**(四)运行效果：**  
【1】运行IntelliJ上的DemoApplication之后，在虚拟机(服务器)上输入hi时，不仅虚拟机上运行的测试消费者能收到，windows上的IntelliJ程序也能收到

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 7][]

![20181207154415860.png][]

【2】在浏览器上输入http://localhost:8080/kafka/send?msg=world，不仅intelliJ上的消费者能收到，在虚拟机(服务器)上运行的测试消费者也能收到：（其中8080是tomcat服务器的端口，springboot默认下带的是tomcat）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 8][]

intelliJ上的效果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 9][]

虚拟机(服务器)上的效果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 10][]

===========================

最后显示一下pom.xml，这个文件一般都自动生成，用来显示项目的依赖，我给出来的原因是给大家参考一下，因为有时候代码一样，但环境错了也会导致运行不了：

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
     
        <groupId>com.example</groupId>
        <artifactId>demo</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <packaging>jar</packaging>
     
        <name>demo</name>
        <description>Demo project for Spring Boot</description>
     
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.0.RELEASE</version>
            <relativePath/> 
        </parent>
     
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <java.version>1.8</java.version>
        </properties>
     
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
            </dependency>
     
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
     
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
        
    </project>

[https_blog.csdn.net_u014453898_article_details_84870761]: https://blog.csdn.net/u014453898/article/details/84870761
[https_github.com_zhujunwen_springboot-kafka]: https://github.com/zhujunwen/springboot-kafka
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70]: /images/20221005/e968d2ee5949486fb1bf7e16117e6fcd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 1]: /images/20221005/36043690e9d6442c9ecc3476dec3eab5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 2]: /images/20221005/8b79f9d2aec0421ebb903e851a2949b1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 3]: /images/20221005/775e99d1dd3142dead4d07410aeeced8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 4]: /images/20221005/ceca1a27f0ec49d3b101e5a7020dcc68.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 5]: /images/20221005/f3bc4276057a433eb6a08c0690041ac2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 6]: /images/20221005/db6f0c33eb874bc4927876480770f645.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 7]: /images/20221005/19075700a0f441af86c8bfe91ae561f0.png
[20181207154415860.png]: /images/20221005/aee9559c235840d5bf828a3bcafef0d8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 8]: /images/20221005/571251660b5a435493300f0bb09ee3b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 9]: /images/20221005/031e706907d7449f807375b264d51f25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0NTM4OTg_size_16_color_FFFFFF_t_70 10]: /images/20221005/7ddcbb1282dc4c14b6528a1384997674.png