✅ kafka
- 대규모 데이터를 처리할 수 있는
메시지 큐= (임시저장소) 메시지 큐의 종류중 하나이다
메시지 큐 : 큐 형태에 데이터를 일시적으로 저장하는 임시 저장소
매시지 큐 를 활용하면 비동기적으로 데이터를 처리할 수 있어서 효율적이다
✅ 그림으로 보는 메시지 큐 를 활용한 통신
위 그림에서 순차적으로 보면
-
사용자가 요청을 보내면
Producer라는 서버에 보낸다 -
Producer서버는 처리해야할 요청의 정보들을 메시지로 만들어메시지 큐에 전달한다 -
Producer서버는메시지 큐에 메시지를 넣자마자 사용자에게 성공응답을 한다 (비동기방식)
💡 비동기 방식으로 처리하기 때문에 효율적이며, 대규모 트래픽을 처리할 때도 유리한 구조이다 -
메시지 큐는Producer로부터 받은 메시지(요청)을 보관하고 있으며
처리해야할 메시지들을 보관하는 임시 저장소 역할을 한다.
메시지를 전달받으면 kafka는메시지 큐에토픽별로 구분해 메시지를 저장해둔다
💡 토픽이란? :메시지 큐에 넣을 메시지의 종류를 구분하는 개념 -
Consumer서버에서는메시지 큐가 보관하고 있는 메시지(요청)을 꺼내서 실제 작업을 수행한다
즉 실제로 비즈니스 로직을 수행하는 서버를Consumer라고 한다
✅ 토픽(TOPIC) 관련 명령어
💡 토픽 관련 작업은 모두 bin/kafka-topics.sh 스크립트로 수행한다
(1) 토픽 생성 : bin/kafka-topics.sh --bootstrap-server <kakfa 주소> --create --topic <토픽명>
ex) bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic email.send
(2) 토픽 조회 : bin/kafka-topics.sh --bootstrap-server <kakfa 주소> --list
ex) bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
(3) 특정 토픽 조회 : bin/kafka-topics.sh --bootstrap-server <kakfa 주소> --describe --topic <토픽명>
ex) bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic email.send
(4) 토픽 삭제 : bin/kafka-topics.sh --bootstrap-server <kafka 주소> --delete --topic <토픽명>
ex) bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic email.send
✅ 프로듀서(producer) 관련 명령어
💡 프로듀서는 메시지큐에 메시지를 보내는 역할을 한다
프로듀서 관련 작업은 모두 bin/kafka-console-producer.sh 스크립트로 수행한다
우선 email.send 라는 토픽을 만들었다고 가정
(1) Kafka의 특정 토픽에 메시지 넣기
bin/kafka-console-producer.sh --bootstrap-server <kafka 주소> --topic <토픽명>
ex) bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic email.send
위 명령어를 입력 후 넣을 메시지를 입력하고 엔터누르면 토픽에 메시지가 잘 들어감
(2) Kafaka에서 메시지 조회하기
ex) bin/kafka-console-consumer.sh --bootstrap-server localhost:9092
--topic email.send --from-beginning
💡 --from-beginning : 토픽에 저장된 가장 처음 메시지부터 출력해온다
위 명령어는 email.send 토픽에 저장된 메시지를 전부 조회한다. 추가로 email.send 토픽에 메시지가 추가되어도 메시지가 실시간으로 조회하는 상태로 변경된다
전통적인 메시지 큐(RabbitMQ, SQS) 는 메시지를 조회하면, 해당 메시지를 큐에서 제거하는 구조이지만
Kafka는 메시지를 조회하면, 읽기만 하고 제거가 되지 않는다
즉 Kafka는 같은 메시지를 여러번 읽는게 가능하다
(3) 메시지를 어디까지 읽었는지 기억하고, 다음 메시지부터 처리하기
Kafka에서는 컨슈머 그룹 을 활용하여 어디까지 메시지를 읽었는지 오프셋 이라는 번호로 기록해둔다
컨슈머 그룹 에 속해있는 컨슈머 들은 안 읽은 메시지로부터 순차적으로 메시지를 읽게 된다
컨슈머 : Kafka 의 메시지를 처리하는 주체
컨슈머 그룹 : 1개 이상의 컨슈머를 하나의 그룹으로 묶은 단위
오프셋 : 메시지의 순서를 나타내는 고유 번호 (0부터 시작) = 배열 인덱스와 비슷
컨슈머 그룹을 지정해서 메시지 읽기
ex) bin/kafka-console-consumer.sh --bootstrap-server localhost:9092
--topic email.send --from-beginning --group email-send-group
--group email-send-group → 컨슈머 그룹 지정
위 명령어처럼 --group email-send-group 와 --from-beginning 을 같이 사용하면
컨슈머 그룹 의 오프셋 기록이 없으면 첫 메시지부터 다 읽고, 기록이 있으면 그 이후 오프셋부터 읽는다
컨슈머 그룹이 잘 생성됬는지 조회
ex) bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
특정 컨슈머 그룹 세부정보 조회하기
ex) bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092
--group email-send-group --describe
세부정보를 조회하면 현재 컨슈머 그룹의 오프셋을 확인할 수 있다
만약 email.send 토픽에 메시지를 추가로 넣고, 컨슈머 그룹 으로 메시지를 조회하면
기존에 읽었던 메시지는 제외하고 방금전에 추가한 메시지만 조회하는것을 알 수 있다
실무에서는 똑같은 요청을 중복해서 여러번 호출하면 안된다. 그래서 반드시 컨슈머 그룹 을 활용하여
메시지를 어디까지 읽었는지 오프셋 값으로 기억해뒀다가, 처리하지 않은 그 다음 메시지부터 처리해야 한다.
위에서 Producer 서버는 메시지 큐 에 메시지를 넣자마자 사용자에게 성공응답을 한다고 했다
이 방식은 비동기방식인데, 비동기방식으로 처리하면 제대로 작업을 처리했는지 알 수 있을까?
즉 Kafka에 메시지만 넣고, 바로 응답할 수 있기 때문에 사용자 입장에서는 기다림 없이 빠르게 처리가 된 것 처럼 느껴진다. 하지만 이 구조는 사용자에게 실제 성공여부를 확인하지 않고 응답을 먼저 보내고 있다.
이러한 비동기 구조의 단점을 보완하기 위해서, 두가지 방법을 주로 많이 활용한다
- 메시지 처리 중 실패가 발생했을 때, 자동으로
재시도 (retry)하는 방식 - 여러번의 재시도 끝에도 실패한 메시지를 별도로 보관하는
Dead Letter Topic (DLT)방식
✅ 재시도 방식
재시도 전략 방식은 별도의 설정을 하지 않아도 이미 기본값으로 세팅이 되어있다
interval : 재시도를 하는 시간 간격으로 (기본값 0) = 즉시 재시도한다는 뜻
maxAttempts : 최대 재시도 횟수로 (기본값 9)
currentAttemps : 지금까지 시도한 횟수로 (기본값 : 최초 시도 횟수 + 최대 재시도 횟수)
위와 같은 재시도 전략의 기본값을 바꾸려면 어떻게 해야할까?
@RetryableTopic 어노테이션을 사용
@RetryableTopic(
attempts = "5", // attemps는 총 시도횟수로 (최초 시도횟수를 제외한 4번까지 재시도를 한다는 뜻)
backoff = @Backoff(delay = 1000, multiplier = 2) // 재시도 간격으로 처음에는 1000ms로 재시도하다가
// 재시도 할때마다 시간이 2배로 증가함 (1000ms -> 2000ms -> 4000ms -> 8000ms 순으로 증가한다.)
)
public void consume(String message) {
...
}💡 현업에서는 보통 재시도횟수를 3~5 회 사이로 정한다. 재시도를 너무 많이 하면 시스템 부하가 커지고
너무 적으면 일시적인 네트워크 장애에 대응하기 어렵기 때문이다
하지만 이렇게 여러번 재시도 를 했는데도 실패했을 경우 어떻게 해야할까?
사용자 입장에서는 성공 메시지가 보여졌지만, 실제로는 실패한 작업으로 남게된다.
이럴 경우 Dead Letter Topic (DLT) 라는 별도 토픽을 활용해서, 재시도 까지 실패한 메시지를
안전하게 보관하여 나중에 관리자가 확인하여 수동적이라도 처리할 수 있도록 구성할 수 있어야한다
✅ Dead Letter Topic
- 오류로 인해 처리할 수 없는 메시지를 임시로 저장하는 토픽
- kafka에서는 재시도까지 실패한 메시지를
다른 토픽에 저장해서 유실을 방지하고 후속 조치를 가능하게 한다
DLT 를 사용하는 이유?
- 실패한 메시지가 유실되는것을 방지
- 사후에 실패 원인을 분석할 수 있음
- 처리되지 못한 메시지를 수동으로 처리할 수 있음
✅ DLT 를 활용해 재시도에 실패한 메시지 따로 보관하기
Spring Kafka는 위에서 선언한 @RetryableTopic 을 사용하면 자동으로 DLT 토픽을 생성하고
메시지를 전송해준다. 기본적으로 (기존토픽명) - dlt 형태로 DLT 토픽이 만들어진다
하지만 아래 코드와 같이 내가 원하는 형태로 DLT 토픽의 명칭을 바꿀 수 있다 -> .dlt 로 변경
@RetryableTopic(
attempts = "5",
backoff = @Backoff(delay = 1000, multiplier = 2)
dltTopicSuffix = ".dlt" // DLT 토픽 이름에 붙일 접미사
)
public void consume(String message) {
...
}이후 CLI 에서 토픽을 조회하면
DLT토픽이 생성된 것을 확인할 수 있다
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
메시지 재시도를 하면서 생긴 토픽들도 보인다. 만들어진 .dlt 토픽을 확인해보자
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic email.send.dlt --from-beginning을 입력하면 재시도에 실패한 메시지가 .dlt 토픽에 저장되어 있음을 확인 가능
전체적인 그림
이렇게 재시도 조차 실패한 메시지를 DLT 토픽에 저장한 후, 이 실패한 메시지를 조치하는 방법에 대해서 알아보자
DLT에 저장된 실패 메시지를 로그 시스템에 전송하여 장애 원인을 추적해야한다DLT에 저장되자마자 관리자에게 알림을 갈 수 있도록 설정한다.- 알림을 받은 관리자는 로그에 쌓인 내용을 보고 장애 원인을 분석하고,
수동으로 처리한다.
여기서 수동 으로 처리한다는 방식은
일시적인 네트워크 장애였을 경우 메시지를 원래 토픽으로 다시 보내는 방법
메시지의 내용이 잘못됬을 경우 그 메시지를 폐기하는 방법
잘못된 메시지가 Kafka에 들어가지 않도록, Producer 의 검증 로직을 보완하는 방법이 있다
@KafkaListener(
topics = "email.send.dlt",
groupId = "email-send-dlt-group"
)
public void consume(String message) {
System.out.println("로그 시스템에 전송 : " + message); // 대충 로그 시스템에 전송하는 로직 ..
System.out.println("Slack에 알림 발송"); // 대충 Slack에 알림 발송하는 로직..
}위 코드처럼, DLT 토픽으로 빠진 메시지는 관리자에게 알림이 발송되어 관리자가 수동으로 처리할 수 있다
