自学内容网 自学内容网

SpringBoot 整合 Avro 与 Kafka 详解

SpringBoot 整合 Avro 与 Kafka 详解

在大数据处理和实时数据流场景中,Apache Kafka 和 Apache Avro 是两个非常重要的工具。Kafka 作为一个分布式流处理平台,能够高效地处理大量数据,而 Avro 则是一个用于序列化数据的紧凑、快速的二进制数据格式。将这两者结合,并通过 Spring Boot 进行整合,可以构建出高效、可扩展的实时数据处理系统。

一、环境准备

在开始整合之前,需要准备好以下环境:

  • Java:确保已经安装了 JDK,推荐使用 JDK 8 或更高版本。
  • Maven:用于管理项目的依赖和构建过程。
  • Spring Boot:作为项目的框架,推荐使用较新的版本,如 Spring Boot 2.x。
  • Kafka:确保 Kafka 已经安装并运行,可以使用 Docker 部署 Kafka 集群。
  • Avro:Avro 依赖 JSON 定义的架构来序列化数据。
二、项目结构

一个典型的 Spring Boot 项目结构可能如下:

spring-boot-kafka-avro
├── src
│   ├── main
│   │   ├── java
│   │   │   └── com
│   │   │       └── example
│   │   │           ├── SpringBootKafkaAvroApplication.java
│   │   │           ├── config
│   │   │           │   └── KafkaConfig.java
│   │   │           ├── producer
│   │   │           │   └── KafkaProducer.java
│   │   │           ├── consumer
│   │   │           │   └── KafkaConsumer.java
│   │   │           └── model
│   │   │               └── ElectronicsPackage.java (由 Avro 自动生成)
│   │   ├── resources
│   │   │   ├── application.properties
│   │   │   └── avro
│   │   │       └── electronicsPackage.avsc (Avro 架构文件)
│   └── test
│       └── java
│           └── com
│               └── example
│                   └── SpringBootKafkaAvroApplicationTests.java
└── pom.xml
三、添加依赖

pom.xml 文件中添加必要的依赖:

<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    
    <!-- Spring Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>2.9.13</version> <!-- 根据需要选择合适的版本 -->
    </dependency>
    
    <!-- Avro -->
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.11.0</version> <!-- 根据需要选择合适的版本 -->
    </dependency>
    
    <!-- Avro Maven Plugin -->
    <plugin>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro-maven-plugin</artifactId>
        <version>${avro.version}</version>
        <executions>
            <execution>
                <phase>generate-sources</phase>
                <goals>
                    <goal>schema</goal>
                </goals>
                <configuration>
                    <sourceDirectory>${project.basedir}/src/main/resources/avro/</sourceDirectory>
                    <outputDirectory>${project.build.directory}/generated/avro</outputDirectory>
                </configuration>
            </execution>
        </executions>
    </plugin>
</dependencies>
四、定义 Avro 架构

src/main/resources/avro/ 目录下创建一个 Avro 架构文件 electronicsPackage.avsc

{
    "namespace": "com.example.model",
    "type": "record",
    "name": "ElectronicsPackage",
    "fields": [
        {"name": "package_number", "type": ["string", "null"], "default": null},
        {"name": "frs_site_code", "type": ["string", "null"], "default": null},
        {"name": "frs_site_code_type", "type": ["string", "null"], "default": null}
    ]
}

这个架构文件定义了 ElectronicsPackage 类,包括三个字段:package_numberfrs_site_codefrs_site_code_type

五、生成 Avro 类

运行 Maven 构建过程,Avro Maven 插件会根据 electronicsPackage.avsc 文件生成相应的 Java 类 ElectronicsPackage.java

六、配置 Kafka

application.properties 文件中配置 Kafka 的相关属性:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=com.example.config.AvroSerializer
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=com.example.config.AvroDeserializer

注意,这里指定了自定义的 AvroSerializerAvroDeserializer 类。

七、实现 Avro 序列化器和反序列化器

创建 AvroSerializerAvroDeserializer 类,用于 Avro 数据的序列化和反序列化。

// AvroSerializer.java
package com.example.config;

import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecord;
import org.apache.kafka.common.serialization.Serializer;

import java.io.ByteArrayOutputStream;
import java.io.IOException;

public class AvroSerializer<T extends SpecificRecord> implements Serializer<T> {
    private final DatumWriter<T> writer;

    public AvroSerializer(Class<T> type) {
        this.writer = new SpecificDatumWriter<>(type);
    }

    @Override
    public byte[] serialize(String topic, T data) {
        if (data == null) {
            return null;
        }
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
        try {
            writer.write(data, encoder);
            encoder.flush();
            out.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return out.toByteArray();
    }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // No-op
    }

    @Override
    public void close() {
        // No-op
    }
}

// AvroDeserializer.java
package com.example.config;

import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificRecord;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.Map;

public class AvroDeserializer<T extends SpecificRecord> implements Deserializer<T> {
    private final Class<T> type;
    private final DatumReader<T> reader;

    public AvroDeserializer(Class<T> type) {
        this.type = type;
        this.reader = new SpecificDatumReader<>(type);
    }

    @Override
    public T deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }
        ByteArrayInputStream in = new ByteArrayInputStream(data);
        Decoder decoder = DecoderFactory.get().binaryDecoder(in, null);
        try {
            return reader.read(null, decoder);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // No-op


原文地址:https://blog.csdn.net/jun778895/article/details/144422902

免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!