[Apache Kafka] Spring Boot와 연동하여 간단한 주문 시스템 만들기 (요청/응답 받기)

 

Spring Boot와 연동하여 간단한 주문 시스템 만들기 - (요청/응답 받기)

이제 스프링부트와 연동하여 아주 간단한(요청/응답 받기 수준) 주문 시스템을 만들 예정입니다. html, css, js 파일은 올려두었으니 혹시 필요하신분이 계시다면 다운로드 받아 스프링 부트 환경에 셋팅해주시면 됩니다. 참고로 현재 셋팅은 타임리프 기준입니다.

style.css

0.00MB

app.js

0.00MB

index.html

0.00MB

 

현재 저는 스프링 부트 3.5.3 버전을 사용중이며 아래와 build.gradle에 개발 시 필요한 web, devtool, lombok, thymeleaf 의존성을 주입 받고 카프카를 사용할 예정이기에 spring-kafka도 추가하여 주세요.

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.kafka:spring-kafka'
    developmentOnly 'org.springframework.boot:spring-boot-devtools'
    implementation 'org.springframework.boot:spring-boot-starter-thymeleaf'
    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.springframework.kafka:spring-kafka-test'
    testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
    
    providedRuntime 'org.springframework.boot:spring-boot-starter-tomcat'
}

 

간단하게 셋팅한 화면은 아래와 같습니다.

 

이후 application.properties에 설정을 셋팅해주세요.

spring.application.name=kafka-project

# Kafka Config
spring.kafka.bootstrap-servers=192.168.223.128:9092

# Consumer Config
spring.kafka.consumer.group-id=order-group
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=*

# Producer Config
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

 

만약 application.yml으로 사용중이시라면 아래와 같이 사용하시면 됩니다. 카프카 서버의 아이피는 이 전 포스팅에서 구축했던 아이피를 작성합니다.

2025.07.19 - [Development/Apache Kafka] - [Apache Kafka] 리눅스에 카프카 설치하기

spring:
  kafka:
    bootstrap-servers: 192.168.223.128:9092
    consumer:
      group-id: order-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

 

properties 기준 각각 설정은 다음과 같습니다.

1. spring.kafka.bootstrap-servers : 카프카 브로커 주소 지정 (Producer와 Consumer가 해당 주소를 통해 접속)
2. spring.kafka.consumer.group-id : 같은 그룹 ID를 가진 Consumer들은 메시지를 나누어 처리하도록 설정
3. spring.kafka.consumer.key-deserializer : (Consumer) 메시지 key 역직렬화 클래스 지정
4. spring.kafka.consumer.value-deserializer : (Consumer)  메시지 value JSON 역직렬화 클래스 지정
5. spring.kafka.consumer.properties.spring.json.trusted.packages : JSON 역직렬화 시 신뢰할 수 있는 클래스 패키지 지정 (현재는 단순 테스트 용도이므로 *으로 하였지만 실제 사용 시 필요한 패키지만 지정해주어야 합니다)
6. spring.kafka.producer.key-serializer : (Producer) 메시지 key 직렬화 클래스 지정
7. spring.kafka.producer.value-serializer : (Producer)  메시지 value JSON 직렬화 클래스 지정

※ 직렬화/역직렬화 흐름
Consumer : byte[] 수신 → JSON → Java 객체 (역직렬화)
Producet : Java 객체 → JSON → byte[] 전송 (직렬화)

 

저는 패키지 구조를 아래와 같이 잡았습니다.

 

domain/Order.java

추후 수신을 확인하기 위해 ToString 어노테이션도 추가해줍니다.

@Setter
@Getter
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class Order {
    private String orderId; // 주문 아이디
    private String itemName; // 상품명
    private int quantity; // 수량
}

 

producer/OrderProducer.java

KafkaTemplate은 메시지를 전송하기 위한 spring-kafka의 객체입니다. key=String, value=Order 형태의 메시지를 전송하겠다는 의미이며 send 메서드를 호출하며 첫번째 인자에는 토픽명(order-topic), 두번째 인자에는 전달할 객체를 작성해줍니다.

@Component
@RequiredArgsConstructor
public class OrderProducer {
    private final KafkaTemplate<String, Order> kafkaTemplate;

    public void send(Order order) {
        kafkaTemplate.send("order-topic", order);
    }
}

 

consumer/OrderConsumer.java

@KafkaListner 어노테이션은 메시지 수신 핸들러 역할을 수행하며 지정된 토픽과 그룹 아이디로 메시지를 자동 수신 합니다. 즉 토픽명이 order-topic이며 order-group 이라는 그룹명을 가진 상태로 수신합니다.

그룹명을 지정한 이유는 서버1, 서버2와 같이 분산 처리 되어있을 때 각각의 서버가 같은 그룹 아이디를 가지고 카프카 서버에 동일한 메시지를 수신할 수 있도록 하기 위함입니다. 또한 @KafkaListenr의 내부 동작으로 인해 주기적으로 메시지를 체크하고 수신합니다.

아래의 코드에서는 정상적으로 메시지를 수신했는지 확인하기 위해 println으로 콘솔에 출력합니다.

@Component
public class OrderConsumer {
    @KafkaListener(topics = "order-topic", groupId = "order-group")
    public void listen(Order order) {
        System.out.println("Consumed Order: " + order);
    }
}

 

controller/OrderController.java

/api/orders로 POST 요청이 온다면 위에서 만들었던 orderProducer의 send 메서드를 호출합니다.

@RestController
@RequiredArgsConstructor
@RequestMapping("/api/orders")
public class OrderController {
    private final OrderProducer orderProducer;

    @PostMapping
    public ResponseEntity<Void> postOrder(@RequestBody Order order) {
        orderProducer.send(order);
        return ResponseEntity.ok().build();
    }
}

 

이제 간단한 연동을 위한 코드는 모두 작성 되었으니 미리 만들어 두었던 카프카 서버를 실행해야 합니다. 우선, 하나의 쉘을 킨 뒤 아래와 같이 카프카 서버를 실행해줍니다.

cd /usr/local/kafka_2.12-3.9.0

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

 

이후 다른 쉘에서 토픽을 생성해 주세요. 각각 옵션의 의미는 아래와 같습니다.

1. --create : 토픽 생성 옵션
2. --topic : 토픽 이름 지정
3. --bootstrap-server : 브로커 주소 지정 (현재 카프카는 1대이므로 로컬 호스트로 진행)
4. --partitions : 파티션 수 설정
5. --replication-factor : 복제본(replica) 수 지정 (현재 단일 브로커 환경이므로 1로 설정)

cd /usr/local/kafka_2.12-3.9.0

bin/kafka-topics.sh --create \
  --topic order-topic \
  --bootstrap-server localhost:9092 \
  --partitions 6 \
  --replication-factor 1

 

파티션 수 설정에 관련된 계산 방법은 다음 포스팅에 이어집니다.

---

결과 확인

index 페이지를 띄워 주문 아이디와 상품명, 수량을 입력한 후 "주문 전송" 버튼을 클릭하면 다음과 같이 "주문이 전송되었습니다"가 나온 뒤 콘솔에 해당 내용이 출력되면 됩니다.

 

또는 카프카에서 kafka-console-consumer.sh를 사용하여 확인하는 방법도 있습니다.

bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.223.128:9092 \
  --topic 토픽명 \
  --from-beginning