Spring Boot에서 Apache Kafka 사용 ... 1/2

아파치 카프카(Apache Kafka)는 분산 스트리밍 플랫폼이며 데이터 파이프 라인을 만들 때 주로 사용되는 오픈소스 솔루션입니다. 이러한 카프카는 대용량의 실시간 로그처리에 특화되어 있는 솔루션이며 데이터를 유실 없이 안전하게 전달하는 것이 주목적인 메세지 시스템에서 Fault-Tolerant한 안정적인 아키텍처와 빠른 퍼포먼스로 데이터를 처리하는 용도로 사용되고 있습니다.

 

최근 MSA에서는 Pub/Sub 구조를 통한 데이터 동기화 솔루션으로 RabbitMQ와 함께 많이 사용되고 있습니다.

 

Apache Kafka의 특징

1. Pub/Sub 모델 : Publishe/Subscriber 모델은 데이터 큐를 중간에 두고 서로 간 독립적으로 데이터를 생산하고 소비하는 모델. 이런 느슨한 결합을 통해 Publisher나 Subscriber가 변경이나 장애시에도 서로 간에 의존성이 없으므로 안정적인 데이터 처리 가능.

2. 고가용성(High availability) 및 확장성(Scalability) : 카프카는 클러스터로서 작동함. 클러스터로서 작동하므로 Fault-tolerant 한 고가용성 서비스를 제공할 수 있고 분산 처리를 통해 빠른 데이터 처리를 가능하게 합니다. 또한 서버를 수평적으로 늘려 안정성 및 성능을 향상시키는 Scale-out이 가능

3. 디스크 순차 저장 및 처리(Sequential Store and Process in Disk) : 메세지를 메모리 큐에 적재하는 기존 메세지 시스템과 다르게 카프카는 메세지를 디스크에 순차적으로 저장

4. 서버에 장애가 나도 메세지가 디스크에 저장되어 있으므로 유실걱정이 상대적으로 적음

5. 디스크가 순차적으로 저장되어 있으므로 디스크 I/O가 줄어들어 성능이 향상됨

6. 분산 처리(Distributed Processing) : 카프카는 파티션(Partition)이란 개념을 도입하여 여러 개의 파티션을 서버들에 분산시켜 나누어 처리할 수 있음. 이로서 메세지를 상황에 맞추어 빠르게 처리할 수 있음

 

Kafka를 이용한 Pub/Sub 구현

 

Kafka를 이용한 Pub/Sub 구현

 

카프카는 위 Pub/Sub 구조에서 볼 수 있듯이 각각 3대의 Copy로 클러스터가 구성된zookeeper, Kafka Engine 그리고 이를 메시지 브로커로 이용하는 Publisher(Producer):메시지 생성자와 Subscriber(Consumer):메시지 소비자로 구성되어 있습니다.

 

카프카를 Java로 구성하기 위해서는 Message Broker를 담당하는 zookeeper와 kafka가 설치되어야 하며 메시지를 생성하는 publisher와 소비하는 consumer를 작성해야 합니다.

 

springboot의 dependency에 kafka 라이브러리를 추가해서 코딩할 수 있습니다. 해당하는 spring boot의 프로젝트는 하나의 프로젝트에서 publisher/subscriber 역할을 모두 담당할 수 있습니다. 즉 토픽이라는 단위를 이용해 메시지를 생성할 수 도 있고 다른 publisher가 발행하는 관심있는 topic을 소비할 수 있습니다.

 

Spring Boot에서 Kafka Publisher 설정 예

Spring Boot에 Apache Kafka Publisher 방식 설정

 

Kafka Publisher Configuration

• bootstrap.servers
  • 연결할 서버 정보. e.g. : host1:port1,host2:port2와 같이 여러개를 나열할 수 있음
  • 초기 커넥션 연결시에 사용하기 때문에, 모든 서버 리스트를 포함할 필요는 없음. (실제 메시지 전송시에는 새로운 커넥션을 맺은 다음에 전송하기 때문)
• key.serializer, value.serializer
  • 메시지를 serialize 할 때 사용할 클래스를 지정
  • ByteArraySerializer, StringSerializer 등등 Serializer를 implements한 클래스들이 있음
• partitioner.class
  • 어떤 파티션에 메시지를 전송할지 결정하는 클래스임
  • 기본값은 DefaultPartitioner이며 메시지 키의 해시값을 기반으로 전송할 파티션을 결정함
• acks
  • 프로듀서가 전송한 메시지를 카프카가 잘 받은 걸로 처리할 기준을 말함
  • 0, 1, all 값으로 세팅할 수 있으며 각각 메시지 손실률과 전송 속도에 대해 차이가 있음
  • 설정값 비교
    

    설정값	손실률	속도	설명
    acks = 0	높음	빠름	프로듀서는 서버의 확인을 기다리지 않고 메시지 전송이 끝나면 성공으로 간주합니다.
    acks = 1	보통	보통	카프카의 leader가 메시지를 잘 받았는지만 확인합니다.
    acks = all	낮음	느림	카프카의 leader와 follower까지 모두 받았는지를 확인합니다.

  • 기본값은 acks=1 
• buffer.memory
  • 프로듀서가 서버로 전송 대기중인 레코드를 버퍼링하는데 사용할 수 있는 메모리 양
  • 레코드가 서버에 전달될 수 있는 것보다더 빨리 전송되면 max.block.ms동안 레코드를 보내지 않음
  • 기본값은 33554432, 약 33MB임
• retries
  • 프로듀서가 에러가 났을때 다시 시도할 횟수를 말함
  • 0보다 큰 숫자로 설정하면 그 숫자만큼 오류 발생시에 재시도 함
• max.request.size
  • 요청의 최대 바이트 크기를 말합니다. 대용량 요청을 보내지 않도록 제한할 수 있음
  • 카프카 서버에도 별도로 설정할 수 있으므로 서로 값이 다를 수 있음
• connections.max.idle.ms
  • 지정한 시간 이후에는 idle 상태의 연결을 닫음
• max.block.ms
  • 버퍼가 가득 찼거나 메타데이터를 사용할 수 없을 때 차단할 시간을 정할 수 있음
• request.timeout.ms
  • 클라이언트가 요청 응답을 기다리는 최대 시간을 정할 수 있음
  • 정해진 시간 전에 응답을 받지 못하면 다시 요청을 보내거나 재시도 횟수를 넘어서면 요청이 실패
• retry.backoff.ms
  • 실패한 요청에 대해 프로듀서가 재시도하기 전에 대기할 시간
• producer.type
  • 메시지를 동기(sync), 비동기(async)로 보낼지 선택할 수 있음
  • 비동기를 사용하는 경우 메시지를 일정 시간동안 쌓은 후 전송하므로 처리 효율을 향상시킬 수 있음

 

Spring Boot에서 Kafka Subscriber 설정 예

Kafka Consumer 환경 구성 예제

 

Kafka Consumer Configuration

• group.id
  • 컨슈머 그룹을 식별하는 고유 아이디
  • 메시지를 전송할 때 지정하는 topic 이름과 다름
  • Zookeeper에서는 각 그룹의 메시지 offset을 관리하는데 그룹 id가 같으면 offset값 또한 공유됨
• bootstrap.servers
  • 프로듀서와 동일. 즉, 연결할 정보를 의미
• fetch.min.bytes
  • 한번에 가져올 수 있는 최소한의 데이터 크기
  • 기본 값인 1의 경우 즉시 가져오는 것을 뜻하며 1보다 크거나 데이터가 설정한 값보다 작은 경우에는 요청을 처리하지 않고 대기
• auto.offset.reset
  • 카프카의 초기 offset이 없거나 데이터가 삭제하여 현재 존재하지 않는 경우에 다음의 설정을 따름
    • earliest : 가장 빠른 오프셋으로 자동 재설정합니다.
    • latest : 최신 오프셋으로 자동 재설정합니다.(기본 값)
    • none : 이전 오프셋이 발견되지 않으면 컨슈머 그룹에 예외를 던집니다.
    • anything else : 컨슈머에게 예외을 던집니다.
• session.timeout.ms
  • 컨슈머의 실패를 감지하는데 쓰이는 타임아웃 
  • 브로커와의 세션 타임아웃 시간인데, 타임아웃이 발생하면 컨슈머는 종료되거나 장애로 인식되고 브로커는 해당 컨슈머를 그룹에서 제외하고 리밸런싱을 시도
  • 참고로 컨슈머는 브로커에게 자기가 살아있음을 나타내기 위해 신호(heartbeat)를 전송
• heartbeat.interval.ms
  • 살아있음을 알리는 신호(heartbeat)의 예상 시간을 설정할 수 있음
  • 새로운 컨슈머가 그룹에 속해지거나 제외될 때 재조정을 위해 유용하게 사용될 수 있음
  • session.timeout.ms 값보다 낮게 설정해야 함
• max.poll.interval.ms
  • 살아있다고 신호를 보내지만 실제로 메시지를 소비하지 않는 경우, poll이 호출되지 않으면 컨슈머는 실패된 것으로 간주
  • 이때, 컨슈머 그룹은 다른 구성원에게 파티션을 재할당하기 위해 재조정됨

 

아래의 2편에서 계속>> 

Spring Boot에서 Apache Kafka 사용 ... 2/2