Kafka Consumer 멀티 쓰레드로 간단하게?

Multi Processor or Thread?

Consumer를 만들다 보면 성능의 문제에 도달하게 된다. 
Kafka의 성능은 매우매우 빠르지만 1대의 Consumer를 통해 데이터를 저장하는 성능에는 서버 스펙으로는 한계가 있기 때문이다.
물론, End Point(ES, RDB, MongoDB 등)의 성능에도 한계가 있어 어느 정도 성능이 나오면 그 이상은 힘들다.

 

보통 토픽의 파티션과 컨슈머의 갯수의 공식은 아래와 같다.

파티션 수 >= 컨슈머 수

최고의 성능을 위해서는 파티션 수와 컨슈머의 수를 맞춰주는게 제일 좋아 보인다. 

그렇다면 파티션 수와 컨슈머의 수를 1:1로 맞추기 위해서는 어떻게 해야 할까?

 

보통 이런 경우에는 2가지 해결책이 있다.

 

1. Multi Processor형태로 Consumer를 Scale out하여 여러대 서버로 늘릴 것인지?
2. 1대의 서버에서 Multi Thread형태로 실행 할 것인지?

 

첫번째 경우 성능이 제일 잘 나온다. 서버의 사양도 넉넉히 사용할 수 있고 다수의 서버에서 각각의 프로세스를 따로 띄우기 때문에 서버가 Down이 되어도 나머지 서버가 죽지 않아서 데이터를 계속 가져올 수 있는 장점 또한 있다. 

하지만 단점으로는 관리 포인트가 많아진다는 점이다. 팀에서 관리할 인원이 충분하지 않다면 위의 방식을 사용하기에 앞서 관리측면을 충분히 고민을 해보아야 한다.

나같은 경우에도 초기에 공부하면서 컨슈머를 만들어서 배포하였을 때 여러 서버에 1대씩 Scale out 형태로 띄웠었다가 현재는 2번으로 다시 재배포하였다.

 

두번째 경우 성능도 어느정도 잘 나오지만 1개의 프로세스만 관리하면 되기 때문에 관리측면에서 보면 아주 좋은 장점을 가지고 있다. 그리고 소스 또한 그리 어렵게 구현을 하지 않아도 된다. 

 

Spring kafka의 옵션 중에 Multi Thread를 지원하는 옵션이 있다. 바로 "concurrency" 옵션을 사용하면 된다.

 

예제 코드는 다음과 같다.

@KafkaListener(id = "${group.name}", topics = "${topic.name}", groupId = "${group.name}", concurrency = "5")
    public void receiveMessage(ConsumerRecords<String, String> records) {
        //message 처리
        records.forEach(record -> consumerService.process(record));
    }
}

 

위와 같이 셋팅을 한 뒤에 컨슈머를 기동해 보면 아래와 같이 5개의 파티션이 5개의 Thread로 할당을 받은 것을 확인 할 수 있다.