kafka를 활용한 재고에 대한 동시성 처리기 - 1

패션 이커머스 도메인에 대한 프로젝트를 하면서 쿠버네티스로 배포를 진행했습니다.

 

근데 배포시에 pod 두개로 서버가 띄워지기 때문에, pod별로 값이 공유되지 않아, 주문시에 동시성 문제가 발생했었습니다.

 

참 난감했는데요.

 

구글링을 하면서 DB락, Redis 등 여러 방법이 있었지만, 저는 먼저 카프카를 이용해서 이 동시성 문제를 해결해볼까 했습니다.

 

나중에도 까먹지 않기 위해 이 해결했던 경험을 기록할까 합니다.

 

서버, 설정

저희의 서버는 pod 두개로 배포가 되어있습니다.

이 pod들은 값을 공유하지 않습니다. 따라서 

각 pod별로 이렇게 주문이 실행되게 될때 재고에 대한 값이 공유되지 않습니다. 따라서 kafka나 다른 기능을 사용해서 이런 값이 공유되지 않는 동시성 문제를 해결해야하는데요.

 

우선은 카프카를 k8s로 띄워두었습니다.

 

그 다음 이 띄워진 카프카와 제 어플리케이션을 연결해야합니다.

이는 application.properties 나 application.yml에서 연결 할 수 있습니다.

  kafka:
    bootstrap-servers: ${custom.kafka.host}:${custom.kafka.port}

 

그리고 우선은 카프카의 흐름을 이해하기 위해 브로커의 파티션을 하나로 설정하고 해보았습니다. 아주 기본적인 설정만 해보겠습니다.

@Configuration
public class KafkaConfig {

    @Value("${custom.kafka.devPort}")
    private String kafkaPort;

    @Bean
    public KafkaAdmin admin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:"+kafkaPort);
        return new KafkaAdmin(configs);
    }

    @Bean
    public NewTopic topic() {
        return new NewTopic("orders", 1, (short) 1);
    }
}

 

이제 주문시 메세지 발급을 해보겠습니다.

 

카프카

먼저 아셔야 할 것이,

Kafka는 크게 Producer, Broker, Consumer의 세 가지 주요 구성요소로 이루어져 있습니다.

 

Producer: 데이터를 생성하고 Kafka로 전송하는 역할을 합니다. 이는 서비스 로그, IoT 디바이스 데이터, 사용자 이벤트 등 다양한 형태의 데이터일 수 있습니다.

 

Consumer: Broker로부터 데이터를 가져와서 처리하는 역할을 합니다.

Broker: Kafka 클러스터를 구성하는 서버입니다. 데이터가 Producer로부터 전달되면 이를 저장하고, 필요 시 Consumer에게 제공합니다.

 

Kafka에서는 '토픽'이라는 개념을 사용하여 데이터를 분류하는데요. 각 Producer는 특정 토픽으로 데이터를 보내고, Consumer는 특정 토픽으로부터 데이터를 읽습니다. 

 

또한, Kafka 토픽은 여러 '파티션'으로 분할될 수 있습니다. 이는 병렬 처리를 가능하게 하고, 대량의 데이터를 처리할 수 있도록 합니다.

 

먼저 producer가 메세지를 발급하는 코드입니다.

kafkaTemplate.send("orders", String.valueOf(saveOrder.getId()), String.valueOf(saveOrder.getId()));

카프카는 특정 토픽으로 메세지를 발급할 수 있는데요.

 

해당 코드는 orders라는 이름의 메세지를  saveOrder의 키값을 가지고 saveOrder의 키값이라는 내용을 보내는 것입니다.

 

카프카에서는 키를 사용해서 특정 파티션에 차곡차곡 메세지를 쌓을 수 있는데요. 따라서 저희는 키값을 같은 값으로 보내야 해당 주문에 맞는 사람들 끼리만 동시성을 해결할 수 있다는 것입니다.

 

이값은 consumer에서 받습니다.

@KafkaListener(topicPattern = "orders", groupId = "group_id")
public void orderToOrder(@Payload String Id) throws JsonProcessingException {

@KafkaListener 어노테이션을 사용해서 이 메소드가 컨슈머 역할을 수행하도록 설정할 수 있습니다. 그럼 메세지는 이 메소드로 받을 수 있다는 얘기입니다.

 

topicPatten은 우리가 아까 발급했던 orders, 특정 토픽의 메세지를 받겠다는 것입니다.

 

@Payload 어노테이션은 아까 내용으로 보낸 값을 해당 매게 변수로 받겠다는 것을 나타내는 것입니다.

 

근데 어떻게 동시성 처리를 한다는 것일까요?

 

아까 처음에 우리는 파티션을 하나로 설정하고 카프카에서 메세지를 발급했습니다.

 

카프카는 특정 키값에 맞춰서 파티션에 메세지를 라우팅 해줍니다. 그러나 지금은 우리의 환경은 파티션이 하나라 어떤 키값을 가지더라도 값은 파티션에 메세지가 차곡차곡 쌓이게 됩니다. 따라서 순서성을 가지고 있는 것이지요. 

 

하지만 이것은 파티션이 하나일 경우에만 유효합니다. 파티션이 여러개 일경우에는 키값으로 구분이 되기때문에 매우 신중이 메세지의 순서를 유지할 요인을 생각해야합니다.

 

다음에는 kafka의 파티션이 여러개 일경우에 해결하는 방법을 작성해보겠습니다.