Spring Webflux + Reactive Kafka (2) - Consumer

1. 개요

첫번째 포스트로 프로듀서 구성에 대한 내용을 정리했고 이번엔 컨슈머를 구성한다. 이번 역시 Reactive Kafka 를 통하여 컨슈머를 구성할 예정이고 앞서 구성한 프로듀서와 연동하여 실제 메시지를 받는것까지 구현해보려 한다.

전체 소스코드 : Github 링크

---

2. 카프카 토픽 구성내용

현재 개인 서버 카프카에 구성된 토픽은 총 2개로 deogicorgi-message, deogicorgi-uri 가 존재한다. 각 토픽의 파티션은 1개씩 구성하였으며 그렇기 때문에 하나의 토픽에는 하나의 컨슈머만 붙일 수 있는 상태이다. (파티션과 컨슈머의 관계는 더 공부 해봐야 한다.)

[appuser@d97c9a0c3c7e ~]$ kafka-topics --list --bootstrap-server localhost:9092
__consumer_offsets
deogicorgi-message
deogicorgi-uri
[appuser@d97c9a0c3c7e ~]$

정상적으로 카프카 토픽이 존재함을 확인할 수 있다. (도커 컨테이너 내에서 수행한 커맨드이다.)

---

3. 컨슈머 설정(Consumer)

우선 메시지를 받는 기능만 구성하기 때문에 간단하게 설정하였다. 차후 메시지를 다시 프로듀싱하거나 RDBMS 또는 NoSQL에 저장하는 로직 역시 추가할 예정이다. 현재 계획하는 전체적인 플로우는 아래와 같다.

프로듀서 - 컨슈머 - 저장매체 or 프로듀서 (이 모든과정이 논블로킹으로 구성되어야 한다)

3-1. application.yml

카프카가 설치된 개인서버 주소 및 메시지를 받을 토픽, 컨슈머의 그룹 아이디 등을 설정한다.

kafka:
  hosts: deogicorgi.home:29092
  receiver :
    uri:
      name : deogicorgiUri
      topic : deogicorgi-uri
      groupId : deogicorgi-uri-1
    message:
      name : deogicorgiMessage
      topic : deogicorgi-message
      groupId : deogicorgi-message-1

3-2. KafkaProperties.java

위 의 application.yml 과 매핑될 클래스이다. yml 파일 내 receiver 2개를 하나의 맵으로 받도록 구성했다.

@Getter
@Setter
@Component
@EnableConfigurationProperties
@ConfigurationProperties("kafka")
public class KafkaProperties {

    // 카프카 호스트
    private String hosts;

    // 리시버 프로퍼티 맵
    private Map<String, KafkaReceiverProperty> receiver = new HashMap<>();

    public void setReceiver(Map<String, KafkaReceiverProperty> receivers) {
        this.receiver = receivers;
    }

    public Optional<Map.Entry<String, KafkaReceiverProperty>> getProperty(String key) {
        return this.receiver.entrySet()
                .stream().filter(entry -> entry.getValue().getName().equals(key))
                .findFirst();
    }
}

3-3. KafkaReceiverProperty.java

위 의 KafkaProperties 에서 리시버 맵에 사용되는 모델클래스이다.

/**
 * 카프카 리시버 설정 프로퍼티
 */
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class KafkaReceiverProperty {

    // 리시버 이름
    private String name;

    // 담당할 토픽
    private String topic;

    // 리시버 그룹 아이디
    private String groupId;
}

3-4. KafkaConfig.java

두개의 토픽을 각각 컨슈밍하는 리시버들를 빈으로 등록했다. 각각의 빈들은 설정된 토픽의 메시지만 담당하여 받게될 것이고 각 토픽들의 파티션은 1개로 구성하였기 때문에 싱글 스레드로 동작하게 될것이다. 여기서는 리시버당 토픽을 1개씩만 지정하였지만 하나의 리시버에 여러개의 토픽을 지정 할 수 도있다. 그렇게 되면 등록된 여러개의 토픽을 싱글 스레드로 받게된다.

@Configuration
@RequiredArgsConstructor
public class KafkaConfig {

    private final KafkaProperties properties;

    // deogicorgi-uri 리시버
    @Bean("uriMessageReceiver")
    public KafkaReceiver<Integer, String> uriMessageReceiver() throws Exception {
        Map.Entry<String, KafkaReceiverProperty> deogicorgiUri = properties.getProperty("deogicorgiUri").orElse(null);

        if (ObjectUtils.isEmpty(deogicorgiUri)) {
            throw new Exception("property is null");
        }

        KafkaReceiverProperty property = deogicorgiUri.getValue();

        ReceiverOptions<Integer, String> receiverOptions =
                ReceiverOptions.<Integer, String>create(getConsumerProps(property))
                        .subscription(Collections.singleton(property.getTopic()));

        return KafkaReceiver.create(receiverOptions);
    }

    // deogicorgi-message 리시버
    @Bean("messageReceiver")
    public KafkaReceiver<Integer, String> messageReceiver() throws Exception {
        Map.Entry<String, KafkaReceiverProperty> deogicorgiUri = properties.getProperty("deogicorgiMessage").orElse(null);

        if (ObjectUtils.isEmpty(deogicorgiUri)) {
            throw new Exception("property is null");
        }

        KafkaReceiverProperty property = deogicorgiUri.getValue();

        ReceiverOptions<Integer, String> receiverOptions =
                ReceiverOptions.<Integer, String>create(getConsumerProps(property))
                        .subscription(Collections.singleton(property.getTopic()));

        return KafkaReceiver.create(receiverOptions);
    }

    /******************************************************************
     ************************ Consumer Options ************************
     ******************************************************************/

    // 컨슈머 옵션
    private Map<String, Object> getConsumerProps(KafkaReceiverProperty property) {
        return new HashMap<>() {{
            put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getHosts());
            put(ConsumerConfig.GROUP_ID_CONFIG, property.getGroupId());
            put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
            put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 1000);
        }};
    }
}

3-5. KafkaMessageReceiver.java

빈으로 등록된 리시버들을 제공받아 실제로 컨슈밍하는 코드이다. KafkaMessageReceiver 는 빈으로 등록 될 때 정의된 모든 리시버를 받아와 컨슈밍을 시작한다.
스프링 빈의 라이프사이클을 알고있다면 해당 클래스가 생성되고 제거될때 수행되는 메서드들을 오버라이딩 하여 전처리 및 후처리를 진행 할 수 있다.

/**
 * 카프카 리시버
 */
@Slf4j
@Component
public class KafkaMessageReceiver {

    /**
     * KafkaMessageReceiver가 생성될 때 모든 카프카 리시버 시작
     */
    public KafkaMessageReceiver(List<KafkaReceiver<Integer, String>> kafkaReceivers) {
        for (KafkaReceiver<Integer, String> receiver : kafkaReceivers) {
            this.start(receiver);
        }
    }

    public void start(KafkaReceiver<Integer, String> receiver) {
        receiver.receive().subscribe(record -> {
            log.info("Kafka Reciever result : Topic >> [{}], message >> [{}], Offset >> [{}]", record.topic(), record.value(), record.receiverOffset());
        });
    }
}

3-6. 로그

Producer의 경우 모든 메시지를 싱글 스레드로 처리 한다. 각각 의 토픽에 하나의 메시지를 보냈을 경우 kafka-producer-network-thread 라는 이름의 스레드를 통하여 처리된다. 해당 스레드는 기본값으로, 다른 스레드풀을 사용하고 싶다면 설정에서 사용하고싶은 스레드를 등록해주면 된다.

Consumer의 경우 각각의 리시버가 다른 스레드를 사용하는 것을 확인할 수 있다. reactive-kafka-* 로 시작하는 스레드를 각각 사용하고 있으며 현재 각 토픽의 파티션이 1개씩이므로 해당 스레드만 계속 사용하여 처리할 것이다. 이후 각 토픽의 파티션을 늘리면 각 리시버들은 등록된 쓰레드풀을 통하여 멀티쓰레드로 메시지를 처리하게 될것이다.