Kafka整合SpringBoot

准备工作

假设你了解过 SpringBoot 和 Kafka。

1、SpringBoot

如果对 SpringBoot 不了解的话，建议去看看 DD 大佬 和 纯洁的微笑 的系列博客。

2、Kafka

Kafka 的话可以看看的博客 ： Kafka 安装及快速入门 ，学习的话自己开台虚拟机自己手动搭建环境吧，有条件的买服务器。

3、版本

  • Kafka服务版本为 kafka_2.11-1.1.0 (Scala)， 也就是1.1.0

  • Kafka的spring-kafka依赖我用的是1.1.1.RELEASE

先启动zookeeper，然后启动kafka

  • 进入到/opt/kafka/kafka_2.11-1.1.0目录下启动kafka自带的zookeeper

    • bin/zookeeper-server-start.sh config/zookeeper.properties

 

 

  • 启动kafka之前先vim config/server.properties，把server.properties中 advertised。listeners=PLAINTEXT://IP:9092 的IP替换成你服务器IP（这个ip会对应springboot项目的spring.kafka.bootstrap-servers=192.168.25.128:9092）

 

  • 用下面命令启动kafka,9092端口

    • bin/kafka-server-start.sh config/server.properties

 

注意！！！

启动完zookeeper和kafka，记得将你的虚拟机或者服务器关闭防火墙或者开启 Kafka 的端口 9092。不然的话会报错，连不上虚拟机的kafka服务

 

创建SpringBoot项目

  • 选择这个快速创建项目

 

 

  • 勾选web

 

 

项目整体架构：

 

1、pom 文件代码如下：

 <?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.sunment</groupId>
<artifactId>kafkatest02</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>

<name>kafkatest02</name>
<description>Demo project for Spring Boot</description>

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.10.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</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.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>
    <!--kafka start
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.11.0.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>0.11.0.1</version>
    </dependency>-->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>1.1.1.RELEASE</version>
    </dependency>

    <dependency>
        <groupId>com.google.code.gson</groupId>
        <artifactId>gson</artifactId>
        <version>2.8.2</version>
    </dependency>

</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
        </plugin>
    </plugins>
</build>
 </project>

主要引入了 spring-kafka 、lombok 、 gson 依赖

2、消息实体类 Message.java 如下：

@Data
public class Message {

private Long id;    //id

private String msg; //消息

private Date sendTime;  //时间戳
}

3、消息发送类 KafkaSender.java

@Component
@Slf4j
public class KafkaSender {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

private Gson gson = new GsonBuilder().create();

//发送消息方法
public void send() {
    Message message = new Message();
    message.setId(System.currentTimeMillis());
    message.setMsg(UUID.randomUUID().toString());
    message.setSendTime(new Date());
    log.info("+++++++++++++++++++++  message = {}", gson.toJson(message));
    kafkaTemplate.send("yang", gson.toJson(message));
}

}

这里关键的代码为 kafkaTemplate.send() 方法，yang 是 Kafka 里的 topic ，这个 topic 在 Java 程序中是不需要提前在 Kafka 中设置的，当然提前在kafka创建这个topic也可以，因为它会在发送的时候自动创建你设置的 topic， gson.toJson(message) 是消息内容。

4、消息接收类 KafkaReceiver.java

@Component
@Slf4j
public class KafkaReceiver {
@KafkaListener(topics = {"yang"})
public void listen(ConsumerRecord<?, ?> record) {

    Optional<?> kafkaMessage = Optional.ofNullable(record.value());

    if (kafkaMessage.isPresent()) {

        Object message = kafkaMessage.get();

        log.info("----------------- record =" + record);
        log.info("------------------ message =" + message);
    }

}

}

客户端 consumer 接收消息特别简单，直接用 @KafkaListener 注解即可，并在监听中设置监听的 topic ，topics 是一个数组所以是可以绑定多个主题的，上面的代码中修改为 @KafkaListener(topics = {"yang","tian"}) 就可以同时监听两个 topic 的消息了。需要注意的是：这里的 topic 需要和消息发送类 KafkaSender.java 中设置的 topic 一致。注意@Component注解，不然扫描不到@KafkaListener。

5、启动类 KafkaApplication.java

@SpringBootApplication
public class Kafkatest02Application {

public static void main(String[] args) {

    ConfigurableApplicationContext context = SpringApplication.run(Kafkatest02Application.class, args);

    KafkaSender sender = context.getBean(KafkaSender.class);

    for (int i = 0; i < 3; i++) {
        //调用消息发送类中的消息发送方法
        sender.send();

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}


}

调用生产者发信息方法，发送3条信息。

6、配置文件 application.properties

#============== kafka ===================

# 指定kafka 代理地址，可以多个

spring.kafka.bootstrap-servers=192.168.25.128:9092

#=============== provider =======================

spring.kafka.producer.retries=0

# 每次批量发送消息的数量

spring.kafka.producer.batch-size=16384

spring.kafka.producer.buffer-memory=33554432

# 指定生产者消息key和消息体的编解码方式

spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

#=============== consumer =======================

# 指定默认消费者group id

spring.kafka.consumer.group-id=test-consumer-group

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

spring.kafka.consumer.enable-auto-commit=true

spring.kafka.consumer.auto-commit-interval=100

# 指定消费者消息key和消息体的编解码方式

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

server.port=8888

192.168.25.128:9092 这个地址要和虚拟机里面的对应。

运行项目

 

 

整合成功

然后看下 Kafka 中的 topic 列表

  • bin/kafka-topics.sh --list --zookeeper localhost:2181

 

表示刚刚再发信息的时候自动创建了yang topic。

再创建一个controller包，再创建一个TestKafkaProducerController生产者类

/**
 * 测试kafka生产者
 */
@RestController
@RequestMapping("kafka")
public class TestKafkaProducerController {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

@RequestMapping("send")
public String send(String msg){
    kafkaTemplate.send("test_topic", msg);
    return "success";
}
}

在comsumer包下，再创建一个TestConsumer消费者类

/**
 * kafka消费者测试
 */
@Component
public class TestConsumer {
@KafkaListener(topics = "test_topic")
public void listen (ConsumerRecord<?, ?> record) throws Exception {
    System.out.printf("topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
}

}

然后重启项目,在浏览器输入地址

  • http://localhost:8888/kafka/send?msg=hello

 

控制台输出

 

换一个参数

  • http://localhost:8888/kafka/send?msg=你好吗

 

控制台又输出一条信息，topic也创建成功。

 

# kafka
Logo
Kafka开源项目指南

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

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南