大数据-56 Kafka SpringBoot与Kafka 基础简单配置和使用 Java代码 POM文件

本文涉及的产品
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: 大数据-56 Kafka SpringBoot与Kafka 基础简单配置和使用 Java代码 POM文件

点一下关注吧!!!非常感谢!!持续更新!!!

目前已经更新到了:

Hadoop(已更完)

HDFS(已更完)

MapReduce(已更完)

Hive(已更完)

Flume(已更完)

Sqoop(已更完)

Zookeeper(已更完)

HBase(已更完)

Redis (已更完)

Kafka (正在更新…)

章节内容

上节我们完成了:


topics.sh、producer.sh、consumer.sh 脚本的基本使用

pom.xml 配置

JavaAPI的使用:producer 和 consumer

简单介绍

在Spring Boot中使用Kafka,是构建分布式消息驱动应用程序的一种常见方法。Kafka的强大之处在于其高吞吐量、低延迟和良好的可扩展性,非常适合处理大量实时数据。


Kafka的基本概念

Producer(生产者): 负责向Kafka的主题(topic)发送消息。

Consumer(消费者): 从Kafka的主题中读取消息。

Broker(代理): Kafka集群中的节点,负责消息的存储和传输。

Topic(主题): 类似于消息队列的概念,用于分类和组织消息。一个topic可以有多个分区(partition),每个分区是一个日志(log)。

Partition(分区): Kafka中的主题被分成多个分区,每个分区内部的消息是有序的,但分区之间是无序的。

Consumer Group(消费者组): 一组消费者组成的一个逻辑订阅者,保证每条消息在消费者组中只被一个消费者消费。

spring-kafka

Spring-Kafka 是 Spring 框架对 Apache Kafka 的集成,使得在 Spring 应用中使用 Kafka 更加简便和直观。它提供了一系列功能和配置选项来帮助开发者快速构建基于消息驱动的微服务架构。


KafkaTemplate

KafkaTemplate 是 Spring-Kafka 提供的用于发送消息的核心类。它简化了生产者与 Kafka 交互的过程。你可以通过这个类轻松地将消息发送到 Kafka 的主题中。


KafkaListener

@KafkaListener 是用于消费 Kafka 消息的注解。通过这个注解,可以非常方便地定义消息消费者,处理从指定主题接收到的消息。


Spring-Kafka 的配置

Spring-Kafka 支持通过配置文件来配置 Kafka 客户端的属性。这些配置可以在 application.properties 或 application.yml 中指定。


架构图

上节已经出现过了,这里再放一次

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>org.example</groupId>
    <artifactId>springboot-kafka</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </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>
    </dependencies>

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

</project>

配置文件

我们常见的配置文件如下图:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    template:
      default-topic: my-topic

Producer

编写代码

编写了一个KafkaProducerController

里边写了两个方法,都是使用了 KafkaTemplate 的工具。

@RestController
public class KafkaProducerController {

    @Resource
    private KafkaTemplate<Integer, String> kafkaTemplate;

    @RequestMapping("/sendSync/{message}")
    public String sendSync(@PathVariable String message) {
        ProducerRecord<Integer, String> record = new ProducerRecord<>("wzk_topic_test", 0, 1, message);
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(record);
        try {
            SendResult<Integer, String> result = future.get();
            System.out.println(result.getProducerRecord().key() + "->" +
                    result.getProducerRecord().partition() + "->" +
                    result.getProducerRecord().timestamp());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "Success";
    }

    @RequestMapping("/sendAsync/{message}")
    public String sendAsync(@PathVariable String message) {
        ProducerRecord<Integer, String> record = new ProducerRecord<>("wzk_topic_test", 0, 2, message);
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(record);
        future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送失败!");
                ex.printStackTrace();
            }

            @Override
            public void onSuccess(SendResult<Integer, String> result) {
                System.out.println("发送成功");
                System.out.println(result.getProducerRecord().key() + "->" +
                        result.getProducerRecord().partition() + "->" +
                        result.getProducerRecord().timestamp());
            }
        });
        return "Success";
    }

}

测试结果

http://localhost:8085/sendSync/wzktest1
http://localhost:8085/sendAsync/wzktest2
http://localhost:8085/sendAsync/wzktest222222

我们观察控制台的效果如下:

Consumer

编写代码

编一个类来实现Consumer:

@Configuration
public class KafkaConsumer {

    @KafkaListener(topics = {"wzk_topic_test"})
    public void consume(ConsumerRecord<Integer, String> consumerRecord) {
        System.out.println(
                consumerRecord.topic() + "\t"
                        + consumerRecord.partition() + "\t"
                        + consumerRecord.offset() + "\t"
                        + consumerRecord.key() + "\t"
                        + consumerRecord.value());
    }

}

测试运行

2024-07-12 13:48:46.831  INFO 15352 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-1, groupId=wzk-test] Setting offset for partition wzk_topic_test-0 to the committed offset FetchPosition{offset=13, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=h121.wzk.icu:9092 (id: 0 rack: null), epoch=0}}
2024-07-12 13:48:46.926  INFO 15352 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : wzk-test: partitions assigned: [wzk_topic_test-0]
wzk_topic_test  0 13  1 wzktest
wzk_topic_test  0 14  2 wzktest222
wzk_topic_test  0 15  2 wzktest222222

控制台的截图如下:

目录
相关文章
|
6天前
|
Java 应用服务中间件
SpringBoot获取项目文件的绝对路径和相对路径
SpringBoot获取项目文件的绝对路径和相对路径
41 1
SpringBoot获取项目文件的绝对路径和相对路径
|
16天前
|
XML Java Kotlin
springboot + minio + kkfile实现文件预览
本文介绍了如何在容器中安装和启动kkfileviewer,并通过Spring Boot集成MinIO实现文件上传与预览功能。首先,通过下载kkfileviewer源码并构建Docker镜像来部署文件预览服务。接着,在Spring Boot项目中添加MinIO依赖,配置MinIO客户端,并实现文件上传与获取预览链接的接口。最后,通过测试验证文件上传和预览功能的正确性。
springboot + minio + kkfile实现文件预览
|
5天前
|
存储 前端开发 JavaScript
|
5天前
|
存储 Java API
|
3天前
|
消息中间件 缓存 Java
java nio,netty,kafka 中经常提到“零拷贝”到底是什么?
零拷贝技术 Zero-Copy 是指计算机执行操作时,可以直接从源(如文件或网络套接字)将数据传输到目标缓冲区, 而不需要 CPU 先将数据从某处内存复制到另一个特定区域,从而减少上下文切换以及 CPU 的拷贝时间。
java nio,netty,kafka 中经常提到“零拷贝”到底是什么?
|
8天前
|
Java 数据库连接 数据库
如何构建高效稳定的Java数据库连接池,涵盖连接池配置、并发控制和异常处理等方面
本文介绍了如何构建高效稳定的Java数据库连接池,涵盖连接池配置、并发控制和异常处理等方面。通过合理配置初始连接数、最大连接数和空闲连接超时时间,确保系统性能和稳定性。文章还探讨了同步阻塞、异步回调和信号量等并发控制策略,并提供了异常处理的最佳实践。最后,给出了一个简单的连接池示例代码,并推荐使用成熟的连接池框架(如HikariCP、C3P0)以简化开发。
21 2
|
8天前
|
消息中间件 存储 Prometheus
Kafka集群如何配置高可用性
Kafka集群如何配置高可用性
|
19天前
|
安全 Java 数据安全/隐私保护
如何配置 Java 安全管理器来避免访问控制异常
配置Java安全管理器以防止访问控制异常,需在启动JVM时通过 `-Djava.security.manager` 参数启用,并设置安全策略文件,定义权限规则,限制代码执行操作,确保应用安全。
|
7天前
|
Java
SpringBoot获取文件将要上传的IP地址
SpringBoot获取文件将要上传的IP地址
22 0
|
13天前
|
缓存 Java 程序员
Java|SpringBoot 项目开发时,让 FreeMarker 文件编辑后自动更新
在开发过程中,FreeMarker 文件编辑后,每次都需要重启应用才能看到效果,效率非常低下。通过一些配置后,可以让它们免重启自动更新。
22 0