Apache Kafka를 활용한 이벤트 기반 Spring Boot 마이크로서비스 구축 — Part1

spring.threads.virtual.enabled=true

# src/main/resources/application.properties
spring.kafka.bootstrap-servers=localhost:9092

package com.example.kafkaapp;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;
import java.util.stream.Stream;

import static java.lang.StringTemplate.STR;

@Component
public class TextDataProducer {
    private static final Logger logger = Logger.getLogger(TextDataProducer.class.getName());

    // 토픽 설정을 위한 상수
    private static final int PARTITION_COUNT = 8;
    private static final String TOPIC = "TEXT-DATA";
    private static final short REPLICATION_FACTOR = 1;

    private final KafkaTemplate<String, String> kafkaTemplate;

    public TextDataProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Autowired
    public void configureTopic(KafkaAdmin kafkaAdmin) {
        kafkaAdmin.createOrModifyTopics(new NewTopic(TOPIC, PARTITION_COUNT, REPLICATION_FACTOR));
    }

    private void sendTextMessage(String text, int lineIndex) {
        if (text == null || text.isEmpty()) {
            return;
        }
        // 텍스트 메시지를 라인 인덱스에 따라 파티션에 분배하여 토픽으로 보냅니다.
        kafkaTemplate.send(TOPIC, "KEY-" + (lineIndex % PARTITION_COUNT), text);
    }

    public void sendContentOf(File file) {
        Instant before = Instant.now();
        try (Stream<String> lines = Files.lines(file.toPath())) {
            AtomicInteger counter = new AtomicInteger();
            lines.forEach(line -> sendTextMessage(line, counter.getAndIncrement()));
            Instant after = Instant.now();
            Duration duration = Duration.between(before, after);
            logger.info(STR."스트리밍된 라인 수: \{counter.get()}개, 소요 시간: \{duration.toMillis()} 밀리초");
        } catch (IOException e) {
            throw new RuntimeException("파일 읽기 오류", e);
        }
    }
}

package com.example.kafkaapp;

import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;

@RestController
public class TextDataController {

    private final TextDataProducer producer;

    public TextDataController(TextDataProducer producer) {
        this.producer = producer;
    }

    @PostMapping("/upload")
    public ResponseEntity<String> uploadTextFile(@RequestParam("file") MultipartFile file) {
        try {
            Path tempFile = Files.createTempFile(file.getOriginalFilename(), null);
            file.transferTo(tempFile);
            Thread.ofVirtual().start(() -> producer.sendContentOf(tempFile.toFile()));
            return ResponseEntity.ok(tempFile.toString());
        } catch (IOException e) {
            return ResponseEntity
            .status(HttpStatus.INTERNAL_SERVER_ERROR)
            .body("파일 업로드 중 오류가 발생했습니다.");
        }
    }
}

package com.example.kafkaapp;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class TextProducerApplication {

    public static void main(String[] args) {
        SpringApplication.run(TextProducerApplication.class, args);
    }
}

./gradlew bootRun

###텍스트와 파일 필드가 있는 폼 전송
POST http://localhost:8080/upload HTTP/1.1
Content-Type: multipart/form-data; boundary=boundary

--boundary
Content-Disposition: form-data; name="file"; filename="shakespeares.txt"

// 'shakespeares.txt' 파일이 업로드됩니다
< /Users/mustafaguc/Desktop/kafka-demo-content/shakespeares.txt

--boundary

FROM openjdk:21-slim
WORKDIR /app
COPY build/libs/text-producer-1.0.jar app.jar
EXPOSE 8080

ENTRYPOINT ["java", "-jar", "app.jar"]

./gradlew assemble

docker build . -t text-producer

docker run -it -p 8080:8080 text-producer

c.e.kafkaapp.TextProducerApplication     : Starting TextProducerApplication using Java 21.0.1 with PID 4871 (/Users/mustafaguc/projects/java/text-producer/build/classes/java/main started by mustafaguc in /Users/mustafaguc/projects/java/text-producer)
c.e.kafkaapp.TextProducerApplication     : No active profile set, falling back to 1 default profile: "default"
o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port 8080 (http)
o.apache.catalina.core.StandardService   : Starting service [Tomcat]
o.apache.catalina.core.StandardEngine    : Starting Servlet engine: [Apache Tomcat/10.1.17]
o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 307 ms
o.a.k.clients.admin.AdminClientConfig    : AdminClientConfig values:
........
o.a.k.clients.admin.AdminClientConfig    : These configurations '[sasl.jaas.config, idompotence.enabled]' were supplied but are not used yet.
o.a.kafka.common.utils.AppInfoParser     : Kafka version: 3.6.1
o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 5e3c2b738d253ff5
o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1708262321726
o.a.kafka.common.utils.AppInfoParser     : App info kafka.admin.client for adminclient-1 unregistered
o.apache.kafka.common.metrics.Metrics    : Metrics scheduler closed
o.apache.kafka.common.metrics.Metrics    : Closing reporter org.apache.kafka.common.metrics.JmxReporter
o.apache.kafka.common.metrics.Metrics    : Metrics reporters closed
o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port 8080 (http) with context path ''
c.e.kafkaapp.TextProducerApplication     : Started TextProducerApplication in 0.764 seconds (process running for 0.976)