Kafka의 기본 구성

Kafka의 기본 구성: Topic, Consumer, Producer

• Producer: 카프카에 메시지(데이터)를 전달하는 주체
• Consumer: 카프카의 메시지(데이터)를 처리하는 주체
• Topic: 카프카에 넣을 메시지의 종류를 구분하는 개념

카프카의 구조: Producer  →  Broker  →  Consumer

프로듀서는 Kafka로 메시지(데이터)를 전달한다.

그러면 Kafka는 메시지 큐에 토픽 별로 구분해 전달받은 메시지를 저장해둔다.

컨슈머는 Kafka에 새로운 메시지가 생겼는 지 주기적으로 체크하다가, 새로운 메시지가 있다는 걸 발견하면 그 메시지를 조회해와서 처리한다.

---

CLI를 활용한 Kafka 조작

백엔드 서버(Spring Boot)로만 Kafka를 조작할 수 있는 것이 아니라, CLI로도 Kafka의 모든 기능을 조작할 수 있다. 

CLI로 Kafka를 조작하는 방법을 알아보자.

토픽 생성하기 / 조회하기 / 삭제하기

# kafka 디렉터리 안에서 아래 명령어를 실행시켜야 함
$ cd kafka_2.13-4.0.0

# 토픽 생성
# bin/kafka-topics.sh --bootstrap-server <kakfa 주소> --create --topic <토픽명> 
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic email.send 

# 토픽 전체 조회
# bin/kafka-topics.sh --bootstrap-server <kakfa 주소> --list
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

# 특정 토픽 세부 정보 조회
# bin/kafka-topics.sh --bootstrap-server <kakfa 주소> --describe --topic <토픽명> 
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic email.send

# 토픽 삭제
# bin/kafka-topics.sh --bootstrap-server <kafka 주소> --delete --topic <토픽명>
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic email.send

 

Kafka에 메시지 넣기 / Kafka에서 메시지 조회하기

Kafka에 메시지 넣기

Kafka에 넣는 메시지는 Key-Value 형태로 넣을 수도 있고, Key는 생략한 채로 Value만 넣을 수도 있다.

# email.send라는 토픽에 메시지 넣기
$ bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic email.send

# 위 명령어 입력 후 넣을 메시지 내용 입력하고 Enter 누르기
# 입력 다 했으면 Ctlr + c로 입력 상태 종료
hello1
hello2
hello3

Kafka에서 메시지 조회하기

전통적인 메시지 큐(RabbitMQ, SQS)는 메시지를 꺼내서 읽어들이면 해당 메시지를 큐에서 제거하는 구조이다.

반면에 Kafka는 메시지를 읽고 제거하는 방식이 아니라, 저장되어 있는 메시지를 읽기만 하고 제거하지 않는 방식으로 작동한다.

이런 구조 덕분에 Kafka는 같은 메시지를 여러 번 읽는게 가능하다.

# email.send라는 토픽에 있는 메시지 꺼내기 (실시간으로 계속 조회)
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic email.send --from-beginning

 

실제 서비스에서는 메시지를 단순히 한 번만 읽고 끝나는 게 아니라, 어디까지 읽었는지를 기억하고 그 다음 메시지부터 계속해서 처리하는 방식이 필요하다.

 

메시지를 어디까지 읽었는 지 기억하고, 그 다음 메시지부터 처리하기 (Consumer Group, Offset)

Consumer Group이 어디까지 메시지를 읽었는지를 offset이라는 번호로 기억해둔다.

이 덕분에 Consumer Group에 속해있는 Consumer들은 안 읽은 메시지부터 순차적으로 메시지를 읽게 된다.

• Consumer Group : 1개 이상의 컨슈머를 하나의 그룹으로 묶은 단위
• offset : 메시지의 순서를 나타내는 고유 번호 (0부터 시작). 파티션 내에서의 메시지 위치를 의미한다.

Kafka는 메시지를 "브로커가 푸시하는 방식"이 아니라 "컨슈머가 당겨가는(pull) 방식"으로 설계되었기 때문에, 읽은 위치는 컨슈머 그룹이 관리한다.

오프셋은 컨슈머 '그룹' 단위로 관리하는데, 같은 토픽을 여러 애플리케이션이 독립적으로 소비할 수 있어야 하기 때문이다.
(ex. consumer A 그룹 서버들은 데이터 분석용으로 데이터를 읽고, consumer B 그룹 서버들은 알림 발송용으로 처리)

컨슈머 그룹은 어디까지 메시지를 읽었는 지에 대한 정보(CURRENT-OFFSET)를 알고 있다. CURRENT-OFFSET : 다음에 읽을 메시지의 오프셋 번호

# 컨슈머 그룹을 활용해 메시지 조회하기
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic email.send \
	--from-beginning \ # 오프셋 기록이 없으면 처음부터 읽고, 있으면 오프셋부터
    --group email-send-group # 없으면 생성 & 이 컨슈머 그룹으로 메시지를 읽는다

# 컨슈머 그룹 전체 조회하기
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# 컨슈머 그룹 세부 정보 조회하기
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
	--group email-send-group --describe

 

실제 서비스에서 똑같은 요청을 중복해서 여러번 처리하면 안 된다. 그래서 반드시 컨슈머 그룹을 활용해서 메시지를 읽어야 한다. 그래야 컨슈머 그룹이 메시지를 어디까지 읽었는 지 메시지의 오프셋 값으로 기억해뒀다가, 아직 처리하지 않은 그 다음 메시지부터 처리할 수 있게 된다.

---

Spring Boot로 Kafka 조작

Producer 서버 구현

Spring Boot에 Kafka 연결

application.yml에 Kafka 연결을 위한 정보 작성.
Kafka는 byte[] 형태로 데이터를 전송하기 때문에 직렬화 과정이 필요.

spring:
  kafka:
    # Kafka 서버 주소 (EC2에 카프카를 설치했기 때문에 EC2 주소를 입력해야 한다.)
    bootstrap-servers: 52.79.47.232:9092
    producer:
      # 메시지의 key 직렬화 방식 : String을 byte[]로 바꿔 전송하는 역할 (Kafka는 byte[] 형태로 데이터를 전송하기 때문에 직렬화 과정이 필요하다.)
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 메시지의 value 직렬화 방식 : String을 byte[]로 바꿔 전송하는 역할
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

Spring Boot로 Kafka에 메시지 넣기

메시지 객체를 만들고, kafka 보낸다.

@Service
@RequiredArgsConstructor
public class EmailService {

	// Spring Kafka에서 제공하는 Producer 전용 클래스. Spring에서 Kafka Producer를 쉽게 사용하게 해주는 고수준 API
	private final KafkaTemplate<String, String> kafkaTemplate;

	public void sendEmail(SendEmailRequestDto request) {
		EmailSendMessage emailSendMessage = EmailSendMessage.from(request);

		kafkaTemplate.send("email.send", toJsonString(emailSendMessage));
	}

	private String toJsonString(Object object) {
		return new ObjectMapper().writeValueAsString(object);
	}
}

public record EmailSendMessage(
	String from,
	String to,
	String subject,
	String body
) {
	public static EmailSendMessage from(SendEmailRequestDto request) {
		return new EmailSendMessage(
			request.from(),
			request.to(),
			request.subject(),
			request.body()
		);
	}
}

 

확인을 위해 postman으로 요청을 보내고 터미널로 확인해보자

 

Consumer 서버 구현

Spring Boot에 Kafka 연결

Broker: Kafka에서 실제로 메시지를 저장하고 관리하는 서버(노드)

server:
  port: 0 # 사용 가능한 랜덤 포트를 찾아서 서버를 실행 (Producer 서버와의 포트 충돌을 방지)

spring:
  kafka:
    # Kafka 브로커 주소 (클라이언트가 최초로 연결할 브로커 목록)
    # 실제 운영에서는 여러 브로커를 콤마로 구분하여 설정 가능
    bootstrap-servers: 52.79.47.232:9092
    consumer:
      # Kafka 브로커로부터 수신한 key(byte[])를 String으로 변환하는 역직렬화 클래스
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # Kafka 프로토콜/저장 형식은 기본적으로 key/value를 byte[]로 취급
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

      # 컨슈머 그룹이 없으면, 직접 생성해서 메시지를 처음부터 읽음.
      # 있으면, 해당 컨슈머 그룹이 읽었던 메시지부터 읽음. (like --from-beginning)
      auto-offset-reset: earliest
      # 이 옵션을 주지 않으면 컨슈머 그룹을 직접 생성해서 메시지를 읽을 때,
      # 기존에 쌓여있던 메시지를 읽지 않고 컨슈머 그룹이 생성된 이후에 들어온 메시지부터 읽어버린다.
      # 그럼 컨슈머 그룹이 생성되기 전에 쌓여있던 메시지들이 처리되지 않고 누락돼버린다.

Spring Boot로 Kafka에 메시지 조회하기

@KafkaListener를 사용하면, 메시지가 새로 들어왔는지 계속 체크한다.(polling 방식)

컨슈머 그룹으로 메시지를 읽기 때문에 중복 처리하지 않는다.

 

@KafkaListener는 Bean에서만 동작하므로 Spring bean으로 등록하기 위해 @Service 애노테이션은 붙인다.
(@Component를 붙여도 되지만, 비즈니스 로직을 담당하는 클래스라는 의미 표현으로 @Service를 붙였다.)

 

Spring Kafka는 애플리케이션 시작 시,

1. Spring Context 안에 등록된 Bean 들을 스캔
2. @KafkaListener 가 붙은 메서드를 찾음
3. 해당 메서드를 Kafka 리스너 컨테이너에 연결

@Service
public class EmailSendConsumer {

	@KafkaListener(topics = "email.send", groupId = "email-send-group")
	public void consume(String message) {
		System.out.println("Kafka로부터 받아온 메시지 = " + message);

		EmailSendMessage emailSendMessage = EmailSendMessage.fromJson(message);

		// ... 실제 이메일 발송 로직은 생략 ...
		System.out.println("이메일 발송 완료");
	}
}

public record EmailSendMessage(
	String from,
	String to,
	String subject,
	String body
) {
	public static EmailSendMessage fromJson(String json) {
		return new ObjectMapper().readValue(json, EmailSendMessage.class);
	}
}

---

비동기 성능 이점 확인과 한계

메시지 큐를 활용한 통신 방식은 비동기적으로 작업을 처리하기 때문에 모든 작업이 다 처리되는 것과 상관없이 빠르게 응답을 받을 수 있다고 했다. 실제로 그런지 확인하기 위해, 이메일 발송 로직에 3초 sleep하도록 코드를 추가해보자.

try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        throw new RuntimeException("이메일 발송 실패");
    }

이메일 발송에 3초가 걸리더라도, 메시지 큐를 사용하면 API는 Kafka에 메시지를 넣고 25ms 내외로 바로 응답할 수 있다. 덕분에 사용자 입장에선 처리 시간이 느리다는 걸 전혀 체감하지 못한다.

 

하지만 비동기로 처리하면 Consumer가 제대로 작업을 처리했는 지 어떻게 확신할 수 있을까?

 

REST API 방식을 활용해 동기적으로 처리할 때는 이메일 발송 처리 작업이 끝날 때까지 기다렸다가 응답을 한다. 그러다보니 비교적 응답 속도가 느릴 수 밖에 없지만, 이메일 발송의 성공 여부를 확인하고 그에 맞게 응답을 할 수 있다. 

 

하지만 메시지 큐를 활용한 비동기 처리는 빠르지만 한 가지 중요한 한계를 가지고 있다.

바로 사용자에게 작업의 실제 성공 여부를 확인하지 않고 응답을 먼저 보내버린다는 점이다

 

이러한 비동기 구조의 단점을 보완하기 위해 시스템에서는 다양한 보완 전략을 사용한다. 

대표적으로는 메시지 처리 중 실패가 발생했을 때 자동으로 재시도(retry)하는 방식, 

여러 번의 재시도 끝에도 실패한 메시지를 별도로 보관하는 Dead Letter Topic(DLT)을 활용하는 방식을 주로 활용한다.

 

다음 글에서 실패한 메시지를 재시도(retry) 하는 방법과 Dead Letter Topic(DLT)를 활용하는 방법을 다뤄보겠다. 

---

참고 자료 & 이미지 출처
실전에서 바로 써먹는 kafka 입문