上周刚刚欢天喜地的在linux上部了kafka,这周打算用spring-boot框架写个简单demo跑一下,结果悲剧就此展开。

  1. 首先建立maven工程:pom中添加spring boot kafka依赖:

    <?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>
       <parent>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-parent</artifactId>
          <version>2.1.5.RELEASE</version>
          <relativePath/> <!-- lookup parent from repository -->
       </parent>
       <groupId>com.example</groupId>
       <artifactId>kafkaproducer</artifactId>
       <version>0.0.1-SNAPSHOT</version>
       <name>kafkaproducer</name>
       <description>Demo project for Spring Boot</description>
    
       <properties>
          <java.version>1.8</java.version>
       </properties>
    
       <dependencies>
          <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
          </dependency>
    
          <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
             <optional>true</optional>
          </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</artifactId>
          </dependency>
       </dependencies>
    
       <build>
          <plugins>
             <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
             </plugin>
          </plugins>
       </build>
    
    
    </project>

     

  2. 配置文件如下:

    server.port=8089
    spring.kafka.bootstrap-servers=ip:port
    spring.kafka.producer.retries= 0
    spring.kafka.producer.batch-size=16384
    spring.kafka.producer.buffer-memory=33554432
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.linger.ms=1

     

  3. 然后新建一个Producer类

    package com.example.kafkaproducer;
    
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;
    
    @Component
    public class KafkaProducer {
    
    
        @Autowired
        KafkaTemplate kafkaTemplate;
    
    
        public void produce(){
            kafkaTemplate.send("test","hello word");
            System.out.println("发送消息");
        }
    }

     

  4. 在test类中调用

    package com.example.kafkaproducer;
    
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class KafkaproducerApplicationTests {
    
    
       @Autowired KafkaProducer kafkaProducer;
    
    
       @Test
       public void contextLoads() {
          kafkaProducer.produce();
       }
    
    }

     

  5. 然后控制台就会打印一个莫名奇妙的错误,没有打印任何堆栈信息,大概意思只是表达了连接不上。

    Exception thrown when sending a message with key='null' and payload='' to topic

     

  6. telnet ip+port 是可以通的

  7. 随后发现,xshell上启动的kafka-server在报这样一个错,更详细的没有留存。

    ERROR Closing socket for /127.0.0.1 because of error (kafka.network.Processor)
    kafka.common.KafkaException: Wrong request type 18

     

  8. 百度了一下,很可能是Linux上的kafka版本和pom中引入的spring-kafka依赖不匹配造成的,于是查看对应关系

  9. 查看kafka,发现装的是一个0.8.2.1 版本的kafka,该版本的kafka是2015年3月发布的版本,可以说是十分古老,真是不知道为什么当初要选这么老的版本。

  10. 换了几次spring-kafka的pom之后,依然在报这个问题,于是我选择换更新的kafka的包。

  11. 换了2.2.0版本kafka的包,问题得到解决。

  12. 其中consumer的创建命令和老版本的不太一样,且consumer和producer需使用相同的端口号,而不是像之前producer配置为broker的端口,consumer配置为zookeeper的端口号。

    ./bin/kafka-console-consumer.sh --bootstrap-server ip:9092  --topic test

     

  13. 且config文件夹下server.properties文件中的一些配置和之前不太一样,需要注意的是,以下两行配置原来是被注解了的,需要在这里取消掉注解,并配置自己的ip。

    listeners = PLAINTEXT://your.host.name:9092
    advertised.listeners=PLAINTEXT://your.host.name:9092

     

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐