대규모 스트림 처리

 

 

🗨️ 오리엔테이션

오프닝

대규모 시스템이란 무엇일까요?

• 인터넷 환경에서는 수백만 명의 사용자가 동시에 접속하고 상호작용할 수 있는 시스템을 구축해야 하는 상황이 자주 발생합니다.
• 이러한 시스템은 단순히 많은 사용자를 처리하는 것뿐만 아니라, 안정성과 신뢰성을 유지하면서 고성능을 제공해야 합니다.
• 대규모 시스템을 설계하고 구축하는 과정에서 고려해야 할 중요한 요소들을 알아볼 것입니다. 이를 통해 대규모 스트림 처리에 대한 이해뿐만 아니라, 대규모 시스템의 전반적인 구성과 운영에 대한 깊은 통찰을 얻게 될 것입니다.

그중 큐를 통한 대규모 스트림 처리에 집중해 보겠습니다.

• 큐를 사용하면 대규모 처리를 할 수 있다는 이야기를 많이 들어봤을 것입니다.
• 모든 상황에서 큐를 사용해야 할까요? 큐를 사용해야 하는 상황과 사용하지 않아도 되는 상황이 있을 것입니다.
• 우리는 수업을 통해 어떠한 상황에서 큐를 사용하고 또, 큐를 효과적으로 관리하는 방법을 배울 것입니다. 또한 큐 라우팅을 통해 서비스를 설계하고 개발해 볼 것입니다.

개발한 애플리케이션의 성능 테스트는 필수입니다.

• 애플리케이션을 개발한 후에는 반드시 테스트를 거쳐야 합니다.
• 기능에 대한 테스트를 통해 애플리케이션이 정상적으로 동작하는 것을 확인하는 것은 당연한 것입니다.
• 우리는 성능 테스트를 통해 우리의 애플리케이션의 성능을 측정하여 애플리케이션이 실제 환경에서 어떻게 동작하는지 미리 검증하고, 신뢰성 있는 시스템을 구축할 수 있게 될 것입니다.

대규모 시스템

📌 메시징 시스템을 개발하는 것을 예로 들어보겠습니다. 어떠한 기준으로 시스템을 설계해야 할까요?

동시 접속자와 초당 요청량(TPS)

사용자수

• 대규모 시스템을 설계할 때 가장 중요한 요소 중 하나는 사용자 수입니다.
  • 얼마나 많은 사용자가 시스템을 사용할 것인지 파악하는 것이 중요합니다. 기존 시스템에 새로운 기능을 추가하는 경우, 시스템 모니터링을 통해 하루에 몇 명의 사용자가 접속하는지 알 수 있습니다. 그러나 단순히 하루 접속량을 파악하는 것만으로는 충분하지 않습니다. 더 중요한 것은 동시 접속자의 요청 수입니다.

TPS (TPS, Transactions Per Second)

• TPS란?
  • TPS (Transactions Per Second)는 초당 처리되는 트랜잭션의 수를 나타내는 지표입니다. 이는 시스템의 성능을 평가하는 중요한 지표 중 하나로, 특히 대규모 시스템에서 중요한 역할을 합니다. TPS는 시스템이 얼마나 많은 요청을 동시에 처리할 수 있는지를 나타내며, 시스템의 처리 능력을 가늠할 수 있게 해 줍니다.
• 시스템이 초당 요청량(TPS, Transactions Per Second)을 견딜 수 있어야 합니다.
  • 이를 위해서는 일간 접속량이 아닌, 특정 시간대에 초당 접속자 요청량이 가장 많은 시간을 파악해야 합니다. 이 정보는 시스템의 용량 계획을 세우는 데 매우 중요합니다.
  • 예를 들어, 기존 시스템이 오전 9시 30분에 초당 200건의 요청을 처리했다고 가정해 보겠습니다. 이러한 데이터를 기반으로 시스템의 용량을 결정해야 합니다. 우리의 목표는 시스템이 예상치 못한 최대 부하를 견딜 수 있도록 하는 것입니다. 따라서 200건을 기준으로 1.5배인 300건을 처리할 수 있도록 시스템을 설계하는 것이 바람직합니다. 이는 시스템이 예기치 않은 트래픽 급증에도 견딜 수 있는 여유를 제공할 것입니다.
• 예상치 못한 이벤트로 인해 설계 예상 이상의 요청이 몰린다면 시스템이 중단될 수 있습니다.
  • 이러한 상황을 대비하기 위해서는 다양한 방법을 고려해야 합니다. 이러한 방법들을 통해 시스템의 안정성을 높일 수 있습니다.
    1. 첫째, 애플리케이션의 수를 늘리는 방법이 있습니다.
    2. 둘째, 오류 상황에서 사용자가 대기할 수 있도록 대기열을 설정하는 것도 하나의 방법입니다.
    3. 셋째, 자동 스케일링을 통해 시스템의 자원을 동적으로 할당하여 부하를 분산시키는 것도 중요합니다.

 

요청 종류에 따른 개발

• 📌 시스템이 읽기 전용인지, 쓰기 및 업데이트를 위한 것인지도 중요합니다. 이를 파악하고 처리 속도를 빠르게 하여 응답한다면 보다 많은 사용자를 수용할 수 있습니다.
  • 데이터 제공 및 저장에서 가장 많은 시간을 소모하는 부분은 대부분 DB에서 데이터를 조회하거나 쓰는 것입니다.
  • 요청 종류에 따라 이 부분의 허들을 최소화하는 것이 중요합니다. 

읽기 요청 최적화

• 캐시 사용
  • 모든 사용자가 같은 데이터를 요청하는 경우, 이를 해소하기 위해 DB의 데이터를 레디스 같은 캐시에 올려두면 요청에 대한 응답 속도를 빠르게 할 수 있습니다.
  • 캐시를 사용하면 데이터를 빠르게 제공할 수 있으며, DB의 부하를 줄일 수 있습니다.
  • 개인화 데이터(사용자마다 결과가 달라지는 데이터)가 아니라면 캐시를 사용하여 데이터 로드에 시간 소요를 최소화하는 방법이 중요합니다.
  • 데이터에서 필터 된 데이터를 가져오는것보다 레디스에서 필터된 데이터를 가져오거나 애플리케이션 로직에서 필터링을 수행하는 것이 더 큰 성능상의 이점을 가질 것입니다.
  • 엣지 단( 사용자와 가장 가까운 지점)에서 캐싱(데이터를 미리 임시 보관해서 바로 제공)을 처리하면 시스템의 요청 처리량이 줄어들어 적은 수의 애플리케이션으로도 요청을 처리할 수 있습니다. 엣지 캐싱은 사용자와 가까운 곳에서 데이터를 제공하므로 네트워크 지연을 최소화하고, 사용자 경험을 향상시킬 수 있습니다.
  • 캐싱 전략을 설계할 때는 캐시 갱신 정책도 중요합니다. 예를 들어, 공지사항이 자주 변경되지 않는 경우에는 캐시의 유효 기간을 길게 설정할 수 있습니다. 반면, 변경이 잦은 경우에는 짧은 유효 기간을 설정하거나, 변경 시점에 캐시를 무효화하여 최신 데이터를 제공할 수 있도록 해야 합니다. 이러한 캐싱 전략을 통해 시스템의 효율성을 극대화할 수 있습니다.
  • 데이터 소실의 위험을 줄이기 위해 데이터의 유효성을 지속적으로 검증하고, 데이터가 손실되지 않았는지 확인해야 합니다. 또한, 캐시 계층에서 데이터 소실 시 재요청을 통해 데이터베이스에서 데이터를 다시 가져올 수 있도록 해야 합니다.

🔥 Redis 캐시 도입 시 반드시 주의해야 할 실무 포인트

• 메모리(RAM) 기반으로 엄청난 속도를 자랑하는 Redis는 조회수, 세션, 장바구니 등 빠른 업데이트와 휘발성 데이터를 다루는 데 최적화된 데이터베이스입니다. 하지만 대규모 트래픽 환경(예: 시간당 10만 명 이상)에서는 무작정 도입했다가 큰 코를 다칠 수 있습니다.

🚨 1. 무심코 쓴 캐시가 서버를 죽인다: 용량 초과(OOM) 위험

• Redis의 최대 용량(예: 50GB)이 다 차버리면 시스템 전체가 다운되는 대참사가 발생합니다.
  • 원인: Spring에서 제공하는 캐시 어노테이션을 무심코 사용하면, 객체 전체가 직렬화(문자열 형태)되어 저장되면서 불필요한 정보까지 막대한 공간을 차지하게 됩니다. (예: 10KB짜리 페이징 데이터도 사용자와 페이지 수가 늘어나면 순식간에 50GB를 초과함)
  • 해결책: 객체를 통째로 넣지 말고, 꼭 필요한 데이터만 선별적으로 추출하여 가볍게 캐싱해야 합니다.

📊 2. 골든타임을 확보하라: 촘촘한 모니터링과 알람 설정

• 대규모 시스템에서 모니터링은 선택이 아닌 필수입니다. 장애가 터진 후 수습하는 것이 아니라, 미리 예측하고 방어해야 합니다.
  • 사전 대응: Redis 최대 용량이 50GB라면, 30GB 도달 시점에 알람이 오도록 설정하세요.
  • 조치 방법: 알람을 확인한 즉시 트래픽 증가 원인을 파악하여 용량을 100GB로 스케일업(확장)하거나, 불필요한 데이터를 정리할 수 있는 '골든타임'을 확보할 수 있습니다.

🛡️ 3. 레디스가 죽어도 시스템은 살려라: 다중 방어선 구축

• Redis에 장애가 생겼다고 해서 전체 서비스가 멈추거나 DB까지 뻗게 두어서는 안 됩니다.
  • 커넥션 타임아웃 설정 (필수): 타임아웃을 3~4초로 짧게 설정하세요. 이 설정이 없다면 Redis 장애 시 모든 트래픽이 메인 DB로 직접 쏟아져 내려 DB마저 연쇄적으로 다운될 수 있습니다.
  • 유연한 에러 핸들링: 레디스 장애 시에는 빈 데이터 또는 노데이터 응답을 주어 사용자에게 에러로 인식되지 않게 처리하는 방법이 있습니다.
  • 프론트엔드 더미 데이터: 프론트엔드 단에서 미리 준비된 더미(Dummy) 데이터를 보여주도록 2차 방어선을 구축하면, 사용자는 장애를 눈치채지 못하고 운영자는 긴장감 없이 복구에 집중할 수 있습니다.

🧠 4. 중복 저장은 죄악이다: 데이터 통합 처리와 필터링

• 사용자 A, B, C가 각자 필요한 데이터가 있을 때, 사용자별로 데이터를 따로 캐싱하면 데이터가 기하급수적으로 쌓여 금방 포화 상태(천 개, 만 개 이상)가 됩니다.
  • 문제점: 사용자별 맞춤 데이터를 각각 저장하면 동일한 데이터가 계속 중복 적재되어 메모리를 낭비합니다.
  • 해결책 (애플리케이션 레벨 필터링):
    1. Redis에는 여러 사용자가 공통으로 쓸 수 있는 '가벼운 전체 데이터 목록(Common Data Set)' 딱 하나만 저장합니다.
    2. 실제 요청이 들어오면, 애플리케이션 함수 내부에서 "이 데이터가 A에게 필요한가?"를 판단하여 필요한 데이터만 필터링(가공)하여 반환합니다.
  • 효과: 저장 비용과 메모리 사용량을 극적으로 줄이면서도, 유연하게 사용자 맞춤형 데이터를 제공할 수 있습니다.

💡 요약하자면: 대용량 환경에서의 Redis는 "꼭 필요한 공통 데이터만 가볍게 저장하고(최적화), 철저히 감시하며(모니터링), 만약의 사태에 대비한 플랜 B(타임아웃 및 더미 응답)를 세워두는 것"이 핵심입니다.

 

🗨️ 데이터베이스 최적화

데이터베이스 인덱싱

• 데이터베이스 인덱싱은 조회 성능을 크게 향상시킬 수 있는 방법입니다. 인덱스를 사용하면 데이터베이스는 데이터를 빠르게 검색할 수 있습니다. 올바르게 설계된 인덱스는 읽기 성능을 최적화하고 쿼리 응답 시간을 줄입니다. 주의할 점은 인덱스가 너무 많으면 쓰기 성능이 저하될 수 있다는 것입니다.
  • 데이터베이스 인덱싱: 검색 속도를 높이기 위해 특정 컬럼에 인덱스를 만들어두는 작업
  • 인덱스: 책의 목차처럼 원하는 데이터를 빠르게 찾기 위한 별도의 검색 구조

데이터베이스 샤딩

• 데이터베이스 샤딩은 데이터베이스를 여러 샤드(분산 저장된 데이터 조각)로 분할하여 각각의 샤드가 독립적으로 쿼리를 처리하도록 하는 방법입니다. 이를 통해 단일 데이터베이스에 대한 부하를 분산시킬 수 있으며, 읽기 요청에 대한 응답 속도를 향상시킬 수 있습니다. 예를 들어, 사용자 데이터를 여러 샤드에 분산 저장하여 읽기 성능을 최적화할 수 있습니다.
  • 예를 들어 테이블 파티셔닝을 생각해볼 수 있습니다. 이는 기본적인 형태의 샤딩으로, 데이터베이스 수준에서 데이터를 분할합니다.

-- 주문 테이블 생성
CREATE TABLE orders (
  order_id SERIAL PRIMARY KEY, -- 주문 ID (자동 증가 기본 키)
  customer_id INT,             -- 주문한 고객 ID
  order_date DATE,             -- 주문 날짜
  amount DECIMAL               -- 주문 금액
) PARTITION BY RANGE (order_date);
-- order_date(주문 날짜)를 기준으로 범위(RANGE) 파티셔닝 수행
-- 즉 날짜별로 데이터를 나누어 저장하겠다는 의미


-- 2023년 1월 데이터만 저장하는 파티션 테이블
CREATE TABLE orders_2023_01
PARTITION OF orders
FOR VALUES FROM ('2023-01-01') TO ('2023-02-01');
-- 2023-01-01 이상 ~ 2023-02-01 미만 데이터 저장


-- 2023년 2월 데이터만 저장하는 파티션 테이블
CREATE TABLE orders_2023_02
PARTITION OF orders
FOR VALUES FROM ('2023-02-01') TO ('2023-03-01');
-- 2023-02-01 이상 ~ 2023-03-01 미만 데이터 저장

 

읽기 전용 데이터베이스

• 읽기 전용 데이터베이스는 주로 읽기 요청을 처리하는 데이터베이스 인스턴스입니다. 데이터를 주기적으로 동기화하여 최신 상태를 유지하면서 읽기 요청을 처리합니다. 이 방법은 주 데이터베이스에 대한 읽기 부하를 줄이고 성능을 최적화할 수 있습니다.
  • 읽기 전용 DB 및 Redis 사용 시 데이터 불일치 문제와 해결 전략
    1. Redis에서 데이터를 제거(퍼지)한 뒤 write DB에서는 이미 데이터가 'abc'로 변경됐지만, read DB에는 반영이 안 되어 redis에 다시 구 데이터('a')가 적재되는 데이터 불일치 현상이 발생할 수 있습니다.
    2. 이 현상은 데이터가 Redis에서 지워지고, read DB는 아직 업데이트 전 상태라서 사용자가 예전 데이터('a')를 다시 읽는 경우에 촉발됩니다.
    3. 대규모 사용자 요청이 동시에 처리될 때 이런 문제가 더욱 빈번히 발생할 수 있습니다.
    4. 근본 대책은 퍼지(furge) 대신 데이터 자체를 Redis에서 최신 값으로 직접 갱신하거나, 문제가 발생 시 Redis를 퍼지하여 강제로 DB에서 다시 가져오게 만드는 방식이 있습니다. 또한 캐시 데이터에 문제가 발생한 경우에는 Redis 데이터를 삭제한 뒤, 이후 요청 시 DB에서 다시 조회하여 Redis에 재저장하도록 처리할 수 있습니다.

쿼리 최적화

• 쿼리 최적화는 SQL 쿼리를 효율적으로 작성하여 데이터베이스의 읽기 성능을 향상시키는 방법입니다. 예를 들어, 불필요한 조인을 줄이고, 필요한 컬럼만 선택하며, 적절한 조건을 사용하는 것이 중요합니다. 쿼리 실행 계획을 분석하여 병목 지점을 찾아내고, 이를 최적화할 수 있습니다.

쓰기 요청 최적화

• 쓰기에서 가장 많은 시간을 소요하는 부분은 DB에 데이터를 생성하는 부분입니다. 이를 해결하기 위한 다양한 방법이 있습니다.

비동기 처리

• 쓰기 요청을 비동기 방식으로 처리하면 DB에 직접 접근하지 않고도 빠르게 응답을 반환할 수 있습니다
• 예를 들어 메시지를 바로 DB에 쓰지 않고, 큐에 넣어 나중에 처리하는 방법이 있습니다
• 이를 통해 애플리케이션은 요청을 빠르게 처리할 수 있습니다
• 비동기 처리는 특히 높은 트래픽을 처리할 때 유용합니다
• 사용자는 요청을 보낸 후 즉시 응답을 받기 때문에, 사용자 경험이 향상됩니다
• 비동기 처리 시 데이터 소실이나 오류를 방지하기 위해 큐에 데이터를 넣을 때 적절한 검증을 수행하고, 큐에 쌓인 데이터를 지속적으로 모니터링하여 실패한 요청을 재시도할 수 있는 메커니즘을 마련해야 합니다
• 또한, 데이터의 순서를 보장하고, 중복 처리를 방지하기 위한 고유 식별자(ID)를 사용해야 합니다

배치 처리

• 실시간으로 처리할 필요가 없는 쓰기 요청은 배치 처리를 통해 한꺼번에 처리할 수 있습니다
• 예를 들어 일정 시간마다 큐에 쌓인 메시지를 DB에 쓰는 방법이 있습니다.
• 예를 들어, 매일 자정에 하루 동안 수집된 로그 데이터를 한 번에 DB에 저장하는 방식으로 사용할 수 있습니다.
• 이를 통해 실시간 처리의 부담을 줄이고, 시스템 자원을 효율적으로 사용할 수 있습니다.
• 배치 처리 시에도 데이터 소실을 방지하기 위해 배치 작업 중 오류가 발생한 경우 이를 기록하고, 재시도할 수 있는 메커니즘을 마련해야 합니다.
• 또한, 배치 작업의 상태를 모니터링하고, 배치 작업이 완료되었는지 확인하는 프로세스를 마련해야 합니다.

분산 DB

• 단일 DB로 모든 쓰기 요청을 처리하기 어렵다면, 분산 DB를 사용하여 부하를 분산시킬 수 있습니다
• 예를 들어 여러 개의 DB 인스턴스를 사용하여 각 인스턴스가 특정 사용자 그룹의 데이터를 처리하도록 할 수 있습니다
• 분산 DB는 데이터를 여러 개의 노드에 분산 저장하여 고가용성과 확장성을 제공합니다
• 또한, 샤딩(Sharding) 기법을 통해 데이터베이스를 수평으로 분할하여 각 샤드가 독립적으로 쓰기 작업을 처리하도록 할 수 있습니다. 이를 통해 단일 노드의 부하를 줄이고, 시스템의 전체 성능을 향상시킬 수 있습니다.

📌 분산 DB 사용 시 데이터 일관성을 유지하기 위해 트랜잭션 관리와 데이터 동기화에 신경 써야 합니다. 예를 들어, 분산 트랜잭션이나 이벤트 소싱을 사용하여 데이터 일관성을 유지할 수 있습니다.

 

🗨️ 데이터 일관성 유지

📌 대규모 시스템에서는 데이터 일관성을 유지하는 것이 중요합니다. 이를 위해 분산 트랜잭션, 이벤트 소싱, CQRS(Command Query Responsibility Segregation) 등의 기법을 사용할 수 있습니다.

분산 트랜잭션

• 분산 트랜잭션은 여러 개의 독립된 시스템이나 데이터베이스에서 동시에 일어나는 트랜잭션을 일관되게 관리하는 방법입니다.
• 단일 트랜잭션이 여러 시스템에 걸쳐 발생할 때, 모든 시스템이 해당 트랜잭션을 성공적으로 완료하거나, 모든 시스템이 트랜잭션을 실패로 처리하도록 보장합니다. 이를 통해 데이터의 일관성을 유지할 수 있습니다.
• 분산 트랜잭션이 필요한 이유는 마이크로서비스 아키텍처에서 여러 서비스가 독립적으로 운영되기 때문입니다

주요 개념

• 트랜잭션(Transaction):
  • 데이터베이스의 상태를 변환하는 작업의 단위로, ACID(원자성, 일관성, 고립성, 지속성) 속성을 보장합니다.
    1. 원자성 (Atomicity): 트랜잭션은 전부 성공하거나 전부 실패하여, 부분적인 작업 수행이 없는 것을 보장합니다.
    2. 일관성 (Consistency): 트랜잭션이 완료된 후에도 데이터베이스는 모든 무결성 제약 조건을 유지합니다.
    3. 격리성 (Isolation): 동시에 실행되는 트랜잭션이 서로 간섭하지 않도록 보장합니다.
    4. 지속성 (Durability): 트랜잭션이 성공적으로 완료된 후의 결과는 시스템 장애가 발생해도 영구적으로 유지됩니다.
• 분산 트랜잭션(Distributed Transaction):
  • 여러 분산된 데이터 소스에 걸쳐 트랜잭션을 수행하는 작업입니다. 예를 들어, 여러 마이크로서비스나 데이터베이스에서 데이터를 동시에 업데이트하는 경우가 이에 해당합니다.
• 2PC (Two-Phase Commit):
  • 분산 트랜잭션을 관리하는 프로토콜로, 준비(Prepare) 단계와 커밋(Commit) 단계로 나누어 트랜잭션을 처리합니다.
    1. 준비 단계(Prepare Phase): 각 참여 노드는 트랜잭션 준비 상태를 확인하고, 준비 완료를 마스터 노드에 알립니다.
    2. 커밋 단계(Commit Phase): 마스터 노드는 모든 참여 노드가 준비되었음을 확인하고, 트랜잭션을 커밋하도록 지시합니다. 만약 준비가 완료되지 않은 노드가 있다면 트랜잭션을 롤백합니다.
• 사가 패턴(Saga Pattern):
  • 트랜잭션을 여러 단계로 나누어 처리하고, 각 단계가 독립적으로 커밋됩니다. 실패 시 보상 트랜잭션을 실행하여 상태를 롤백합니다.
    1. 주문 생성 단계: 사용자가 주문을 생성합니다.
    2. 결제 처리 단계: 결제 서비스가 주문 결제를 처리합니다.
    3. 재고 감소 단계: 재고 서비스가 주문된 상품의 재고를 감소시킵니다.
    4. 각 단계가 성공적으로 완료되면 다음 단계로 넘어가고, 실패하면 이전 단계에서 수행된 작업을 취소합니다.
• 이벤트 소싱(Event Sourcing):
  • 상태 변화를 이벤트로 기록하고, 해당 이벤트를 재생하여 현재 상태를 유지합니다. 이를 통해 분산 트랜잭션의 일관성을 유지할 수 있습니다.
    1. 데이터의 최종 상태만 저장하는 게 아니라, “무슨 일이 발생했는지”를 이벤트 로그처럼 순서대로 저장해두고,
       그 기록들을 다시 읽어 현재 상태를 만드는 방식
    2. 즉, 상태값 자체보다 “변경 과정”을 저장하는 방식
    3. ① 5,000원 입금 ② 10,000원 입금 ③ 5,000원 출금 ➔ "이 기록들(이벤트)을 순서대로 쭉 계산(재생)해보니 지금 10,000원이네!"
    4. 나중에 분산된 시스템 간에 데이터가 꼬이더라도, 발생한 일(이벤트)들의 기록이 고스란히 남아있으니 처음부터 다시 쭉 계산해서 정확한 현재 상태를 맞출 수 있다는 게 핵심

분산 트랜잭션의 장점

• 데이터 일관성 보장:
  • 분산된 여러 데이터 소스에 걸쳐 일관된 데이터 상태를 유지할 수 있습니다. 모든 트랜잭션이 성공적으로 완료되거나 모두 실패하도록 보장합니다.
• 확장성:
  • 분산 트랜잭션을 통해 여러 시스템이 독립적으로 동작하면서도, 필요한 경우 협력하여 일관된 상태를 유지할 수 있습니다. 이를 통해 시스템의 확장성을 높일 수 있습니다.
• 신뢰성:
  • 트랜잭션의 ACID 속성을 분산 환경에서도 유지할 수 있어 시스템의 신뢰성을 높입니다. 데이터 무결성과 일관성을 유지할 수 있습니다.
• 복구 가능성:
  • 트랜잭션 실패 시 롤백 메커니즘을 통해 상태를 복구할 수 있어, 시스템 안정성을 높일 수 있습니다.

분산 트랜잭션의 단점

• 복잡성 증가:
  • 분산 트랜잭션을 구현하고 관리하는 것은 복잡합니다. 특히, 여러 시스템 간의 트랜잭션 동기화와 데이터 일관성을 유지하는 것은 어려운 작업입니다.
• 성능 저하:
  • 2PC와 같은 프로토콜을 사용할 경우, 트랜잭션의 준비와 커밋 단계에서 지연이 발생할 수 있습니다. 이는 시스템의 성능을 저하시킬 수 있습니다.
• 네트워크 오버헤드:
  • 여러 시스템 간의 통신이 필요하므로 네트워크 오버헤드가 증가할 수 있습니다. 이는 트랜잭션 처리 속도를 저하시킬 수 있습니다.
• 복구의 어려움:
  • 분산 트랜잭션 실패 시, 모든 시스템에서 일관된 상태로 롤백하는 것이 어려울 수 있습니다. 특히, 부분적으로 실패한 트랜잭션을 처리하는 데 어려움이 있을 수 있습니다.

분산 트랜잭션의 예시

• 주문 생성과 결제 처리:
  • 한 사용자가 온라인 쇼핑몰에서 주문을 생성하고, 결제를 진행합니다. 이 과정에서 주문 서비스와 결제 서비스가 각각 독립된 데이터베이스를 사용한다고 가정합니다.
  • 분산 트랜잭션을 통해 두 서비스가 일관되게 주문과 결제를 처리해야 합니다. 트랜잭션이 성공적으로 완료되면 두 데이터베이스에 주문 및 결제 정보가 일관되게 저장됩니다.
• 2PC를 사용한 트랜잭션 처리:
  • 준비 단계: 주문 서비스와 결제 서비스가 트랜잭션을 준비합니다. 두 서비스가 모두 준비 완료 상태를 마스터 노드에 알립니다.
  • 커밋 단계: 마스터 노드는 두 서비스가 모두 준비되었음을 확인하고, 트랜잭션을 커밋하도록 지시합니다. 만약 하나의 서비스라도 준비되지 않았으면, 트랜잭션을 롤백합니다.

2PC는 중앙 조율자가 모든 서비스의 커밋 여부를 통제하는 방식이고, Saga 패턴은 각 작업을 순차적으로 처리하다가 실패 시 보상 트랜잭션으로 이전 작업을 되돌리는 방식이다.

 

🗨️ 2PC를 사용한 트랜잭션 처리

이벤트 소싱

• 이벤트 소싱은 데이터 상태 변화를 이벤트로 기록하고, 해당 이벤트들을 순차적으로 재생하여 현재 상태를 파악하는 방법입니다.
• 전통적인 데이터 저장 방식과 달리, 이벤트 소싱에서는 데이터 변경 자체가 아닌 변경 이벤트를 저장합니다.
• 이러한 방식은 특히 복잡한 비즈니스 로직을 다루는 시스템에서 데이터 일관성과 추적 가능성을 높이는 데 유용합니다.
• 하지만 복잡성이 증가할 수 있으므로, 시스템의 요구사항에 따라 신중하게 적용해야 합니다.

주요 개념

• 이벤트(Event):
  • 데이터의 상태 변화를 나타내는 기록입니다. 예를 들어, "주문 생성", "결제 완료", "주문 취소" 등이 이벤트가 될 수 있습니다.
• 이벤트 스토어(Event Store):
  • 이벤트를 저장하는 저장소입니다. 전통적인 데이터베이스 대신 이벤트를 순서대로 저장하는 스토리지입니다. 이는 이벤트의 불변성과 순차성을 보장합니다.
• 애그리게이트(Aggregate):
  • 관련된 이벤트를 모아 현재 상태를 재현할 수 있는 엔터티입니다. 애그리게이트는 도메인 모델의 일부분으로, 이벤트를 적용하여 상태를 변화시킵니다.
• 커맨드(Command):
  • 애그리게이트에 특정 동작을 지시하는 명령입니다. 커맨드는 이벤트를 생성하는 트리거 역할을 합니다.
• 프로젝션(Projection):
  • 이벤트를 읽기 모델로 변환하여 조회 성능을 최적화하는 방식입니다. 이벤트를 기반으로 읽기 전용 데이터베이스를 업데이트합니다.

이벤트 소싱의 장점

• 데이터 변경 이력 추적:
  • 모든 상태 변화를 이벤트로 기록하므로, 데이터 변경 이력을 완벽하게 추적할 수 있습니다. 이는 감사와 디버깅에 유용합니다.
• 복구 및 재생:
  • 이벤트를 재생하여 시스템의 현재 상태를 복구할 수 있습니다. 이는 데이터 손실이나 시스템 장애 시 유용합니다.
• CQRS와의 자연스러운 통합:
  • 이벤트 소싱은 CQRS(Command Query Responsibility Segregation)와 잘 어울립니다. 명령과 조회를 분리하여 성능과 확장성을 최적화할 수 있습니다.

이벤트 소싱의 단점

• 복잡성 증가:
  • 시스템 설계와 구현의 복잡성이 증가합니다. 이벤트 모델링과 이벤트 스토어 관리가 필요합니다.

• 읽기 성능:
  • 이벤트를 재생하여 현재 상태를 계산해야 하므로, 읽기 성능이 저하될 수 있습니다. 이를 해결하기 위해 프로젝션과 CQRS를 활용할 수 있습니다.

 

CQRS

• CQRS는 명령(Query)과 조회(Query)의 책임을 분리하는 소프트웨어 디자인 패턴입니다.
• 이 패턴은 읽기 작업과 쓰기 작업을 서로 다른 모델로 분리하여, 각 작업에 최적화된 구조를 사용할 수 있도록 합니다.
• CQRS는 시스템의 성능, 확장성, 유지보수성을 향상시키는 데 도움이 됩니다.

주요 개념

• 명령(Command):
  • 데이터를 변경하는 작업입니다. 예를 들어, 주문 생성, 결제 처리, 계정 업데이트 등이 명령에 해당합니다. 명령은 데이터베이스에 대한 쓰기 작업을 수행합니다.
  • 명령 모델은 데이터의 상태 변경을 담당합니다. 이는 복잡한 비즈니스 로직을 포함할 수 있으며, 데이터 무결성을 보장하기 위해 트랜잭션을 사용합니다.
• 조회(Query):
  • 데이터를 조회하는 작업입니다. 예를 들어, 주문 내역 조회, 계정 정보 조회 등이 조회에 해당합니다. 조회는 데이터베이스에 대한 읽기 작업을 수행합니다.
  • 조회 모델은 읽기 전용 데이터베이스 또는 캐시를 사용하여 빠른 응답을 제공합니다. 이는 단순한 데이터 조회를 위한 최적화된 구조를 가질 수 있습니다.

CQRS의 장점

• 성능 향상:
  • 읽기와 쓰기를 분리하여 각 작업에 최적화된 데이터 저장소와 인프라를 사용할 수 있습니다. 예를 들어, 조회 성능을 높이기 위해 읽기 전용 데이터베이스를 사용하거나, 캐시를 활용할 수 있습니다.
• 확장성:
  • 읽기와 쓰기를 독립적으로 확장할 수 있습니다. 예를 들어, 읽기 요청이 많은 경우 조회 모델을 수평으로 확장하여 부하를 분산시킬 수 있습니다.
• 유지보수성:
  • 비즈니스 로직이 명령 모델에 집중되므로, 복잡한 상태 변경 로직을 관리하기 쉽습니다. 또한, 읽기 모델은 단순화되어 유지보수하기 용이합니다.
• 데이터 일관성:
  • CQRS는 이벤트 소싱과 잘 어울립니다. 이벤트 소싱을 통해 데이터 상태 변경을 이벤트로 기록하고, 이벤트를 재생하여 현재 상태를 유지할 수 있습니다. 이를 통해 데이터 일관성을 보장할 수 있습니다.

CQRS의 단점

• 복잡성 증가:
  • 시스템 설계와 구현의 복잡성이 증가합니다. 명령 모델과 조회 모델을 각각 설계하고 관리해야 합니다.
• 데이터 동기화:
  • 명령 모델과 조회 모델 간의 데이터 동기화가 필요합니다. 이는 추가적인 구현과 관리가 필요합니다.

 

모니터링과 로깅

• 대규모 시스템에서 모니터링과 로깅은 시스템의 안정성과 성능을 유지하는 데 필수적인 역할을 합니다.
• 이를 통해 시스템의 상태를 실시간으로 파악하고, 문제 발생 시 신속하게 대응할 수 있습니다.
• 모니터링과 로깅 모두 다음의 항목을 주요 사항으로 봅니다.
  • 시스템 안정성 유지
  • 성능 최적화
  • 문제 예방 및 대응

모니터링

• 애플리케이션, DB, 캐시 등 각 컴포넌트의 성능을 모니터링합니다
• 이를 위해 Prometheus, Grafana 같은 도구를 사용할 수 있습니다.
• 시스템의 주요 지표(TPS, 응답 시간, 에러율 등)를 모니터링하고, 이상 징후를 감지하면 알림을 받을 수 있습니다
• 모니터링을 통해 시스템의 상태를 실시간으로 파악하고, 문제 발생 시 빠르게 대응할 수 있습니다. 또한, 모니터링 데이터를 기반으로 시스템의 성능을 분석하고, 최적화할 수 있습니다.

주요사항

1. 실시간 상태 파악:
   • 모니터링 도구는 시스템의 주요 지표(TPS, 응답 시간, 에러율 등)를 실시간으로 확인할 수 있게 합니다. 이를 통해 시스템의 현재 상태를 파악하고, 이상 징후를 빠르게 감지할 수 있습니다
2. 자동 알림:
   • 특정 임계치를 초과하는 경우 자동으로 알림을 받을 수 있어, 잠재적인 문제를 조기에 발견하고 대응할 수 있습니다.
3. 성능 분석:
   • 모니터링 데이터를 기반으로 시스템의 성능을 분석할 수 있습니다. 예를 들어, 특정 시간대에 트래픽이 급증하는 경우 해당 시간대에 자원을 추가로 할당하여 성능을 최적화할 수 있습니다.
4. 병목 지점 파악:
   • 모니터링을 통해 시스템의 병목 지점을 파악하고, 이를 최적화하여 전체 성능을 향상시킬 수 있습니다.
5. 사전 예방:
   • 실시간 모니터링을 통해 문제가 발생하기 전에 예방 조치를 취할 수 있습니다. 예를 들어, 디스크 사용량이 급증하는 경우 디스크 용량을 미리 확장할 수 있습니다.
6. 신속한 대응:
   • 문제가 발생했을 때 빠르게 감지하고 대응할 수 있어, 시스템 다운타임을 최소화할 수 있습니다.

완벽한 시스템 설계와 최적화가 이상적이지만, 데드라인과 변화하는 요구사항 등 현실적 제약으로 인해 빠른 프로토타입 구현과 점진적 개선 전략이 필요합니다.

빠른 프로토타입 개발과 필수 기능 우선 구현을 통해 애자일하게 시스템을 점진적으로 발전시키는 전략이 중요합니다

 

로깅

• 애플리케이션의 주요 이벤트를 로깅하여 문제 발생 시 원인을 추적할 수 있습니다
• 로그는 Elasticsearch, Logstash, Kibana(ELK 스택) 등을 사용하여 수집, 저장, 분석할 수 있습니다
• 이를 통해 시스템의 상태를 파악하고, 문제를 해결할 수 있습니다.
• 로깅은 시스템의 모든 중요한 이벤트를 기록하므로, 문제 발생 시 정확한 원인을 파악하는 데 큰 도움이 됩니다

주요사항

• 이벤트 추적:
  • 로깅을 통해 시스템에서 발생하는 모든 중요한 이벤트를 기록합니다. 이는 문제가 발생했을 때 원인을 추적하고 분석하는 데 도움이 됩니다
• 디버깅 및 오류 해결:
  • 로그를 분석하여 오류의 원인을 파악하고, 이를 기반으로 문제를 해결할 수 있습니다. 이는 특히 복잡한 대규모 시스템에서 문제 해결 시간을 단축시킵니다.
• 패턴 분석:
  • 로그 데이터를 분석하여 사용자의 행동 패턴이나 시스템의 성능 패턴을 이해할 수 있습니다. 이를 통해 시스템을 더 효율적으로 운영할 수 있습니다.
• 장기적 최적화:
  • 로그 데이터를 장기적으로 분석하여 시스템의 성능 저하 원인을 찾아내고, 지속적인 최적화를 진행할 수 있습니다.
• 원인 분석:
  • 문제가 발생한 후 로그를 분석하여 정확한 원인을 파악할 수 있습니다. 이는 같은 문제가 다시 발생하지 않도록 예방하는 데 도움이 됩니다.
• 법적 및 규제 요구 사항 준수:
  • 많은 산업에서는 로그 데이터를 보관하고 분석하는 것이 법적 및 규제 요구 사항을 준수하는 데 필수적입니다. 이는 감사 및 규제 대응에 중요한 역할을 합니다.

접속량 체크와 로그 수집

• 일반적으로 구글 애널리틱스(GA) 스크립트를 프론트에 삽입해 접속량을 확인하지만, 실제로는 정확하지 않은 경우가 많습니다.
• API 요청량과 같은 시스템의 실제 사용량을 알기 위해서는 로그를 남기는 습관이 중요하며, ELK 같은 로그 최적화 도구를 활용할 수 있습니다.
• 요청량 로깅이 되어야 대시보드를 통해 성능 측정 및 최적화가 가능하며, API 요청량 등은 반드시 로그로 적재해야 합니다.
• 인프라 자원은 곧 비용(돈)이므로, 정확한 데이터 수집을 통해 비용 절감 효과를 얻을 수 있습니다.
• 로그 및 모니터링 시스템 구축은 단순해 보이지만, 실제로는 구현되어 있지 않은 경우도 자주 있으며, 요구사항에 초기에 포함되지 않았거나 시간 부족 때문에 누락되는 일이 많으므로 반드시 챙겨야 합니다.

 

테스트와 배포

• 대규모 시스템에서는 테스트와 배포도 중요한 요소입니다.
• 시스템이 정상적으로 동작하는지 확인하고, 문제 발생 시 빠르게 대응할 수 있어야 합니다.   

테스트

• 단위 테스트(Unit Test):
  • 시스템의 개별 구성 요소를 테스트하여 각 부분이 예상대로 동작하는지 확인합니다.
  • JUnit, TestNG와 같은 도구를 사용하여 자동화된 단위 테스트를 작성하고 실행합니다.
  • 단위 테스트는 개발 초기 단계에서 발생할 수 있는 결함을 조기에 발견하고 수정하는 데 유용합니다.
• 통합 테스트(Integration Test):
  • 여러 구성 요소가 함께 동작하는지를 테스트합니다. 이는 개별 구성 요소들이 올바르게 상호작용하는지를 검증합니다.
  • Spring Boot에서는 @SpringBootTest 어노테이션을 사용하여 통합 테스트를 작성할 수 있습니다.
  • 통합 테스트는 시스템의 다양한 부분들이 함께 올바르게 동작하는지를 확인하여, 인터페이스 간의 문제를 발견하는 데 유용합니다.
• 부하 테스트(Load Test):
  • 시스템이 높은 트래픽 상황에서도 안정적으로 동작하는지를 테스트합니다.
  • Apache JMeter와 같은 도구를 사용하여 다양한 부하 시나리오를 설정하고 테스트를 수행합니다.
  • 부하 테스트를 통해 시스템의 성능 한계를 파악하고, 병목 지점을 찾아 최적화할 수 있습니다.
• 회귀 테스트(Regression Test):
  • 새로운 코드 변경이 기존 기능에 영향을 미치지 않는지 확인합니다.
  • 기존 테스트 케이스를 자동화하여 주기적으로 실행함으로써, 코드 변경으로 인한 결함을 방지할 수 있습니다.
  • 회귀 테스트는 지속적인 코드 변경에도 시스템의 안정성을 유지하는 데 중요합니다.
• 사용자 수용 테스트(UAT, User Acceptance Test):
  • 실제 사용자 환경에서 시스템을 테스트하여, 사용자가 요구하는 기능이 모두 제대로 동작하는지 확인합니다.
  • 사용자 피드백을 반영하여 시스템을 최종 조정하고, 배포 준비를 완료합니다.
  • 사용자 수용 테스트는 시스템이 실제 운영 환경에서 기대대로 동작하는지 확인하는 중요한 단계입니다.

배포

• 지속적인 통합(CI, Continuous Integration):
  • 개발자가 변경한 코드를 자주, 자동으로 빌드하고 테스트하여, 코드 변경 시점에서 발생할 수 있는 문제를 조기에 발견하고 해결합니다.
  • Jenkins, GitLab CI, Travis CI와 같은 도구를 사용하여 CI 파이프라인을 설정합니다.
  • CI는 코드 통합을 빠르고 효율적으로 수행하여 개발 주기를 단축시키고, 코드 품질을 높입니다.
• 지속적인 배포(CD, Continuous Deployment):
  • CI 파이프라인을 통해 검증된 코드를 자동으로 프로덕션 환경에 배포합니다.
  • Argo CD와 같은 도구를 사용하여 CD 파이프라인을 설정합니다.
  • CD는 코드 변경 사항을 빠르고 안전하게 프로덕션 환경에 적용하여, 새로운 기능을 신속하게 제공할 수 있습니다.
• Canary 배포:
  • 새로운 버전을 전체 시스템에 배포하기 전에, 일부 사용자에게만 배포하여 문제가 없는지 확인합니다.
  • 문제가 발생할 경우 빠르게 이전 버전으로 롤백할 수 있습니다.
  • Canary 배포는 리스크를 최소화하면서 새로운 기능을 단계적으로 도입할 수 있는 방법입니다.
• 블루-그린 배포(Blue-Green Deployment):
  • 두 개의 환경(블루와 그린)을 사용하여 하나는 현재 운영 중인 환경이고, 다른 하나는 새로운 버전을 배포하는 환경으로 사용합니다.
  • 새로운 버전을 그린 환경에 배포한 후, 모든 트래픽을 그린 환경으로 전환합니다. 문제가 발생하면 블루 환경으로 빠르게 롤백할 수 있습니다.
  • 블루-그린 배포는 무중단 배포를 가능하게 하며, 배포 실패 시 신속한 복구를 제공합니다.
• 롤링 배포(Rolling Deployment):
  • 새로운 버전을 점진적으로 배포하여, 각 서버를 순차적으로 업데이트합니다.
  • 시스템 가동 시간을 유지하면서 점진적으로 새로운 버전을 도입할 수 있습니다.
  • 롤링 배포는 대규모 시스템에서 무중단 배포를 구현하는 데 유용합니다.

 

🗨️ RabbitMQ

RabbitMQ란

• RabbitMQ는 메시지 브로커입니다. 메시지 브로커는 데이터(메시지)를 송신자(프로듀서)로부터 수신자(컨슈머)에게 전달하는 중간 매개체 역할을 합니다.
• RabbitMQ는 이러한 메시지를 큐(queue)에 저장하고, 필요할 때 적절한 수신자에게 전달합니다.

RabbitMQ의 역할

• 비동기 처리: 데이터를 비동기적으로 처리하여 시스템의 응답성을 높입니다.
  • 특정 작업이 완료될 때까지 기다리지 않고 다음 작업을 즉시 수행하는 방식
• 부하 분산: 여러 소비자에게 메시지를 분산시켜 시스템의 부하를 균형 있게 분산합니다.
• 내결함성: 메시지를 안전하게 저장하여 시스템 장애 시 데이터 손실을 방지합니다.
  • 컴퓨터나 시스템의 구성 요소 중 일부(하드웨어, 소프트웨어 등)에 고장이나 오류가 발생하더라도 서비스 중단 없이 시스템이 계속 정상적으로 작동하는 능력

장단점

장점

• 신뢰성(Reliability)
  • 메시지 지속성: RabbitMQ는 메시지를 디스크에 저장하여 시스템 장애 발생 시에도 메시지가 손실되지 않도록 합니다.
  • 확인 메커니즘: 메시지가 성공적으로 소비자에게 전달되었는지 확인하는 ACK(acknowledgment) 메커니즘을 지원합니다.
• 유연성(Flexibility)
  • 다양한 메시지 패턴: RabbitMQ는 여러 가지 메시지 전달 패턴(단일 소비자, 다중 소비자, 라운드 로빈, 팬아웃, 주제 기반 등)을 지원합니다.
  • 프로토콜 지원: 기본적으로 AMQP(Advanced Message Queuing Protocol)를 사용하지만, STOMP, MQTT 등 다양한 프로토콜도 지원합니다.
• 확장성(Scalability)
  • 클러스터링: RabbitMQ는 클러스터링을 통해 여러 노드로 구성된 환경에서 높은 가용성과 부하 분산을 제공합니다.
  • 분산 아키텍처: 페더레이션(Federation) 등을 사용하여 여러 RabbitMQ 인스턴스 간에 메시지를 전달함으로써 분산된 메시징 시스템을 구축할 수 있습니다.
• 관리 및 모니터링(Manageability and Monitoring)
  • 관리 인터페이스: 웹 기반 관리 인터페이스를 통해 큐, 익스체인지, 바인딩 등을 쉽게 관리할 수 있습니다.
  • 플러그인 시스템: 다양한 플러그인을 통해 기능을 확장할 수 있습니다(예: 관리 플러그인, 모니터링 플러그인 등).
• 성능(Performance)
  • 높은 처리량: 적절히 구성된 RabbitMQ는 높은 메시지 처리량을 제공하여 대규모 애플리케이션에서도 효과적으로 사용할 수 있습니다.

단점

• 설정 및 운영 복잡성(Setup and Operational Complexity)
  • 복잡한 설정: RabbitMQ의 초기 설정이 다소 복잡할 수 있으며, 클러스터링 및 분산 환경에서는 더욱 많은 설정이 필요합니다.
  • 운영 관리: 대규모 환경에서 RabbitMQ를 운영하고 관리하는 데 있어서 추가적인 노력이 필요할 수 있습니다.
• 성능 문제(Performance Issues)
  • 메시지 브로커 오버헤드: RabbitMQ는 모든 메시지를 중앙 브로커를 통해 전달하기 때문에, 높은 트래픽 상황에서는 브로커의 오버헤드가 발생할 수 있습니다.
  • 대규모 메시지 처리: 매우 대규모의 메시지를 처리할 때 성능 저하가 발생할 수 있으며, 이러한 경우에는 적절한 클러스터링 및 최적화가 필요합니다.
• 운영 비용(Operational Costs)
  • 리소스 소비: RabbitMQ는 메모리와 CPU 자원을 많이 소비할 수 있어, 충분한 리소스를 제공해야 원활하게 운영될 수 있습니다.
  • 모니터링 및 유지보수: 지속적인 모니터링과 유지보수가 필요하며, 이를 위해 추가적인 인력과 비용이 발생할 수 있습니다.
• 제한된 메시지 크기(Limited Message Size)
  • 메시지 크기 제한: RabbitMQ는 매우 큰 메시지 처리에 제한이 있을 수 있으며, 대용량 파일 전송에는 적합하지 않을 수 있습니다.
• 러닝 커브(Learning Curve)
  • 학습 필요성: RabbitMQ의 개념과 설정을 이해하는 데 시간이 걸릴 수 있습니다.

 

RabbitMQ의 기본 구성 요소

• 메시지(Message)
  • 메시지는 RabbitMQ를 통해 전달되는 데이터 단위입니다. 예를 들어, 사용자 등록 정보나 주문 내역이 메시지가 될 수 있습니다.
• 프로듀서(Producer)
  • 메시지를 생성하고 RabbitMQ에 보내는 역할을 합니다. 예를 들어, 웹 애플리케이션이 사용자 등록 정보를 RabbitMQ에 보내는 경우 프로듀서가 됩니다.
• 큐(Queue)
  • 메시지를 저장하는 장소입니다. 메시지는 큐에 저장되었다가 소비자에게 전달됩니다. 큐는 FIFO(First In, First Out) 방식으로 메시지를 처리합니다.
• 컨슈머(Consumer)
  • 큐에서 메시지를 가져와 처리하는 역할을 합니다. 예를 들어, 이메일 발송 서비스가 큐에서 사용자 등록 정보를 가져와 환영 이메일을 보내는 경우 컨슈머가 됩니다.
• 익스체인지(Exchange)
  • 메시지를 적절한 큐로 라우팅하는 역할을 합니다. 프로듀서는 메시지를 직접 큐에 보내지 않고, 익스체인지에 보내며, 익스체인지는 메시지를 적절한 큐로 전달합니다.

 

RabbitMQ와 AMQP

• RabbitMQ는 AMQP(Advanced Message Queuing Protocol)를 사용합니다.
• AMQP는 메시지 브로커를 위한 프로토콜로, 메시지의 생성, 전송, 큐잉, 라우팅 등을 표준화하여 메시지 브로커가 상호 운용될 수 있게 합니다.
• 여기서 프로토콜(Protocol)은 컴퓨터 네트워크에서 데이터를 주고받기 위한 일련의 규칙과 절차를 말합니다. 쉽게 말해, 컴퓨터나 장치들이 서로 통신할 때 어떻게 소통해야 하는지를 정해 놓은 약속입니다.

AMQP의 주요 개념

• 메시지(Message): 전송되는 데이터 단위입니다.
• 큐(Queue): 메시지를 저장하고 전달하는 구조입니다.
• 익스체인지(Exchange): 메시지를 큐로 라우팅하는 역할을 합니다.
• 바인딩(Binding): 익스체인지와 큐를 연결하는 설정입니다. 바인딩을 통해 메시지가 어느 큐로 전달될지 정의합니다.
  • 큐의 이름과 바인딩의 이름은 일치시킨다!

익스체인지 유형

• 메시지 브로커가 메시지를 교환기에서 큐로 라우팅하는 방식입니다.
• 익스체인지는 다양한 방식으로 메시지를 라우팅할 수 있으며, 주로 메시지의 라우팅 키와 바인딩 키 또는 패턴을 기반으로 작동합니다.

1) Direct Exchange

• 라우팅 키가 정확히 일치하는 큐로 메시지를 전달합니다.
• 예를 들어, 라우팅 키가 error인 메시지는 error라는 바인딩 키를 가진 큐로 전달됩니다.

→ 현업에서는 거의 다 이걸로 작업. 이것만 잘 알아두자!

 

2) Topic Exchange

• 라우팅 키의 패턴을 사용하여 메시지를 라우팅합니다. 패턴에는 와일드카드 * (단어 하나)와 # (0개 이상의 단어)가 사용됩니다.
• 예를 들어, 라우팅 키가 quick.orange.rabbit인 메시지는 바인딩 키가 *.orange.*인 큐로 전달됩니다.

3) Fanout Exchange

• 라우팅 키를 무시하고 교환기에 바인딩된 모든 큐로 메시지를 브로드캐스트합니다.
• 모든 바인딩된 큐로 메시지가 전달됩니다.

4) Headers Exchange

• 라우팅 키 대신 메시지의 헤더를 기반으로 메시지를 라우팅합니다.
• 헤더 값과 바인딩된 헤더 값이 일치하는 큐로 메시지를 전달합니다.

 

실습

 

 

RabbitMQ

• 도커를 사용하여 RabbitMQ를 설치합니다.

docker run -d --name rabbitmq -p5672:5672 -p 15672:15672 --restart=unless-stopped rabbitmq:management

• localhost:15672에 접속하면 로그인 페이지가 보입니다. guest/guest를 입력하여 접속하면 대시보드를 볼 수 있습니다.

 

Order Application

• start.spring.io 에 접속하여 프로젝트를 생성합니다.

프로젝트를 생성하면 build.gradle의 디펜던시는 아래와 같습니다.

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	implementation 'org.springframework.boot:spring-boot-starter-web'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

application.properties

# 현재 스프링 애플리케이션 이름
# Eureka 같은 서비스 등록 시 사용되기도 함
spring.application.name=order

# RabbitMQ Exchange 이름
# Exchange는 메시지를 어느 Queue로 보낼지 결정하는 중간 라우터 역할
message.exchange=market

# 상품 서비스용 Queue 이름
# Product 관련 메시지가 저장될 Queue
message.queue.product=market.product

# 결제 서비스용 Queue 이름
# Payment 관련 메시지가 저장될 Queue
message.queue.payment=market.payment

# RabbitMQ 서버 주소
# localhost = 현재 내 컴퓨터
spring.rabbitmq.host=localhost

# RabbitMQ 기본 연결 포트
# 애플리케이션이 RabbitMQ와 통신할 때 사용
spring.rabbitmq.port=5672

# RabbitMQ 로그인 계정
spring.rabbitmq.username=guest

OrderApplicationQueueConfig.java

// RabbitMQ의 Queue, Exchange, Binding 등을
// 스프링 Bean으로 등록하는 설정 클래스

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
// 스프링 설정 클래스라는 의미
// 내부 @Bean 메서드들을 스프링 컨테이너에 등록함
public class OrderApplicationQueueConfig {

    // application.properties 값 주입
    // ${...} 문법으로 properties 값 가져옴
    @Value("${message.exchange}")
    private String exchange;

    @Value("${message.queue.product}")
    private String queueProduct;

    @Value("${message.queue.payment}")
    private String queuePayment;

    @Bean
    // RabbitMQ Exchange 생성
    // Exchange는 메시지를 적절한 Queue로 전달하는 역할
    public TopicExchange exchange() {
        return new TopicExchange(exchange);
    }

    @Bean
    // 상품 Queue 생성
    // Queue는 메시지가 실제 저장되는 공간
    public Queue queueProduct() {
        return new Queue(queueProduct);
    }

    @Bean
    // 결제 Queue 생성
    public Queue queuePayment() {
        return new Queue(queuePayment);
    }

    @Bean
    // Product Queue와 Exchange 연결
    // with(queueProduct) 부분은 Routing Key
    // 즉 어떤 메시지를 어떤 Queue로 보낼지 기준값
    public Binding bindingProduct() {
        return BindingBuilder
                .bind(queueProduct()) // Queue 선택
                .to(exchange())       // Exchange 연결
                .with(queueProduct);  // Routing Key 설정
    }

    @Bean
    // Payment Queue와 Exchange 연결
    public Binding bindingPayment() {
        return BindingBuilder
                .bind(queuePayment())
                .to(exchange())
                .with(queuePayment);
    }
}

OrderController.java

import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

@RestController
// REST API Controller
// 반환값이 JSON/String 형태로 HTTP 응답됨
@RequiredArgsConstructor
// final 필드 생성자 자동 생성
// 생성자 주입을 쉽게 해주는 Lombok 기능
public class OrderController {

    // OrderService 의존성 주입
    // final = 생성자 주입 대상
    private final OrderService orderService;

    @GetMapping("/order/{id}")
    // GET 요청 매핑
    // ex) /order/1
    public String order(@PathVariable String id) {

        // URL의 {id} 값을 메서드 파라미터로 받음
        // ex) /order/1 -> id = "1"

        // 주문 생성 로직 실행
        orderService.createOrder(id);

        return "Order complete";
    }
}

OrderService.java

import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

@Service
// 비즈니스 로직 담당 클래스
@RequiredArgsConstructor
public class OrderService {

    // 상품 Queue 이름 주입
    @Value("${message.queue.product}")
    private String productQueue;

    // 결제 Queue 이름 주입
    @Value("${message.queue.payment}")
    private String paymentQueue;

    // RabbitMQ 메시지 전송 객체
    // RabbitTemplate = RabbitMQ와 통신하는 핵심 클래스
    private final RabbitTemplate rabbitTemplate;

    public void createOrder(String orderId) {

        // 상품 Queue로 메시지 전송
        // convertAndSend():
        // 객체를 메시지로 변환(convert) 후 전송(send)
        //
        // 첫 번째 파라미터:
        // 어떤 Queue(Routing Key)로 보낼지
        //
        // 두 번째 파라미터:
        // 실제 보낼 데이터
        rabbitTemplate.convertAndSend(productQueue, orderId);

        // 결제 Queue로 동일한 주문 ID 전송
        // 즉 하나의 주문 요청을
        // Product Service와 Payment Service가 각각 처리 가능
        rabbitTemplate.convertAndSend(paymentQueue, orderId);
    }
}

 

주문 요청 발생
↓
OrderService
↓
RabbitMQ로 메시지 전송
↓
Product Queue 저장
Payment Queue 저장
↓
각 서비스가 비동기로 처리

 

• 애플리케이션을 런하여 /order/1 로 요청을 보냅니다.
• 그 후 http://localhost:15672로 접속하여 RabbitMQ의 Exchange와 Queue를 확인할 수 있습니다. 또한 Queue 에서는 현재 발행된 메시지가 Total에 쌓여 있는 것을 확인할 수 있습니다.

• Queue and Stream 페이지에서 큐 이름을 클릭하여 상세페이지로 이동한 후 스크롤하여 Get Messages 섹션으로 가서 Get Message를 클릭하면 현재 큐에 쌓여 있는 메시지를 조회 할 수 있습니다.

 

Payment Application

• start.spring.io 에 접속하여 프로젝트를 생성합니다.

프로젝트를 생성하면 build.gradle의 디펜던시는 아래와 같습니다.

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

payment - application.properties

# 현재 스프링 애플리케이션 이름
# 로그나 서비스 식별용으로 사용 가능
spring.application.name=payment

# Payment Service가 구독할 Queue 이름
# Order Service에서 market.payment로 보낸 메시지를 여기서 받음
message.queue.payment=market.payment

# RabbitMQ 서버 주소
# localhost = 내 컴퓨터에서 실행 중인 RabbitMQ
spring.rabbitmq.host=localhost

# RabbitMQ 애플리케이션 연결 포트
spring.rabbitmq.port=5672

# RabbitMQ 로그인 계정
spring.rabbitmq.username=guest

# RabbitMQ 로그인 비밀번호
spring.rabbitmq.password=guest

PaymentEndpoint.java

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Slf4j
// log.info(), log.error() 같은 로그 객체를 자동 생성해주는 Lombok 어노테이션
@Component
// 스프링 Bean으로 등록
// Controller는 HTTP 요청을 받는 역할이고,
// 여기서는 RabbitMQ 메시지를 받는 객체라 Component로 등록
public class PaymentEndpoint {

    @Value("${spring.application.name}")
    // application.properties의 spring.application.name 값을 주입
    // 여기서는 "payment" 값이 들어감
    private String appName;

    @RabbitListener(queues = "${message.queue.payment}")
    // RabbitMQ Queue를 구독하는 리스너
    // market.payment Queue에 메시지가 들어오면 이 메서드가 자동 실행됨
    public void receiveMessage(String orderId) {

        // Queue에서 꺼낸 메시지가 orderId 파라미터로 들어옴
        // Order Service에서 "1"을 보냈다면 orderId = "1"
        log.info("receive orderId:{}, appName : {}", orderId, appName);

        // 실제 프로젝트라면 여기서 결제 처리 로직을 수행함
        // ex) 결제 승인, 결제 상태 변경, 결제 실패 처리 등
    }
}

• 애플리케이션을 실행하면 receiveMessage의 로그가 찍히는 것을 확인할 수 있습니다. Order 프로젝트에서 발행한 메시지가 Payment Consumer가 실행되자마자 소모된 것입니다.

 

Payment 애플리케이션 실행
↓
@RabbitListener가 market.payment Queue 감시 시작
↓
Queue 안에 메시지 있음
↓
자동으로 메시지 꺼냄
↓
receiveMessage(String orderId) 실행
↓
log.info(...) 찍힘
↓
정상 처리되면 메시지 ack
↓
Queue에서 메시지 사라짐

 

Product Application

• start.spring.io 에 접속하여 프로젝트를 생성합니다.

프로젝트를 생성하면 build.gradle의 디펜던시는 아래와 같습니다.

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

product - application.properties

# 현재 스프링 애플리케이션 이름
# 로그나 서비스 식별용으로 사용 가능
spring.application.name=product

# Product Service가 구독할 Queue 이름
# Order Service에서 market.product로 보낸 메시지를 여기서 받음
message.queue.product=market.product

# RabbitMQ 서버 주소
spring.rabbitmq.host=localhost

# RabbitMQ 애플리케이션 연결 포트
spring.rabbitmq.port=5672

# RabbitMQ 로그인 계정
spring.rabbitmq.username=guest

# RabbitMQ 로그인 비밀번호
spring.rabbitmq.password=guest

ProductEndpoint.java

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Slf4j
// 로그 출력을 위한 Lombok 어노테이션
@Component
// 스프링이 관리하는 객체로 등록
// RabbitMQ 메시지를 받기 위해 스프링 컨테이너에 등록되어야 함
public class ProductEndpoint {

    @Value("${spring.application.name}")
    // application.properties의 spring.application.name 값을 가져옴
    // 여기서는 "product" 값이 들어감
    private String appName;

    @RabbitListener(queues = "${message.queue.product}")
    // market.product Queue를 감시함
    // 해당 Queue에 메시지가 들어오면 receiveMessage()가 자동 호출됨
    public void receiveMessage(String orderId) {

        // RabbitMQ에서 받은 메시지 출력
        // Order Service에서 보낸 주문 ID를 확인하는 용도
        log.info("receive orderId:{}, appName : {}", orderId, appName);

        // 실제 프로젝트라면 여기서 상품/재고 관련 로직을 처리함
        // ex) 재고 확인, 재고 차감, 상품 상태 변경 등
    }
}

 

Order Service
↓ 메시지 전송
RabbitMQ Queue
↓
Payment Service는 market.payment를 듣고 있음
Product Service는 market.product를 듣고 있음
↓
각자 자기 Queue에 메시지가 들어오면 자동 실행

 

• 컨슈머가 라운드로빈으로 메시지를 전달받는 것을 확인하기 위해서 Intellij에서 구성편집에 들어가 두 개의 Product를 생성합니다. application.name 옵션을 통해 2를 구분할 수 있도록 합니다.

• order에서 요청할 때마다 Product Application이 번갈아가면서 메시지를 수신 받는 것을 확인할 수 있습니다.

 

RabbitMQ 예제 프로젝트 구조 정리

• 이번 예제는 Order, Product, Payment 총 3개의 Spring Boot 프로젝트로 구성된다.
• 각 프로젝트는 직접 HTTP로 서로를 호출하는 것이 아니라, RabbitMQ Queue를 통해 메시지를 주고받는다.

프로젝트	역할	주요 코드	Queue
Order	주문 요청을 받고 메시지를 발행하는 Producer	OrderController, OrderService, OrderApplicationQueueConfig	market.product, market.payment로 메시지 전송
Product	상품/재고 관련 메시지를 소비하는 Consumer	ProductEndpoint	market.product 구독
Payment	결제 관련 메시지를 소비하는 Consumer	PaymentEndpoint	market.payment 구독

 

각 프로젝트의 차이

구분	Order 프로젝트	Product 프로젝트	Payment 프로젝트
역할	메시지 보내는 쪽	메시지 받는 쪽	메시지 받는 쪽
RabbitMQ 기준	Producer	Consumer	Consumer
주요 어노테이션	@RestController, RabbitTemplate	@RabbitListener	@RabbitListener
하는 일	/order/{id} 요청을 받으면 Queue에 주문 ID 전송	market.product Queue의 메시지를 받아 상품 처리	market.payment Queue의 메시지를 받아 결제 처리
실행 결과	Queue에 메시지가 쌓임	메시지를 소비하면 Queue에서 사라짐	메시지를 소비하면 Queue에서 사라짐

 

전체 흐름

1. 사용자가 Order 프로젝트에 /order/1 요청을 보낸다.
2. Order Service는 RabbitMQ에 주문 ID 1을 메시지로 발행한다.
3. 메시지는 각각 market.product, market.payment Queue에 들어간다.
4. Product 프로젝트는 market.product Queue를 구독하고 있다가 메시지를 가져간다.
5. Payment 프로젝트는 market.payment Queue를 구독하고 있다가 메시지를 가져간다.
6. Consumer가 메시지를 정상 처리하면 Queue에서 메시지가 제거된다.

핵심 정리

• RabbitMQ 구조에서는 Order 서비스가 Product, Payment 서비스를 직접 호출하지 않는다.
• 대신 RabbitMQ에 메시지만 넣고, Product와 Payment 서비스가 각자 자신의 Queue를 구독하여 필요한 작업을 처리한다.
• 즉, 서비스 간 직접 의존성을 줄이고 비동기적으로 작업을 분리할 수 있다.

 

🗨️ Kafka

Kafka란?

• Kafka는 분산 스트리밍 플랫폼으로, 주로 실시간 데이터 피드의 빅 데이터 처리를 목적으로 사용됩니다.
  • 분산 스트리밍 플랫폼 풀어서 말하면:
    1. 분산 = 여러 서버가 나눠 처리
    2. 스트리밍 = 데이터가 계속 흐름
    3. 플랫폼 = 그걸 관리하는 시스템
  • Kafka는 대량의 데이터를 실시간으로 여러 서버에 전달하고 처리하기 위한 메시지 저장/전송 시스템이다.
• Kafka는 메시지 큐와 유사하지만, 대용량 데이터 스트림을 저장하고 실시간으로 분석하거나 처리하는 데 중점을 둡니다.
  • RabbitMQ가 메시지 전달 자체에 초점을 둔다면, Kafka는 대량의 데이터 흐름을 저장하고 실시간 처리하는 데 더 강점을 가진다.

Kafka의 역할

• 실시간 데이터 처리: 대용량 데이터를 실시간으로 처리하고 분석합니다.
• 데이터 통합: 다양한 소스에서 데이터를 수집하고 이를 통합하여 분석합니다.
• 내결함성: 데이터 손실 없이 안정적으로 데이터를 저장하고 전송합니다.

장단점

장점

• 신뢰성(Reliability)
  • 데이터 복제: 데이터를 여러 브로커에 복제하여 저장하므로, 단일 브로커 장애 시에도 데이터 손실을 방지할 수 있습니다.
  • 확인 메커니즘: 데이터가 소비자에게 성공적으로 전달되었는지 확인하는 기능을 제공합니다.
• 유연성(Flexibility)
  • 다양한 소비자 패턴: 여러 소비자가 동시에 데이터를 구독할 수 있습니다.
  • 프로토콜 지원: 기본적으로 Kafka의 프로토콜을 사용하지만, 다양한 클라이언트를 통해 다른 언어에서도 사용할 수 있습니다.
• 확장성(Scalability)
  • 분산 시스템: 클러스터링을 통해 여러 노드에서 데이터를 분산 처리할 수 있습니다.
  • 수평 확장: 브로커와 파티션을 추가하여 쉽게 확장할 수 있습니다.
• 성능(Performance)
  • 높은 처리량: 대용량 데이터를 실시간으로 빠르게 처리할 수 있습니다.
  • 저지연: 데이터 전송의 지연을 최소화하여 실시간 처리가 가능합니다.
• 관리 및 모니터링(Manageability and Monitoring)
  • 관리 도구: 다양한 관리 도구를 통해 클러스터를 모니터링하고 관리할 수 있습니다.
  • 플러그인 시스템: 다양한 플러그인을 통해 기능을 확장할 수 있습니다.

단점

• 설정 및 운영 복잡성(Setup and Operational Complexity)
  • 복잡한 설정: 초기 설정이 다소 복잡할 수 있으며, 클러스터링 및 분산 환경에서는 더 많은 설정이 필요합니다.
  • 운영 관리: 대규모 환경에서 운영하고 관리하는 데 추가적인 노력이 필요할 수 있습니다.
• 성능 문제(Performance Issues)
  • 브로커 오버헤드: 높은 트래픽 상황에서는 브로커의 오버헤드가 발생할 수 있습니다.
  • 대규모 메시지 처리: 매우 대규모의 메시지를 처리할 때 성능 저하가 발생할 수 있으며, 이러한 경우 적절한 클러스터링 및 최적화가 필요합니다.
• 운영 비용(Operational Costs)
  • 리소스 소비: 메모리와 CPU 자원을 많이 소비할 수 있어, 충분한 리소스를 제공해야 원활하게 운영될 수 있습니다.
  • 모니터링 및 유지보수: 지속적인 모니터링과 유지보수가 필요하며, 이를 위해 추가적인 인력과 비용이 발생할 수 있습니다.
• 러닝 커브(Learning Curve)
  • 학습 필요성: 개념과 설정을 이해하는 데 시간이 걸릴 수 있으며 다소 난이도가 있습니다.

 

Kafka의 기본 구성 요소

Producer가 Topic에 메시지를 저장하면, 메시지는 여러 Partition으로 나뉘어 Broker들에 분산 저장되고, Consumer Group의 Consumer들이 각 Partition을 병렬 처리한다.

 

• 메시지(Message)
  • 메시지는 Kafka를 통해 전달되는 데이터 단위입니다. 예를 들어, 로그 데이터나 이벤트 데이터가 메시지가 될 수 있습니다.
  • 메시지는 키(key), 값(value), 타임스탬프(timestamp), 그리고 몇 가지 메타데이터로 구성됩니다.
• 프로듀서(Producer)
  • 메시지를 생성하고 Kafka에 보내는 역할을 합니다. 예를 들어, 웹 애플리케이션이 로그 데이터를 Kafka에 보내는 경우 프로듀서가 됩니다.
  • 프로듀서는 특정 토픽(topic)에 메시지를 보냅니다.
• 토픽(Topic)
  • 메시지를 저장하는 장소입니다. 메시지는 토픽에 저장되었다가 소비자에게 전달됩니다.
  • 토픽은 여러 파티션(partition)으로 나누어질 수 있으며, 파티션은 메시지를 순서대로 저장합니다. 파티션을 통해 병렬 처리가 가능합니다.
  • 예: “user-activity”라는 토픽에 사용자의 활동 로그를 저장할 수 있습니다.
• 파티션(Partition)
  • 파티션은 토픽을 물리적으로 나눈 단위로, 각 파티션은 독립적으로 메시지를 저장하고 관리합니다.
  • 각 파티션은 메시지를 순서대로 저장하며, 파티션 내의 메시지는 고유한 오프셋(offset)으로 식별됩니다.
  • 파티션을 통해 데이터를 병렬로 처리할 수 있으며, 클러스터 내의 여러 브로커에 분산시켜 저장할 수 있습니다.
• 키(Key)
  • 키는 메시지를 특정 파티션에 할당하는 데 사용되는 값입니다.
  • 동일한 키를 가진 메시지는 항상 동일한 파티션에 저장됩니다.
  • 예를 들어, 특정 사용자 ID를 키로 사용하여 해당 사용자의 모든 이벤트가 동일한 파티션에 저장되도록 할 수 있습니다.
• 컨슈머(Consumer)
  • 토픽에서 메시지를 가져와 처리하는 역할을 합니다.
  • 컨슈머는 특정 컨슈머 그룹(consumer group)에 속하며, 같은 그룹에 속한 컨슈머들은 토픽의 파티션을 분산 처리합니다.
  • 기본적으로 컨슈머는 스티키 파티셔닝(Sticky Partitioning)을 사용합니다. 이는 특정 컨슈머가 특정 파티션에 붙어서 계속해서 데이터를 처리하는 방식으로, 이는 데이터 지역성을 높여 캐시 히트율을 증가시키고 전반적인 처리 성능을 향상시킵니다.

Kafka는 Topic을 여러 Partition으로 나누고, Consumer Group의 Consumer들이 각 Partition을 나눠 병렬 처리한다.

Topic: order

Partition 0
[메시지1][메시지2]

Partition 1
[메시지3][메시지4]

Partition 2
[메시지5][메시지6]

↓

Consumer Group A

Consumer1 → Partition0 처리
Consumer2 → Partition1 처리
Consumer3 → Partition2 처리

• 브로커(Broker)
  • Kafka 클러스터의 각 서버를 의미하며, 메시지를 저장하고 전송하는 역할을 합니다.
  • 하나의 Kafka 클러스터는 여러 브로커로 구성될 수 있으며, 각 브로커는 하나 이상의 토픽 파티션을 관리합니다.

Broker는 Kafka 메시지를 저장하고 관리하는 서버이며, Kafka Cluster는 여러 Broker 서버들을 묶어 하나의 Kafka 시스템처럼 동작하는 구조이다.

Kafka Cluster
 ├─ Broker1
 │    └─ Partition0
 ├─ Broker2
 │    └─ Partition1
 └─ Broker3
      └─ Partition2

• 주키퍼(Zookeeper)
  • Kafka 클러스터를 관리하고 조정하는 데 사용되는 분산 코디네이션 서비스입니다.
  • 주키퍼는 브로커의 메타데이터를 저장하고, 브로커 간의 상호작용을 조정합니다.

Zookeeper는 Kafka Broker들의 상태와 메타데이터를 관리하며, 여러 Broker가 정상적으로 동작하도록 조율하는 관리 시스템이다.

 

Kafka와 RabbitMQ의 차이점

1) 설계 철학

• RabbitMQ: 전통적인 메시지 브로커로, 메시지의 안정적 전달과 큐잉에 중점을 둡니다.
• Kafka: 분산 스트리밍 플랫폼으로, 대규모 실시간 데이터 스트림의 저장과 분석에 중점을 둡니다.

2) 메시지 모델

• RabbitMQ: 큐(queue)를 중심으로 메시지를 전달합니다. 메시지는 큐에 저장되고, 큐에서 하나 이상의 컨슈머에게 전달됩니다.
• Kafka: 토픽(topic)을 중심으로 메시지를 저장합니다. 메시지는 토픽의 파티션에 저장되고, 컨슈머는 이 파티션에서 메시지를 읽습니다.

3) 메시지 지속성

• RabbitMQ: 메시지를 메모리나 디스크에 저장할 수 있으며, 일반적으로 단기 저장을 목표로 합니다.
• Kafka: 메시지를 디스크에 저장하며, 장기 저장을 목표로 합니다. 데이터 로그는 설정된 기간 동안 보존됩니다.

4) 사용

• RabbitMQ: 작업 큐, 요청/응답 패턴, 비동기 작업 처리 등 전통적인 메시지 큐 사용 사례에 적합합니다.
• Kafka: 실시간 데이터 스트리밍, 로그 수집 및 분석, 이벤트 소싱 등 대규모 데이터 스트림 처리에 적합합니다.

항목	RabbitMQ	Apache Kafka
핵심 목적	메시지를 빠르고 안정적으로 전달	대량의 데이터를 오래 저장하고 실시간 처리
데이터 구조	Queue 기반	Topic + Partition 기반
메시지 처리 방식	Consumer가 가져가면 Queue에서 제거되는 경우가 많음	메시지를 로그처럼 저장하고 Consumer가 읽어감
데이터 저장 방식	메모리 기반 처리에 강함, 필요 시 디스크 저장	기본적으로 디스크에 계속 저장
메시지 보관	보통 처리 후 빠르게 소비됨	설정 기간 동안 계속 보관 가능
재조회	이미 소비한 메시지는 다시 읽기 어려움	Offset 기반으로 과거 메시지 재조회 가능
처리 특징	작업 처리(Job Queue), 비동기 요청 처리에 적합	대규모 로그/이벤트 스트림 처리에 적합
병렬 처리	Queue 단위 소비 중심	Partition 기반 대규모 병렬 처리
대표 사용 예시	주문 처리, 이메일 발송, 알림 큐	로그 수집, 실시간 분석, 이벤트 소싱
핵심 느낌	“메시지 전달 시스템”	“실시간 데이터 저장/스트리밍 시스템”

 

• RabbitMQ는 “메시지를 잘 전달하는 것”에 더 초점이 맞춰져 있다.
• Kafka는 “대량의 데이터를 로그처럼 저장하고 계속 재사용하는 것”에 더 초점이 맞춰져 있다.
• RabbitMQ는 작업 처리(Job Queue) 느낌에 가깝고, Kafka는 실시간 데이터 파이프라인 느낌에 가깝다.
• Kafka는 메시지를 디스크에 장기간 저장하며, Consumer가 나중에 다시 읽을 수 있다.
• Kafka는 Topic을 Partition으로 나누어 여러 Consumer가 병렬 처리하기 쉽다.

 

 

+) 대규모 예약 시스템의 순서 보장과 동시성 처리

• 대규모 판매 시스템(예: 한정 수량 판매)에서 순서 보장이 필요한 경우, 다양한 구현 방식을 고민해야 하며, 면접에서도 자주 출제되는 문제입니다.
• 티켓 예매나 예약 시스템은 ▲ 자리 선점과 동시에 일시적으로 예약 상태로 만드는 방식 ▲ 결제가 완료된 시점에 자리를 확정하는 방식 등, 두 가지 대표적인 처리가 있습니다.
• 예약 과정에서 새로고침을 하면, 트랜잭션이 취소되거나 대기열에서 밀릴 수 있으므로 주의해야 하며, 사용자는 자신의 요청이 어느 시점에 큐에 들어가 있는지 파악하는 것이 유리합니다.

 

🗨️ Kafka 실습

이 실습은 Kafka에 저장된 메시지를 Consumer 프로젝트가 읽어오는 구조를 확인하는 예제다.

Producer가 Kafka Topic에 메시지 저장
↓
Kafka는 Topic/Partition에 메시지 보관
↓
Consumer가 @KafkaListener로 Topic을 구독
↓
메시지가 들어오면 메서드가 자동 실행됨

Kafka 설치

• 도커 컴포즈를 사용하여 kafka 컨테이너를 생성합니다. docker-compose.yml 파일을 생성합니다.

version: '3.8'
services:
  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    platform: linux/amd64
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: wurstmeister/kafka:latest
    platform: linux/amd64
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INSIDE://kafka:29092,OUTSIDE://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
      KAFKA_LISTENERS: INSIDE://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    platform: linux/amd64
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
      KAFKA_CLUSTERS_0_READONLY: "false"

 

+) 설명

# Docker Compose 문법 버전
version: '3.8'

# 실행할 컨테이너 목록
services:

  # Kafka 상태를 관리하는 서버
  # Broker 상태/메타데이터 관리 역할
  zookeeper:

    # 사용할 Docker 이미지
    image: wurstmeister/zookeeper:3.4.6

    # 실행 플랫폼
    # 맥 M칩 등에서 amd64 강제 실행 시 사용
    platform: linux/amd64

    # 포트 연결
    # 내 컴퓨터 2181 ↔ 컨테이너 2181
    ports:
      - "2181:2181"

    # 컨테이너 내부 설정값
    environment:

      # Zookeeper 접속 포트
      ZOOKEEPER_CLIENT_PORT: 2181

      # Zookeeper 내부 시간 설정
      ZOOKEEPER_TICK_TIME: 2000


  # 실제 Kafka 메시지 서버(Broker)
  kafka:

    # Kafka Docker 이미지
    image: wurstmeister/kafka:latest

    # 실행 플랫폼
    platform: linux/amd64

    # Kafka 접속 포트
    # Spring Boot에서 localhost:9092 로 접속 가능
    ports:
      - "9092:9092"

    # Kafka 내부 설정값
    environment:

      # Kafka 접속 주소 설정
      # 외부(localhost:9092)와 내부(kafka:29092) 연결용
      KAFKA_ADVERTISED_LISTENERS: INSIDE://kafka:29092,OUTSIDE://localhost:9092

      # Listener 프로토콜 설정
      # PLAINTEXT = 암호화 없이 통신
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT

      # Kafka가 실제 열어둘 포트
      KAFKA_LISTENERS: INSIDE://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092

      # Broker끼리 통신 시 사용할 Listener
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE

      # Kafka가 연결할 Zookeeper 주소
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181

    # Docker 내부 정보 접근용 설정
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock


  # Kafka 상태를 웹에서 확인하는 UI 서버
  kafka-ui:

    # Kafka UI Docker 이미지
    image: provectuslabs/kafka-ui:latest

    # 실행 플랫폼
    platform: linux/amd64

    # 웹 접속 포트
    # localhost:8080 접속 가능
    ports:
      - "8080:8080"

    # Kafka UI 설정값
    environment:

      # UI에 표시할 Kafka 클러스터 이름
      KAFKA_CLUSTERS_0_NAME: local

      # Kafka Broker 연결 주소
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092

      # 연결할 Zookeeper 주소
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181

      # false = UI에서 수정 가능
      KAFKA_CLUSTERS_0_READONLY: "false"

 

서버(컨테이너)	역할	핵심 기능	접속 포트
Zookeeper	Kafka 관리 서버	Broker 상태 및 메타데이터 관리	2181
Kafka	실제 메시지 서버(Broker)	Topic/Partition에 메시지 저장 및 전달	9092
Kafka UI	Kafka 웹 관리 페이지	Topic, Partition, 메시지 상태 확인	8080

Zookeeper, Kafka, Kafka UI는 서로 역할이 다른 별개의 서버(프로세스)이며, 실습에서는 각각 Docker 컨테이너로 분리 실행하였다. Docker Compose는 여러 컨테이너를 같은 네트워크로 자동 연결해주기 때문에, 컨테이너 이름만으로 서로 통신할 수 있다.

 

• docker-compose.yml 파일이 있는 경로에서 도커 컴포즈를 실행합니다.
  • 버전 관련으로 에러가 발생할 경우 zookeeper의 image를 wurstmeister/zookeeper:latest로 변경

docker compose up -d

• localhost:8080 에 접속하면 kafka UI 에 접속할 수 있습니다.

 

Producer Application

• start.spring.io 에 접속하여 프로젝트를 생성합니다.

프로젝트를 생성하면 build.gradle의 디펜던시는 아래와 같습니다.

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-web'
	implementation 'org.springframework.kafka:spring-kafka'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.kafka:spring-kafka-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

application.properties

# 현재 스프링 애플리케이션 이름
# Kafka Producer 프로젝트 식별용
spring.application.name=producer

# Spring Boot 실행 포트
# localhost:8090 으로 접속 가능
server.port=8090

# Kafka Broker 접속 주소
# localhost:9092 에 실행 중인 Kafka 서버 연결
spring.kafka.bootstrap-servers=localhost:9092

# 메시지 Key 직렬화 클래스
# Kafka는 데이터를 byte 형태로 저장하기 때문에
# String -> byte 변환 과정(직렬화)이 필요함
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer

# 메시지 Value 직렬화 클래스
# 실제 메시지 내용(String)을 byte로 변환
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

ProducerApplicationKafkaConfig.java

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
// Kafka Producer 설정 클래스
// KafkaTemplate Bean 등을 스프링 컨테이너에 등록
public class ProducerApplicationKafkaConfig {

    @Bean
    // Kafka Producer 객체를 생성하는 Factory Bean
    public ProducerFactory<String, String> producerFactory() {

        // Kafka 설정값 저장 Map
        Map<String, Object> configProps = new HashMap<>();

        // Kafka Broker 주소 설정
        // localhost:9092 Kafka 서버 연결
        configProps.put(
                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "localhost:9092"
        );

        // 메시지 Key 직렬화 방식
        // String 데이터를 byte 배열로 변환
        configProps.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class
        );

        // 메시지 Value 직렬화 방식
        // 실제 메시지 데이터 직렬화
        configProps.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class
        );

        // ProducerFactory 생성 반환
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    // Kafka 메시지 전송 객체
    // 실제로 Kafka에 메시지를 보내는 핵심 클래스
    public KafkaTemplate<String, String> kafkaTemplate() {

        // ProducerFactory 기반 KafkaTemplate 생성
        return new KafkaTemplate<>(producerFactory());
    }
}

ProducerController.java

import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
// REST API Controller
@RequiredArgsConstructor
// final 필드 생성자 자동 생성
public class ProducerController {

    // ProducerService 의존성 주입
    private final ProducerService producerService;

    @GetMapping("/send")
    // GET 요청 API
    // ex)
    // /send?topic=test&key=user1&message=hello
    public String sendMessage(

            // URL 파라미터 topic 값 받기
            // ex) topic=test
            @RequestParam("topic") String topic,

            // 메시지 Key 값 받기
            // 같은 Key는 같은 Partition으로 저장됨
            @RequestParam("key") String key,

            // 실제 메시지 내용
            @RequestParam("message") String message
    ) {

        // Kafka 메시지 전송 서비스 호출
        producerService.sendMessage(topic, key, message);

        return "Message sent to Kafka topic";
    }
}

ProducerService.java

import lombok.RequiredArgsConstructor;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
// 비즈니스 로직 처리 클래스
@RequiredArgsConstructor
public class ProducerService {

    // Kafka 메시지 전송 객체
    private final KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(
            String topic,
            String key,
            String message
    ) {

        // 메시지 10개 반복 전송
        for (int i = 0; i < 10; i++) {

            // Kafka 메시지 전송
            //
            // 첫 번째 파라미터:
            // Topic 이름
            //
            // 두 번째 파라미터:
            // 메시지 Key
            // 같은 Key면 같은 Partition 저장 가능
            //
            // 세 번째 파라미터:
            // 실제 메시지 데이터
            kafkaTemplate.send(
                    topic,
                    key,
                    message + " " + i
            );
        }
    }
}

 

개념	의미
Producer	Kafka에 메시지 보내는 역할
KafkaTemplate	Kafka 메시지 전송 객체
Topic	메시지 저장 장소
Key	메시지를 어느 Partition에 넣을지 결정
Serializer	String → byte 변환
bootstrap-servers	Kafka Broker 접속 주소

 

사용자 요청
↓
Controller
↓
ProducerService
↓
KafkaTemplate.send()
↓
Kafka Topic 저장
↓
Partition 분배

 

Consumer Application

프로젝트를 생성하면 build.gradle의 디펜던시는 아래와 같습니다.

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-web'
	implementation 'org.springframework.kafka:spring-kafka'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.kafka:spring-kafka-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

application.properties

spring.application.name=consumer
server.port=8091

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

ConsumerApplicationKafkaConfig.java

• 이 클래스는 Kafka Consumer가 어떻게 메시지를 읽을지 설정하는 클래스다.

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

// 이 클래스는 Kafka 컨슈머 설정을 위한 Spring 설정 클래스입니다.
@EnableKafka // Kafka 리스너를 활성화하는 어노테이션입니다.
@Configuration // Spring 설정 클래스로 선언하는 어노테이션입니다.
public class ConsumerApplicationKafkaConfig {

    // Kafka 컨슈머 팩토리를 생성하는 빈을 정의합니다.
    // ConsumerFactory는 Kafka 컨슈머 인스턴스를 생성하는 데 사용됩니다.
    // 각 컨슈머는 이 팩토리를 통해 생성된 설정을 기반으로 작동합니다.
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        // 컨슈머 팩토리 설정을 위한 맵을 생성합니다.
        Map<String, Object> configProps = new HashMap<>();
        // Kafka 브로커의 주소를 설정합니다.
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 메시지 키의 디시리얼라이저 클래스를 설정합니다.
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // 메시지 값의 디시리얼라이저 클래스를 설정합니다.
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // 설정된 프로퍼티로 DefaultKafkaConsumerFactory를 생성하여 반환합니다.
        return new DefaultKafkaConsumerFactory<>(configProps);
    }

    // Kafka 리스너 컨테이너 팩토리를 생성하는 빈을 정의합니다.
    // ConcurrentKafkaListenerContainerFactory는 Kafka 메시지를 비동기적으로 수신하는 리스너 컨테이너를 생성하는 데 사용됩니다.
    // 이 팩토리는 @KafkaListener 어노테이션이 붙은 메서드들을 실행할 컨테이너를 제공합니다.
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        // ConcurrentKafkaListenerContainerFactory를 생성합니다.
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 컨슈머 팩토리를 리스너 컨테이너 팩토리에 설정합니다.
        factory.setConsumerFactory(consumerFactory());
        // 설정된 리스너 컨테이너 팩토리를 반환합니다.
        return factory;
    }
}

 

+) ConsumerApplicationKafkaConfig

ConsumerApplicationKafkaConfig.java

 

이 클래스는 Kafka Consumer가 어떻게 메시지를 읽을지 설정하는 클래스다.

 

@EnableKafka

 

@KafkaListener를 사용할 수 있게 활성화하는 어노테이션이다.
이게 있어야 Kafka 메시지 리스너가 정상 동작한다.

 

@Configuration

 

이 클래스가 스프링 설정 클래스라는 뜻이다.
안에 있는 @Bean 메서드들이 스프링 컨테이너에 등록된다.

 

public ConsumerFactory<String, String> consumerFactory()

 

Consumer 객체를 만들기 위한 공장이다.
<String, String>은 메시지의 key와 value를 둘 다 String으로 읽겠다는 뜻이다.

 

Kafka Consumer는 그냥 자동으로 생기는 게 아니라, “어떤 Kafka 서버에 연결할지”, “메시지를 어떤 방식으로 읽을지” 같은 설정이 먼저 필요하다. 그래서 먼저 Consumer를 만들기 위한 설정 공장인 ConsumerFactory를 만든다.

 

 

Map<String, Object> configProps = new HashMap<>();

 

Kafka Consumer 설정값을 담는 Map이다.
Kafka가 정해둔 설정 key에 맞춰 값을 넣는다.

 

configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

 

이 메서드는 Kafka Consumer를 생성하기 위한 설정 객체를 만드는 역할이다. 

여기서: Kafka Broker 주소를 설정한다.
실제로는 "bootstrap.servers" = "localhost:9092"를 넣는 것과 비슷하다.

즉, “Kafka 서버는 localhost:9092에 있다”는 설정이고,

 

configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

 

는 Kafka에서 가져온 byte 데이터를 String으로 다시 변환하는 방법을 설정하는 것이다.

메시지 key를 String으로 역직렬화한다.

 

configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

 

메시지 value를 String으로 역직렬화한다.

 

return new DefaultKafkaConsumerFactory<>(configProps);

 

위에서 만든 설정값을 기반으로 ConsumerFactory를 생성한다.

 

즉 ConsumerFactory는:

"Kafka Consumer를 어떤 설정으로 만들지 정의하는 공장"

 

이라고 이해하면 된다.

+) kafkaListenerContainerFactory()

kafkaListenerContainerFactory()

 

Consumer만 만든다고 메시지가 자동으로 들어오는 건 아니다. 실제로 Kafka Topic을 계속 감시하다가 메시지가 들어오 면 @KafkaListener 메서드를 실행해줄 실행 환경이 필요하다. 그 역할을 하는 게:

 

public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory()

 

이다.

이건 @KafkaListener가 붙은 메서드를 실행시키기 위한 리스너 컨테이너 설정이다.

 

쉽게 말하면:

@KafkaListener 메서드들이 Kafka 메시지를 받을 수 있게 도와주는 실행 환경

 

이다.

 

factory.setConsumerFactory(consumerFactory());

 

아까 만든 ConsumerFactory를 연결한다.
즉, "아까 만든 Consumer 설정을 사용해서 @KafkaListener가 실제로 메시지를 읽게 해라" 는 뜻이다.

ConsumerFactory는 Consumer를 만드는 설정이고,
ConcurrentKafkaListenerContainerFactory는 그 Consumer를 이용해서 @KafkaListener를 실제로 돌리는 실행 환경임.

구분	ConsumerFactory<string, string>	ConcurrentKafkaListenerContainerFactory<string, string>
한 줄 정의	Kafka Consumer를 만드는 공장	@KafkaListener를 실행시키는 컨테이너를 만드는 공장
쉽게 말하면	“어떤 설정으로 Consumer를 만들지” 정함	“그 Consumer를 가지고 리스너 메서드를 어떻게 돌릴지” 정함
담당 역할	Kafka 접속 정보, 역직렬화 방식 등 Consumer 기본 설정	@KafkaListener가 메시지를 자동으로 받도록 실행 환경 구성
들어가는 설정	bootstrap.servers, key.deserializer, value.deserializer	ConsumerFactory, 동시 처리 개수, Ack 방식 등
코드 위치	new DefaultKafkaConsumerFactory<>(configProps)	new ConcurrentKafkaListenerContainerFactory<>()
연결 관계	먼저 만들어짐	factory.setConsumerFactory(consumerFactory())로 ConsumerFactory를 사용함
비유	“직원 채용 기준서”	“직원을 배치해서 일하게 하는 작업장”
실제 의미	Kafka 메시지를 읽을 Consumer 객체 생성 준비	@KafkaListener 메서드가 자동 실행되도록 Consumer를 붙여줌
없으면?	Consumer 자체를 만들 설정이 없음	@KafkaListener가 어떤 Consumer로 어떻게 동작할지 모름
핵심 기억	Consumer 설정 담당	Listener 실행 담당

ConsumerService.java

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class ConsumerService {

    // 이 메서드는 Kafka에서 메시지를 소비하는 리스너 메서드입니다.
    // @KafkaListener 어노테이션은 이 메서드를 Kafka 리스너로 설정합니다.
    @KafkaListener(groupId = "group_a", topics = "topic1")
    // Kafka 토픽 "test-topic"에서 메시지를 수신하면 이 메서드가 호출됩니다.
    // groupId는 컨슈머 그룹을 지정하여 동일한 그룹에 속한 다른 컨슈머와 메시지를 분배받습니다.
    public void consumeFromGroupA(String message) {
        log.info("Group A consumed message from topic1: " + message);
    }

    // 동일한 토픽을 다른 그룹 ID로 소비하는 또 다른 리스너 메서드입니다.
    @KafkaListener(groupId = "group_b", topics = "topic1")
    public void consumeFromGroupB(String message) {
        log.info("Group B consumed message from topic1: " + message);
    }

    // 다른 토픽을 다른 그룹 ID로 소비하는 리스너 메서드입니다.
    @KafkaListener(groupId = "group_c", topics = "topic2")
    public void consumeFromTopicC(String message) {
        log.info("Group C consumed message from topic2: " + message);
    }

    // 다른 토픽을 다른 그룹 ID로 소비하는 리스너 메서드입니다.
    @KafkaListener(groupId = "group_c", topics = "topic3")
    public void consumeFromTopicD(String message) {
        log.info("Group C consumed message from topic3: " + message);
    }

    @KafkaListener(groupId = "group_d", topics = "topic4")
    public void consumeFromPartition0(String message) {
        log.info("Group D consumed message from topic4: " + message);
    }
}

1. ConsumerService는 @KafkaListener를 통해 Kafka Topic을 구독한다.
2. 메시지가 Topic에 들어오면 해당 메서드가 자동 실행된다.
3. groupId는 Consumer Group을 의미하며,
4. 같은 Topic이라도 groupId가 다르면 각각 독립적으로 메시지를 읽을 수 있다.

 

확인

• 두 애플리케이션을 실행하고 Kafka UI를 확인합니다. Topic 탭과 Comsumers 탭을 확인합니다.

topic 을 test-topic 으로 지정하고 요청해봅니다.

• kafka ui > Topics > Messages 에 접속하면 방금 요청한 토픽으로 발행된 메시지를 볼 수 있습니다.

topic 을 topic1 으로 지정하고 요청해봅니다.

• 컨슈머 애플리케이션의 로그를 보면 GroupA 와 GroupB가 메시지를 수신한 것을 볼 수 있습니다. 
• 따라서 같은 토픽을 가지고 그룹이 다르면 메시지를 각 그룹마다 수신한다는 것을 알 수 있습니다.

topic 을 topic2 으로 지정하고 요청해봅니다.

• 컨슈머 애플리케이션의 로그를 보면 GroupC이고 topic 이 2 인 리스너가 메시지를 수신한 것을 볼 수 있습니다.

topic 을 topic3 으로 지정하고 요청해봅니다.

• 컨슈머 애플리케이션의 로그를 보면 GroupC이고 topic 이 3 인 리스너가 메시지를 수신한 것을 볼 수 있습니다.

topic 을 topic4 으로 지정하고 요청해봅니다.

• 컨슈머 애플리케이션의 로그를 보면 GroupD이고 topic 이 4 인 리스너가 메시지를 수신한 것을 볼 수 있습니다.

 

Kafka 예제 프로젝트 구조 정리

• 이번 Kafka 예제는 Producer, Consumer 총 2개의 Spring Boot 프로젝트로 구성된다.
• Producer 프로젝트는 Kafka Topic에 메시지를 보내고, Consumer 프로젝트는 Topic의 메시지를 구독해서 처리한다.

프로젝트	역할	주요 코드	Kafka 기준
Producer	메시지를 생성하고 Kafka Topic에 발행	ProducerController, ProducerService, ProducerApplicationKafkaConfig	Producer
Consumer	Kafka Topic에서 메시지를 읽어 처리	ConsumerService, ConsumerApplicationKafkaConfig	Consumer

 

각 프로젝트의 차이

구분	Producer 프로젝트	Consumer 프로젝트
역할	메시지 보내는 쪽	메시지 받는 쪽
실행 포트	8090	8091
Kafka 기준	Producer	Consumer
주요 객체/어노테이션	KafkaTemplate	@KafkaListener
설정 클래스	ProducerFactory, KafkaTemplate 설정	ConsumerFactory, KafkaListenerContainerFactory 설정
하는 일	/send 요청을 받으면 Kafka Topic에 메시지 전송	지정한 Topic을 구독하다가 메시지가 들어오면 자동 처리
메시지 위치	Topic/Partition에 저장됨	Topic/Partition에서 메시지를 읽음
실행 결과	Topic에 메시지가 기록됨	Consumer Group 기준으로 메시지를 소비함

 

Kafka 메시지 흐름

1. 사용자가 Producer 프로젝트에 /send?topic=topic1&key=user1&message=hello 요청을 보낸다.
2. ProducerController가 요청 파라미터를 받는다.
3. ProducerService가 KafkaTemplate.send(topic, key, message)로 Kafka에 메시지를 보낸다.
4. Kafka는 메시지를 지정한 Topic에 저장한다.
5. 메시지는 Key를 기준으로 특정 Partition에 저장될 수 있다.
6. Consumer 프로젝트의 @KafkaListener가 해당 Topic을 구독하고 있다가 메시지를 읽는다.
7. Consumer Group 기준으로 메시지가 분산 처리된다.

ConsumerService 기준 정리

메서드	Consumer Group	Topic	의미
consumeFromGroupA	group_a	topic1	topic1 메시지를 group_a 소비자가 읽음
consumeFromGroupB	group_b	topic1	같은 topic1이지만 다른 그룹이므로 별도로 메시지를 읽음
consumeFromTopicC	group_c	topic2	topic2 메시지를 group_c가 읽음
consumeFromTopicD	group_c	topic3	같은 group_c지만 다른 Topic을 읽음
consumeFromPartition0	group_d	topic4	topic4 메시지를 group_d가 읽음

 

RabbitMQ 예제와 다른 점

구분	RabbitMQ  예제	Kafka 예제
메시지 저장 위치	Queue	Topic / Partition
보내는 객체	RabbitTemplate	KafkaTemplate
받는 어노테이션	@RabbitListener	@KafkaListener
메시지 소비 후	Queue에서 사라지는 경우가 많음	Topic에 설정 기간 동안 남아 있음
소비 기준	Queue 구독	Topic + Consumer Group 구독
핵심 목적	메시지 전달/작업 큐	데이터 스트림 저장/처리

RabbitMQ는 Queue에 메시지를 넣고 Consumer가 가져가면 사라지는 작업 큐 느낌이고,
Kafka는 Topic에 메시지를 기록해두고 Consumer Group이 각자 읽어가는 로그 저장소 느낌이다.

구분	RabbitMQ 예제	Kafka 예제	왜 차이나는지
프로젝트 수	Order, Product, Payment 3개	Producer, Consumer 2개	RabbitMQ 예제는 주문/상품/결제 서비스 분리 실습이고, Kafka 예제는 송신/수신 구조 확인 실습이라 더 단순함
메시지 보내는 쪽	OrderService	ProducerService	둘 다 메시지를 보내는 서비스 로직 담당
메시지 받는 쪽	ProductEndpoint, PaymentEndpoint	ConsumerService	RabbitMQ는 Queue별 Consumer를 서비스처럼 나눴고, Kafka는 여러 Topic/Group 리스너를 한 클래스에 모아둠
전송 객체	RabbitTemplate	KafkaTemplate	RabbitMQ로 보내냐 Kafka로 보내냐에 따라 사용하는 템플릿이 다름
받는 어노테이션	@RabbitListener	@KafkaListener	각 메시징 기술에서 Queue/Topic을 구독하는 방식이 다름
저장 단위	Queue	Topic / Partition	RabbitMQ는 Queue 중심, Kafka는 Topic과 Partition 중심
소비 기준	Queue 이름	Topic + Consumer Group	Kafka는 같은 Topic도 Group이 다르면 각각 따로 읽을 수 있음
설정 코드	Queue, Exchange, Binding 설정	ProducerFactory, ConsumerFactory 설정	RabbitMQ는 메시지를 어느 Queue로 보낼지 연결 구조를 만들고, Kafka는 Producer/Consumer 생성 설정이 중요함
메시지 소비 후	Queue에서 제거되는 흐름	Topic에 남고 Offset으로 읽은 위치 관리	RabbitMQ는 작업 큐 느낌, Kafka는 로그 저장소 느낌
코드 구조 느낌	실제 MSA 서비스 분리 느낌	Kafka 개념 실습 느낌	RabbitMQ 코드는 도메인 서비스처럼 나뉘었고, Kafka 코드는 Topic/Group 동작 확인용에 가까움

 

핵심 정리

• Kafka에서는 Producer가 Topic에 메시지를 저장하고, Consumer는 Topic을 구독하여 메시지를 읽는다.
• 같은 Topic이라도 Consumer Group이 다르면 각각 독립적으로 메시지를 읽을 수 있다.
• 즉, Kafka는 메시지를 “전달하고 끝”이라기보다, Topic에 기록해두고 여러 Consumer Group이 필요에 따라 읽어가는 구조이다.

 

🗨️ 추가 실습 (SAGA Pattern)

 

OrderApplication

• OrderApplication 쪽 Saga 시작 부분. 핵심은 주문 생성 → 상품/결제 큐로 메시지 발행 → 실패 메시지를 받으면 주문 취소로 보상 처리 흐름임.
  • Saga 패턴은 전체 작업을 하나의 트랜잭션으로 묶는 대신, 각 서비스가 자기 작업을 처리하고, 중간에 실패하면 보상 트랜잭션 메시지를 보내 이전 작업을 되돌리는 방식이다.
  • 이 코드에서 Order의 보상 트랜잭션은 cancelOrder()이고, 실패 메시지를 받는 입구는 @RabbitListener(queues = "${message.queue.err.order}")이다.

build.gradle > dependencies

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	implementation 'org.springframework.boot:spring-boot-starter-web'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

application.properties

# 현재 애플리케이션 이름
spring.application.name=order

# 정상 처리용 Exchange
# Order가 Product/Payment 쪽으로 메시지를 보낼 때 사용
message.exchange=market

# 상품 처리 Queue
message.queue.product=market.product

# 결제 처리 Queue
message.queue.payment=market.payment

# 에러 처리용 Exchange
# Product/Payment 처리 중 실패했을 때 보상 트랜잭션 메시지를 보낼 때 사용
message.err.exchange=market.err

# Order가 실패 메시지를 받을 Queue
# 이 Queue로 메시지가 오면 Order는 주문을 취소 처리함
message.queue.err.order=market.err.order

# Product가 실패 메시지를 받을 Queue
# 결제 실패 등으로 상품 재고 복구가 필요할 때 사용할 수 있음
message.queue.err.product=market.err.product

# RabbitMQ 서버 접속 정보
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

 

DeliveryMessage.java

• DeliveryMessage는 서비스 사이에서 주고받는 메시지 DTO다.
  • Order, Product, Payment가 같은 주문 정보를 공유하기 위해 사용한다.

import lombok.*;

import java.util.UUID;

@Data
// getter, setter, toString, equals 등을 자동 생성
@Builder
// DeliveryMessage.builder() 형태로 객체 생성 가능
@ToString
@AllArgsConstructor
// 모든 필드를 받는 생성자 생성
@NoArgsConstructor
// 기본 생성자 생성
// RabbitMQ가 JSON 메시지를 객체로 변환할 때 기본 생성자가 필요할 수 있음
public class DeliveryMessage {

    // 주문 ID
    // 여러 서비스가 같은 주문을 추적하기 위한 핵심 값
    private UUID orderId;

    // 결제 ID
    // 결제 서비스에서 결제 성공 후 생성될 수 있는 값
    private UUID paymentId;

    // 주문한 사용자 ID
    private String userId;

    // 상품 ID
    private Integer productId;

    // 주문 상품 수량
    private Integer productQuantity;

    // 결제 금액
    private Integer payAmount;

    // 실패 원인
    // Saga 보상 처리에서 어떤 이유로 실패했는지 전달할 때 사용
    private String errorType;
}

Order.java

• cancelOrder()는 Saga에서 보상 트랜잭션 역할을 한다.
  • 결제나 상품 처리 실패 시 이미 생성된 주문을 취소 상태로 되돌린다.

import lombok.Builder;
import lombok.Data;
import lombok.ToString;

import java.util.UUID;

@Builder
// Order.builder()로 객체 생성 가능
@Data
// getter, setter 자동 생성
@ToString
public class Order {

    // 주문 ID
    private UUID orderId;

    // 주문한 사용자 ID
    private String userId;

    // 주문 상태
    // 예: RECEIPT, CANCEL
    private String orderStatus;

    // 주문 실패 원인
    private String errorType;

    public void cancelOrder(String receiveErrorType) {

        // Saga 보상 트랜잭션
        // 뒤 단계에서 실패가 발생하면 주문 상태를 취소로 변경
        orderStatus = "CANCEL";

        // 왜 취소됐는지 실패 원인 저장
        errorType = receiveErrorType;
    }
}

• Order는 현재 주문 상태를 관리하는 주문 도메인 객체다. 즉, “이 주문이 지금 접수 상태인지(RECEIPT), 취소 상태인지(CANCEL)“ 같은 실제 주문 데이터를 저장한다. 그래서 orderStore.put(...)처럼 주문 저장소(Map/DB)에 저장되며, 주문 조회나 상태 변경에 사용된다.
• 반면 DeliveryMessage는 서비스끼리 RabbitMQ로 주고받기 위한 메시지 DTO다. Product 서비스나 Payment 서비스가 주문 처리에 필요한 정보만 담아서 전달하는 용도로 사용된다.

OrderApplicationQueueConfig.java

• 정상 흐름용 Exchange/Queue와 에러 보상 흐름용 Exchange/Queue를 따로 둔다.
  • 이게 Saga 패턴에서 “실패 시 보상 트랜잭션 메시지”를 보내기 위한 구조다.

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
// RabbitMQ Queue, Exchange, Binding 설정 클래스
public class OrderApplicationQueueConfig {

    @Bean
    public Jackson2JsonMessageConverter producerJackson2MessageConverter() {

        // Java 객체를 RabbitMQ 메시지로 보낼 때 JSON으로 변환해주는 Converter
        // DeliveryMessage 객체를 그대로 보내기 위해 필요
        return new Jackson2JsonMessageConverter();
    }

    @Value("${message.exchange}")
    private String exchange;

    @Value("${message.queue.product}")
    private String queueProduct;

    @Value("${message.queue.payment}")
    private String queuePayment;

    @Value("${message.err.exchange}")
    private String exchangeErr;

    @Value("${message.queue.err.order}")
    private String queueErrOrder;

    @Value("${message.queue.err.product}")
    private String queueErrProduct;

    @Bean
    public TopicExchange exchange() {

        // 정상 메시지를 보낼 Exchange
        // Product, Payment Queue로 메시지를 라우팅함
        return new TopicExchange(exchange);
    }

    @Bean
    public Queue queueProduct() {

        // 상품 처리용 Queue
        return new Queue(queueProduct);
    }

    @Bean
    public Queue queuePayment() {

        // 결제 처리용 Queue
        return new Queue(queuePayment);
    }

    @Bean
    public Binding bindingProduct() {

        // 정상 Exchange와 상품 Queue를 연결
        // routing key가 queueProduct 값이면 market.product Queue로 메시지 전달
        return BindingBuilder
                .bind(queueProduct())
                .to(exchange())
                .with(queueProduct);
    }

    @Bean
    public Binding bindingPayment() {

        // 정상 Exchange와 결제 Queue를 연결
        // routing key가 queuePayment 값이면 market.payment Queue로 메시지 전달
        return BindingBuilder
                .bind(queuePayment())
                .to(exchange())
                .with(queuePayment);
    }

    @Bean
    public TopicExchange exchangeErr() {

        // 에러/보상 처리용 Exchange
        // 실패가 발생했을 때 보상 트랜잭션 메시지를 라우팅함
        return new TopicExchange(exchangeErr);
    }

    @Bean
    public Queue queueErrOrder() {

        // Order 보상 처리 Queue
        // 이 Queue로 메시지가 오면 주문 취소 처리
        return new Queue(queueErrOrder);
    }

    @Bean
    public Queue queueErrProduct() {

        // Product 보상 처리 Queue
        // 재고 복구 같은 처리를 할 수 있음
        return new Queue(queueErrProduct);
    }

    @Bean
    public Binding bindingErrOrder() {

        // 에러 Exchange와 Order 에러 Queue 연결
        return BindingBuilder
                .bind(queueErrOrder())
                .to(exchangeErr())
                .with(queueErrOrder);
    }

    @Bean
    public Binding bindingErrProduct() {

        // 에러 Exchange와 Product 에러 Queue 연결
        return BindingBuilder
                .bind(queueErrProduct())
                .to(exchangeErr())
                .with(queueErrProduct);
    }
}

OrderEndpoint.java

• OrderEndpoint는 HTTP 요청도 받고, RabbitMQ 에러 메시지도 받는다.
  • @PostMapping("/order")는 주문 시작점이고, @RabbitListener는 실패 시 보상 처리 시작점이다.

import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.UUID;

@Slf4j
// log.info() 사용 가능
@RestController
// HTTP 요청을 받는 Controller
@RequiredArgsConstructor
// final 필드 생성자 자동 생성
public class OrderEndpoint {

    private final OrderService orderService;

    private final RabbitTemplate rabbitTemplate;

    @GetMapping("order/{orderId}")
    public ResponseEntity<Order> getOrder(@PathVariable UUID orderId) {

        // 주문 ID로 주문 조회
        Order order = orderService.getOrder(orderId);

        // 조회한 주문을 HTTP 200 응답으로 반환
        return ResponseEntity.ok(order);
    }

    @PostMapping("/order")
    public ResponseEntity<Order> order(@RequestBody OrderRequestDto orderRequestDto) {

        // 주문 생성 요청
        // 내부에서 주문을 만들고 Product/Payment 쪽으로 메시지를 보낼 수 있음
        Order order = orderService.createOrder(orderRequestDto);

        return ResponseEntity.ok(order);
    }

    @RabbitListener(queues = "${message.queue.err.order}")
    public void errOrder(DeliveryMessage message) {

        // Order 에러 Queue를 감시하다가 메시지가 오면 자동 실행
        // 즉 다른 서비스에서 실패 메시지를 보내면 여기서 받음
        log.info("ERROR RECEIVE !!!");

        // Saga 보상 처리
        // 실패 메시지를 기반으로 주문 취소 처리
        orderService.rollbackOrder(message);
    }

    @Data
    public static class OrderRequestDto {

        // 주문 요청 사용자 ID
        private String userId;

        // 주문 상품 ID
        private Integer productId;

        // 주문 수량
        private Integer productQuantity;

        // 결제 금액
        private Integer payAmount;

        public Order toOrder() {

            // 요청 DTO를 Order 도메인 객체로 변환
            return Order.builder()

                    // 주문 생성 시 새로운 주문 ID 발급
                    .orderId(UUID.randomUUID())

                    // 요청한 사용자 ID 저장
                    .userId(userId)

                    // 최초 주문 상태
                    // 아직 최종 완료가 아니라 접수 상태
                    .orderStatus("RECEIPT")
                    .build();
        }

        public DeliveryMessage toDeliveryMessage(UUID orderId) {

            // 다른 서비스로 보낼 메시지 객체 생성
            return DeliveryMessage.builder()

                    // 생성된 주문 ID를 메시지에 포함
                    .orderId(orderId)

                    // 상품 처리에 필요한 정보
                    .productId(productId)
                    .productQuantity(productQuantity)

                    // 결제 처리에 필요한 정보
                    .payAmount(payAmount)
                    .build();
        }
    }
}

OrderService

• OrderService는 주문을 생성한 뒤 RabbitMQ로 메시지를 발행하고, 이후 다른 서비스에서 실패 메시지가 오면 rollbackOrder()로 주문을 취소하는 Saga 보상 처리를 수행한다.

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@Slf4j
// log.info() 사용 가능
@Service
// 주문 비즈니스 로직 처리 클래스
@RequiredArgsConstructor
// final 필드 생성자 자동 생성
public class OrderService {

    @Value("${message.queue.product}")
    // application.properties의 Queue 이름 주입
    // market.product 값이 들어감
    private String productQueue;

    // RabbitMQ 메시지 전송 객체
    // Queue로 메시지를 보내는 핵심 객체
    private final RabbitTemplate rabbitTemplate;

    // 가짜 메모리 DB
    // 실제 프로젝트라면 JPA Repository/DB 사용
    private Map<UUID, Order> orderStore = new HashMap<>();

    public Order createOrder(OrderEndpoint.OrderRequestDto orderRequestDto) {

        // 요청 DTO를 Order 객체로 변환
        // 주문 상태는 RECEIPT 상태로 생성됨
        Order order = orderRequestDto.toOrder();

        // 다른 서비스(Product/Payment)로 보낼 메시지 생성
        DeliveryMessage deliveryMessage =
                orderRequestDto.toDeliveryMessage(order.getOrderId());

        // 주문 저장
        // 현재는 메모리 Map에 저장
        orderStore.put(order.getOrderId(), order);

        // 전송할 메시지 로그 출력
        log.info("send Message : {}",deliveryMessage.toString());

        // RabbitMQ Queue로 메시지 전송
        //
        // productQueue:
        // 어느 Queue로 보낼지
        //
        // deliveryMessage:
        // 실제 전송 데이터
        //
        // 즉:
        // "상품 서비스야 이 주문 처리해줘"
        // 메시지를 보내는 것
        rabbitTemplate.convertAndSend(
                productQueue,
                deliveryMessage
        );

        // 생성된 주문 반환
        return order;
    }

    public void rollbackOrder(DeliveryMessage message) {

        // 실패 메시지 안의 orderId로 주문 조회
        Order order = orderStore.get(message.getOrderId());

        // Saga 보상 트랜잭션
        // 주문 상태를 CANCEL로 변경
        order.cancelOrder(message.getErrorType());

        // 취소 결과 로그 출력
        log.info(order.toString());
    }

    public Order getOrder(UUID orderId) {

        // 주문 조회
        return orderStore.get(orderId);
    }
}

• 전체 흐름
  1. 사용자가 /order 로 주문 요청
  2. OrderService가 주문 생성
  3. DeliveryMessage를 만들어 Product/Payment Queue로 전송
  4. Product나 Payment에서 실패 발생
  5. 실패한 서비스가 에러 Queue로 메시지 전송
  6. OrderEndpoint의 @RabbitListener가 실패 메시지 수신
  7. orderService.rollbackOrder(message) 실행
  8. 주문 상태 CANCEL 처리 

Order 생성
↓
Order 상태 저장
↓
DeliveryMessage 생성
↓
RabbitMQ로 Product/Payment에 전달
↓
실패 시 DeliveryMessage로 에러 전달
↓
Order 상태 CANCEL 변경

 

ProductApplication

build.gradle > dependencies

dependencies {
	// Jackson 의존성
	implementation 'com.fasterxml.jackson.core:jackson-databind'
	implementation 'com.fasterxml.jackson.core:jackson-core'
	implementation 'com.fasterxml.jackson.core:jackson-annotations'

	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

application.properties

spring.application.name=product

message.exchange=market
message.queue.product=market.product
message.queue.payment=market.payment

message.err.exchange=market.err
message.queue.err.order=market.err.order
message.queue.err.product=market.err.product


spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

DeliveryMessage.java

import lombok.*;

import java.util.UUID;

@Data
// getter, setter, toString, equals, hashCode 자동 생성
@Builder
// DeliveryMessage.builder() 형태로 객체 생성 가능
@ToString
// 객체 내용을 로그로 보기 좋게 출력
@AllArgsConstructor
// 모든 필드를 받는 생성자 생성
@NoArgsConstructor
// 기본 생성자 생성
// RabbitMQ가 JSON 메시지를 DeliveryMessage 객체로 바꿀 때 필요할 수 있음
public class DeliveryMessage {

    // 주문 ID
    // Order, Product, Payment가 같은 주문을 추적하기 위한 값
    private UUID orderId;

    // 결제 ID
    // Payment에서 결제 생성 시 사용할 수 있는 값
    private UUID paymentId;

    // 주문한 사용자 ID
    private String userId;

    // 상품 ID
    // Product 서비스가 어떤 상품을 차감할지 판단할 때 사용
    private Integer productId;

    // 주문 수량
    // Product 서비스가 재고 차감 가능 여부를 판단할 때 사용
    private Integer productQuantity;

    // 결제 금액
    // Payment 서비스가 결제 가능 여부를 판단할 때 사용
    private Integer payAmount;

    // 실패 원인
    // Saga 롤백 과정에서 어떤 이유로 실패했는지 전달
    private String errorType;
}

ProductApplicationQueueConfig.java

• DeliveryMessage 같은 객체를 RabbitMQ 메시지로 주고받으려면 JSON 변환기가 필요하다.

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
// ProductApplication의 RabbitMQ 설정 클래스
public class ProductApplicationQueueConfig {

    @Bean
    public Jackson2JsonMessageConverter producerJackson2MessageConverter() {

        // Java 객체를 JSON 메시지로 변환하거나,
        // JSON 메시지를 Java 객체로 변환하는 Converter
        //
        // DeliveryMessage 객체를 RabbitMQ로 주고받기 위해 필요
        return new Jackson2JsonMessageConverter();
    }
}

ProductEndpoint.java

• ProductEndpoint는 정상 메시지와 에러 메시지를 둘 다 받는다.
  • 정상 메시지는 재고 차감 처리, 에러 메시지는 재고 복구 같은 보상 처리로 이어진다.

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Slf4j
// log.info(), log.error() 사용 가능
@Component
// RabbitMQ 메시지를 받는 Bean으로 등록
@RequiredArgsConstructor
// final 필드 생성자 자동 생성
public class ProductEndpoint {

    // 상품 처리 로직 담당 Service
    private final ProductService productService;

    @RabbitListener(queues = "${message.queue.product}")
    // market.product Queue를 구독
    // Order 서비스가 productQueue로 메시지를 보내면 이 메서드가 자동 실행됨
    public void receiveMessage(DeliveryMessage deliveryMessage) {

        // 상품 재고 차감 로직 실행
        // 성공하면 Payment Queue로 메시지를 넘기고,
        // 실패하면 rollbackProduct()로 보상 처리 시작
        productService.reduceProductAmount(deliveryMessage);

        // Product 서비스가 메시지를 받았다는 로그
        log.info("PRODUCT RECEIVE:{}", deliveryMessage.toString());
    }

    @RabbitListener(queues="${message.queue.err.product}")
    // market.err.product Queue를 구독
    // Payment 서비스에서 결제 실패가 발생하면 Product에게 롤백 메시지를 보냄
    public void receiveErrorMessage(DeliveryMessage deliveryMessage) {

        // 에러 메시지를 받았다는 로그
        log.info("ERROR RECEIVE !!!");

        // Product 보상 트랜잭션 실행
        // 예: 이미 차감한 재고를 복구하는 역할
        productService.rollbackProduct(deliveryMessage);
    }
}

ProductService.java

• ProductService는 상품 처리 성공 시 Payment Queue로 메시지를 넘기고, 상품 처리 실패 또는 결제 실패 롤백 시 Order Error Queue로 메시지를 보내 주문 취소를 유도한다.

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

@Slf4j
// 로그 사용
@Service
// 상품 비즈니스 로직 담당 클래스
@RequiredArgsConstructor
// final 필드 생성자 주입
public class ProductService {

    // RabbitMQ 메시지 전송 객체
    private final RabbitTemplate rabbitTemplate;

    @Value("${message.queue.payment}")
    // 결제 처리 Queue 이름 주입
    // Product 처리가 성공하면 Payment Queue로 메시지를 보냄
    private String paymentQueue;

    @Value("${message.queue.err.order}")
    // Order 에러 Queue 이름 주입
    // Product 처리 실패 또는 롤백 발생 시 Order에게 실패 메시지를 보냄
    private String orderErrorQueue;

    public void reduceProductAmount(DeliveryMessage deliveryMessage) {

        // 메시지에서 상품 ID 꺼내기
        Integer productId = deliveryMessage.getProductId();

        // 메시지에서 주문 수량 꺼내기
        Integer productQuantity = deliveryMessage.getProductQuantity();

        // 실습용 실패 조건
        // productId가 1이 아니거나, 수량이 1보다 크면 실패 처리
        //
        // 실제 프로젝트라면:
        // 상품 존재 여부 확인
        // 재고 수량 확인
        // 재고 차감
        // 같은 로직이 들어감
        if (productId != 1 || productQuantity > 1) {

            // Product 처리 실패 시 보상 처리
            // Order 쪽에 실패 메시지를 보내 주문 취소로 이어지게 함
            this.rollbackProduct(deliveryMessage);
            return;
        }

        // Product 처리 성공
        // 다음 단계인 Payment Queue로 메시지 전달
        //
        // 즉:
        // "재고 처리는 성공했으니 이제 결제 처리해줘"
        rabbitTemplate.convertAndSend(paymentQueue, deliveryMessage);
    }

    public void rollbackProduct(DeliveryMessage deliveryMessage){

        // Product 보상 트랜잭션 로그
        log.info("PRODUCT ROLLBACK!!!");

        // errorType이 비어 있으면 Product 실패로 설정
        //
        // Payment 실패로 넘어온 메시지는 이미 errorType이 있을 수 있음
        // 그 경우 기존 실패 원인을 유지해야 함
        if(!StringUtils.hasText(deliveryMessage.getErrorType())){

            // 실패 원인 저장
            deliveryMessage.setErrorType("PRODUCT ERROR");
        }

        // Order 에러 Queue로 실패 메시지 전송
        //
        // Order 서비스는 이 메시지를 받고
        // 주문 상태를 CANCEL로 변경함
        rabbitTemplate.convertAndSend(orderErrorQueue, deliveryMessage);
    }
}

PaymentApplication

build.gradle > dependencies 

dependencies {
	// Jackson 의존성
	implementation 'com.fasterxml.jackson.core:jackson-databind'
	implementation 'com.fasterxml.jackson.core:jackson-core'
	implementation 'com.fasterxml.jackson.core:jackson-annotations'

	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

application.properties

spring.application.name=payment

message.exchange=market
message.queue.product=market.product
message.queue.payment=market.payment

message.err.exchange=market.err
message.queue.err.order=market.err.order
message.queue.err.product=market.err.product


spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

DeliveryMessage.java

import lombok.*;

import java.util.UUID;

@Data
// getter, setter 자동 생성
@Builder
// builder 패턴으로 객체 생성 가능
@ToString
// 로그 출력용 문자열 자동 생성
@AllArgsConstructor
// 전체 필드 생성자
@NoArgsConstructor
// 기본 생성자
// JSON 메시지를 객체로 변환할 때 필요할 수 있음
public class DeliveryMessage {

    // 주문 ID
    private UUID orderId;

    // 결제 ID
    private UUID paymentId;

    // 사용자 ID
    private String userId;

    // 상품 ID
    private Integer productId;

    // 상품 수량
    private Integer productQuantity;

    // 결제 금액
    private Integer payAmount;

    // 실패 원인
    private String errorType;
}

Payment.java

• Payment는 결제 상태를 나타내는 객체이고, DeliveryMessage는 서비스 간 전달용 메시지 객체다.

import lombok.Builder;
import lombok.Data;

import java.util.UUID;

@Data
// getter, setter 자동 생성
@Builder
// Payment.builder()로 결제 객체 생성 가능
public class Payment {

    // 결제 ID
    private UUID paymentId;

    // 결제 사용자 ID
    private String userId;

    // 결제 금액
    // 여기서는 String으로 되어 있지만,
    // 실제 프로젝트라면 Integer, Long, BigDecimal 등을 쓰는 게 더 자연스러움
    private String payAmount;

    // 결제 상태
    // 예: SUCCESS, CANCEL
    private String payStatus;
}

PaymentApplicationQueueConfig.java

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
// PaymentApplication의 RabbitMQ 설정 클래스
public class PaymentApplicationQueueConfig {

    @Bean
    public Jackson2JsonMessageConverter producerJackson2MessageConverter() {

        // Java 객체와 JSON 메시지를 서로 변환해주는 Converter
        //
        // Payment 서비스도 DeliveryMessage 객체를 RabbitMQ로 받기 때문에 필요
        return new Jackson2JsonMessageConverter();
    }
}

PaymentEndpoint.java

• PaymentEndpoint는 Payment Queue를 감시하다가 메시지가 들어오면 결제 처리를 시작한다.

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Slf4j
// 로그 사용
@Component
// RabbitMQ 메시지를 받는 컴포넌트로 등록
@RequiredArgsConstructor
// final 필드 생성자 주입
public class PaymentEndpoint {

    // 결제 처리 로직 담당 Service
    private final PaymentService paymentService;

    @RabbitListener(queues = "${message.queue.payment}")
    // market.payment Queue를 구독
    // Product 서비스가 결제 Queue로 메시지를 보내면 이 메서드가 자동 실행됨
    public void receiveMessage(DeliveryMessage deliveryMessage) {

        // Payment Queue에서 받은 메시지 로그
        log.info("PAYMENT RECEIVE : {}", deliveryMessage.toString());

        // 결제 생성/처리 로직 실행
        // 결제 금액 조건에 따라 성공 또는 롤백 처리
        paymentService.createPayment(deliveryMessage);
    }
}

PaymentService.java

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.UUID;

@Slf4j
// 로그 사용
@Service
// 결제 비즈니스 로직 담당 클래스
@RequiredArgsConstructor
// final 필드 생성자 주입
public class PaymentService {

    // RabbitMQ 메시지 전송 객체
    private final RabbitTemplate rabbitTemplate;

    @Value("${message.queue.err.product}")
    // Product 에러 Queue 이름 주입
    // 결제 실패 시 Product에게 롤백하라고 메시지를 보냄
    private String productErrorQueue;

    public void createPayment(DeliveryMessage deliveryMessage) {

        // 결제 객체 생성
        // 실습에서는 실제 DB 저장은 하지 않고 객체만 생성
        Payment payment = Payment.builder()

                // 새로운 결제 ID 생성
                .paymentId(UUID.randomUUID())

                // 메시지에서 사용자 ID 가져오기
                .userId(deliveryMessage.getUserId())

                // 일단 결제 성공 상태로 생성
                .payStatus("SUCCESS")
                .build();

        // 메시지에서 결제 금액 꺼내기
        Integer payAmount = deliveryMessage.getPayAmount();

        // 실습용 결제 실패 조건
        // 결제 금액이 10000 이상이면 실패 처리
        //
        // 실제 프로젝트라면:
        // 카드 승인 실패
        // 잔액 부족
        // 결제 한도 초과
        // 같은 조건이 들어갈 수 있음
        if (payAmount >= 10000) {

            // 결제 실패 로그
            log.error("Payment amount exceeds limit: {}", payAmount);

            // 결제 상태 취소로 변경
            payment.setPayStatus("CANCEL");

            // 실패 원인 설정
            deliveryMessage.setErrorType("PAYMENT_LIMIT_EXCEEDED");

            // Payment 보상 처리 시작
            // Product 에러 Queue로 메시지를 보내서
            // Product가 재고 롤백을 하게 만듦
            this.rollbackPayment(deliveryMessage);
        }
    }

    public void rollbackPayment(DeliveryMessage deliveryMessage) {

        // Payment 롤백 로그
        log.info("PAYMENT ROLLBACK !!!");

        // Product 에러 Queue로 실패 메시지 전송
        //
        // Product는 이 메시지를 받고 rollbackProduct() 실행
        // 이후 Order 에러 Queue로 다시 메시지를 보내 주문 취소까지 이어짐
        rabbitTemplate.convertAndSend(productErrorQueue, deliveryMessage);
    }
}

 

Saga 패턴 예제 프로젝트 구조 정리

• 이번 예제는 Order → Product → Payment 순서로 작업을 진행하고, 중간에 실패하면 이전 단계로 에러 메시지를 보내 보상 트랜잭션(롤백)을 수행하는 Saga 패턴 구조이다.

프로젝트	역할	주요 코드	정상 흐름	실패 시 역할
OrderApplication	주문 생성 및 Saga 시작	OrderEndpoint, OrderService	주문 생성 후 Product Queue로 메시지 발행	Product 실패 메시지를 받으면 주문 취소(CANCEL)
ProductApplication	상품/재고 처리	ProductEndpoint, ProductService	재고 처리 성공 시 Payment Queue로 메시지 발행	실패 시 Order Error Queue로 메시지 발행
PaymentApplication	결제 처리	PaymentEndpoint, PaymentService	결제 성공 처리	실패 시 Product Error Queue로 메시지 발행

 

Queue 흐름 정리

Queue	의미	사용하는 서비스
market.product	상품 처리 요청 Queue	Order → Product
market.payment	결제 처리 요청 Queue	Product → Payment
market.err.product	Product 롤백 요청 Queue	Payment → Product
market.err.order	Order 롤백 요청 Queue	Product → Order

 

전체 Saga 흐름

Order 생성
↓
Order → Product Queue 메시지 발행
↓
Product 재고 처리
↓
성공 시 Payment Queue로 메시지 전송
↓
Payment 결제 처리
↓
결제 실패 발생
↓
Payment → Product Error Queue 메시지 전송
↓
Product 롤백 실행
↓
Product → Order Error Queue 메시지 전송
↓
Order 롤백 실행
↓
주문 상태 CANCEL

 

Saga 패턴 핵심 정리

개념	의미
Saga Pattern	각 서비스가 자기 작업을 처리하고 실패 시 이전 단계로 보상 메시지를 보내 상태를 되돌리는 방식
보상 트랜잭션	실패 시 이전 작업을 취소/복구하는 처리
메시지 기반 처리	서비스끼리 직접 호출하지 않고 RabbitMQ 메시지로 통신
롤백 흐름	Payment 실패 → Product 롤백 → Order 롤백
핵심 특징	전체를 하나의 DB 트랜잭션으로 묶지 않음

• Saga 패턴은 하나의 큰 트랜잭션으로 처리하는 대신, 각 서비스가 자기 작업을 처리하고, 실패 시 이전 단계로 롤백 메시지를 보내 상태를 보상 처리하는 방식이다.
• 이 예제에서는 Payment 실패 시 Product가 롤백되고, 이후 Order까지 롤백되어 최종적으로 주문 상태가 CANCEL 된다.

 

🗨️ 프로덕트 에러 확인해보기

• 이번 실습은 RabbitMQ를 사용해 분산 시스템에서 발생한 에러를 메시지 큐로 전달하고, 각 서비스가 이를 받아 롤백 처리하는 흐름을 확인하는 내용이다.
• 주문(Order), 상품(Product), 결제(Payment) 애플리케이션이 서로 연동되는 과정에서 에러가 발생하면, 해당 에러를 큐를 통해 관련 서비스에 알리고, 각 서비스는 필요한 보상 작업을 수행한다. 핵심은 에러도 메시지로 전달하고, 각 서비스가 그 메시지를 받아 데이터 일관성을 맞추는 것이다.

1. 메시지 큐를 통한 초기 에러 처리 검증

• 먼저 애플리케이션 코드를 수정한 뒤 재실행하지 않아 원하는 결과가 나오지 않는 문제를 확인하고, 다시 실행하여 테스트를 진행하였다.
• 정상적인 상품 수량을 입력한 뒤 결제 금액을 크게 설정하면, Payment 애플리케이션에서 결제 한도 초과 에러가 발생한다. 이때 에러 정보는 큐를 통해 전달되고, 최종적으로 Order Error Queue까지 도착하는 것을 확인할 수 있다.
• 반대로 결제 금액은 정상으로 두고 상품 수량을 잘못 입력하면, 이번에는 Product 애플리케이션에서 상품 수량 관련 에러가 발생한다. 이 경우에도 Product에서 발생한 에러 메시지가 큐를 통해 Order 쪽으로 전달되어야 한다.
• 즉 이 단계의 핵심은 Payment 에러와 Product 에러가 각각 Queue를 통해 정상적으로 전달되는지 검증하는 것이다.

2. 오더 서비스의 에러 큐 리스너 구현

• Order 서비스가 에러 메시지를 받을 수 있도록 새로운 리스너를 추가한다.

@RabbitListener(queues = "${message.queue.err.order}")

• 이 리스너는 Order Error Queue를 감시하다가, 에러 메시지가 들어오면 자동으로 메서드를 실행한다.
• 수신된 메시지는 errOrder() 같은 에러 처리 메서드로 전달된다.

public void errOrder(DeliveryMessage message)

• 이 구조를 통해 Product나 Payment에서 어떤 에러가 발생하더라도, 최종적으로 Order 서비스가 에러 메시지를 받아 대응할 수 있게 된다.

3. 에러 처리 준비와 rollbackOrder() 생성

• Order 서비스가 에러 메시지를 받았는지 확인하기 위해 로그를 남긴다.

log.info("ERROR RECEIVE !!!");

• 이 로그는 Order가 에러 메시지를 정상적으로 수신했는지 확인하는 용도다.
• 이후 실제 롤백 처리를 위해 rollbackOrder() 메서드를 만든다.

orderService.rollbackOrder(message);

• 즉 흐름은 다음과 같다.

Order Error Queue에 에러 메시지 도착
↓
@RabbitListener가 메시지 수신
↓
ERROR RECEIVE 로그 출력
↓
rollbackOrder(message) 실행

 

4. order error 큐 리스닝과 rollback 처리

• rollbackOrder()는 DeliveryMessage를 받아 롤백을 시작한다.
• 이때 DeliveryMessage 안에는 롤백 대상이 되는 orderId가 들어 있다.

message.getOrderId()

• Order 서비스는 이 orderId를 이용해 메모리에 저장해둔 주문 정보를 찾는다.

Order order = orderStore.get(message.getOrderId());

• 여기서 orderStore는 실습용 메모리 저장소다. 실제 프로젝트라면 DB에서 주문을 조회하는 로직으로 바뀔 수 있다.
• 핵심은 에러 메시지 안의 orderId를 기준으로, 어떤 주문을 롤백해야 하는지 찾는 것이다.

5. 에러 메시지 기반 주문 롤백 구현

• 조회한 주문은 취소 상태로 변경된다.

order.cancelOrder(message.getErrorType());

• cancelOrder()는 주문 상태를 CANCEL로 바꾸고, 에러 원인도 함께 저장한다.

public void cancelOrder(String receiveErrorType) {
    orderStatus = "CANCEL";
    errorType = receiveErrorType;
}

• 예를 들어 Payment에서 결제 한도 초과가 발생했다면:

PAYMENT_LIMIT_EXCEEDED

• 같은 에러 타입이 Order에 기록된다.
• 또한 실습 중 Order 애플리케이션을 재실행하면 메모리 저장소인 orderStore가 초기화될 수 있다. 이 경우 기존 Queue에 남아 있던 에러 메시지를 처리하려 해도, 메모리에 주문 정보가 없어서 정상 처리가 어렵다. 그래서 테스트 전에는 RabbitMQ Queue를 purge하여 기존 메시지를 삭제한 뒤 다시 테스트하는 것이 필요하다.

6. 정상/에러 처리 테스트

• Queue를 비운 뒤 Order 애플리케이션을 재실행하고 정상 케이스를 입력하면, 주문 데이터가 정상 저장되고 errorType은 비어 있는 상태로 유지된다.
• 상품 수량을 잘못 입력하면 Product에서 에러가 발생하고, 이 에러 메시지가 Order까지 전달되어 주문 상태가 CANCEL로 변경된다.
• 정상 수량으로 변경한 뒤 결제 금액을 크게 입력하면 Payment에서 결제 한도 초과 에러가 발생한다. 이 경우에도 에러 메시지가 Product를 거쳐 Order까지 전달되고, Order 객체가 수정되어 최종적으로 취소 상태가 된다.
• 정리하면 테스트 흐름은 다음과 같다.

정상 요청
→ Order 정상 저장

상품 수량 오류
→ Product Error
→ Order CANCEL

결제 금액 초과
→ Payment Error
→ Product Rollback
→ Order CANCEL

 

7. 전체 개발 순서와 구현 흐름

• 이번 실습은 한 번에 모든 코드를 작성한 것이 아니라, Queue 라우팅 흐름을 기준으로 단계별로 구현하였다.
• 먼저 Order에서 사용자의 요청을 받는 엔드포인트를 만들고, 주문 정보를 내부 메모리 저장소인 orderStore에 저장하였다. 이후 RabbitMQ를 통해 Product Queue로 메시지를 전달하고, Product가 이를 받아 처리하도록 구현하였다.
• Product가 정상 처리되면 Payment Queue로 메시지를 보내고, Payment 애플리케이션이 이를 수신해 결제 처리를 한다.
• 이후 Payment 단계에서 에러를 발생시키면 market.err.product Queue로 메시지가 이동하고, Product가 이 에러를 받아 롤백 처리한 뒤 다시 market.err.order Queue로 메시지를 전달한다.
• 마지막으로 Order 애플리케이션은 market.err.order Queue를 바라보고 있다가 에러 메시지를 수신하면, orderStore의 주문 정보를 업데이트하여 주문 상태를 CANCEL로 변경한다.

최종 정리

• 이번 실습의 핵심은 정상 처리 흐름뿐 아니라 에러 처리 흐름도 Queue를 통해 설계한다는 것이다.

Order
→ Product
→ Payment

• 정상 흐름이 진행되다가 Payment나 Product에서 실패가 발생하면:

Payment Error
→ Product Rollback
→ Order Rollback

또는

Product Error
→ Order Rollback

• 흐름으로 보상 트랜잭션이 수행된다.
• 현재 실습에서는 DB 저장 기능은 아직 반영되지 않았고, orderStore라는 메모리 저장소를 사용한다. 실제 서비스에 가깝게 만들려면 DB 연결을 추가하여 주문 상태와 에러 정보를 영구 저장하도록 확장할 수 있다.
• 가장 중요한 것은 코드보다 먼저 Queue 라우팅 흐름을 이해하는 것이다. 어떤 Queue로 메시지가 이동하고, 어떤 서비스가 그 Queue를 리스닝하는지 먼저 파악해야 Saga 패턴의 전체 흐름을 이해할 수 있다.

 

 

🗨️ JMeter로 성능 확인해보기

👉 애플리케이션의 성능을 테스트할 수 있습니다.

• 테스트계획 > 추가 > 쓰레드들 > 쓰레드 그룹을 추가합니다.

• 쓰레드 속성을 입력하여 테스트할 시나리오를 작성합니다. 아래의 이미지는 100명의 사용자가 각 10번씩 요청하여 1000번의 요청이 이루어지는 테스트 시나리오입니다.

• 쓰레드 그룹 > 추가 > 표본추출기 > HTTP 요청을 클립합니다.

• HTTP 요청의 항목을 입력합니다.

• HTTP 요청 > 추가 > 설정 엘리먼트 > HTTP 헤더 관리자를 추가합니다.

• 헤더 관리자의 속성을 추가합니다.

• 쓰레드 그룹 > 추가 > 리스너 > 결과들의 트리 보기 및 요약 보고서를 추가합니다.

• 상단의 재생 버튼을 클릭하면 테스트를 진행할 수 있습니다. 테스트를 재실행하기 전에는 클리어 버튼을 클릭하여 테스트를 초기화 할 수 있습니다.
• 요약 보고서에서는 처리량을 통해 TPS를 확인할 수 있습니다.
• 결과들의 트리보기에서는 테스트 요청에 대한 응답 결과를 볼 수 있습니다.

 

요약 보고서

JMeter 성능 테스트와 메시지 큐 기반 부하 분산 정리

• 이번 실습에서는 JMeter를 사용하여 RabbitMQ 기반 Saga 구조의 성능을 테스트하고, 메시지 큐를 통한 비동기 처리와 부하 분산의 효과를 확인하였다.

1. 현재 코드 구조의 한계와 DB 연동 필요성

• 현재 createOrder() 로직은 실제 DB에 데이터를 저장하지 않고, 단순히 메모리 객체를 생성한 뒤 RabbitMQ Queue로 메시지만 전달한다.
• 즉 흐름은:

Order 생성
↓
RabbitMQ Product Queue로 메시지 전송
↓
즉시 응답 반환

• 정도만 수행하기 때문에, 실제로는 무거운 작업이 거의 없다.

rabbitTemplate.convertAndSend(productQueue, deliveryMessage);

• 이후 바로 응답을 반환하므로 TPS(초당 처리량)가 매우 높게 측정될 수밖에 없다.
• 하지만 실제 서비스에서는 아래와 같은 작업이 추가된다.
  • JPA DB 저장
  • 재고 차감
  • 결제 처리
  • 외부 API 호출
• 특히 Order, Product, Payment 각각이 DB 작업을 수행하면 처리 시간이 증가하고 TPS는 자연스럽게 감소하게 된다.
• 즉 이번 테스트의 핵심은: 현재 높은 TPS는 "DB 작업이 거의 없는 구조"이기 때문이라는 점이다.

2. 메시지 큐 기반 비동기 처리와 부하 분산 효과

• 이번 구조의 핵심은 각 서비스가 독립적으로 동작한다는 점이다.

Order
→ Product
→ Payment

• 흐름으로 이어지지만, 각 서비스는 Queue를 통해 비동기적으로 연결된다.
• 즉 Order는 Product와 Payment의 처리 완료를 기다리지 않고, 메시지만 Queue에 넣은 뒤 자신의 작업을 끝낼 수 있다.

Order는 메시지를 Queue에 넣고 빠르게 응답
↓
Product와 Payment는 각자 처리 가능한 만큼 메시지를 소비

• 하는 구조이다.
• 이 방식의 장점:
  • 부하 분산
  • 서비스 간 결합도 감소
  • 특정 서비스 장애 시 전체 시스템 중단 방지
  • 높은 확장성
• 반대로 Queue 없이 Order가 모든 처리를 직접 담당하면:

주문 저장
→ 재고 처리
→ 결제 처리
→ 응답 반환

• 모든 작업이 Order에 집중되어 성능이 급격히 저하될 수 있다.
• 즉 Queue를 도입하면 메시지가 Queue에 쌓이고, 각 서비스가 자신이 처리 가능한 만큼만 가져가 처리하므로 전체 시스템 안정성이 높아진다.

3. Order 앞단 Queue 도입과 사용자 경험 개선

• 현재 구조에서는 Order가 DB 저장 후 바로 응답을 반환할 수 있기 때문에 TPS를 높게 유지할 수 있다.
• 하지만 DB 저장 자체도 부하가 될 수 있으므로, 더 나아가면:

사용자 요청
↓
Order Front Queue 적재
↓
비동기 Order 처리

• 구조로 확장할 수도 있다.
• 즉 DB 저장 자체를 Queue 기반 비동기 처리로 바꾸는 것이다.
• 이 경우 사용자는 주문 즉시:

"주문이 요청되었습니다.
마이페이지에서 주문 내역을 확인해주세요."

• 같은 빠른 응답을 받을 수 있다.
• 실제 주문 처리와 결제/재고 처리는 뒤에서 순차적으로 진행된다.
• 즉 핵심은 사용자는 빠른 응답을 받고, 실제 무거운 처리는 뒤에서 비동기로 처리하는 구조다.

4. DB 연동 이후의 실제 성능 테스트

• 실제 서비스에서는 사용자가 주문 직후 주문 조회를 시도할 수 있다.
• 하지만 비동기 구조에서는 아직 처리가 끝나지 않았을 수도 있으므로:

"주문이 아직 처리되지 않았습니다."
"잠시만 기다려주세요."

• 같은 상태 메시지를 보여줄 수 있다.
• 또한 Payment 상태에 따라 Queue 라우팅을 통해 Order 데이터를 미리 업데이트하는 방식도 사용할 수 있다.

예)

결제 성공
→ Order 상태 COMPLETE

결제 실패
→ Order 상태 CANCEL

• 다만 Queue 라우팅 구조는 매우 유연한 대신 아래와  같은 단점도 존재한다.
  • 라우팅 복잡도 증가
  • Queue 관리 증가
  • 기능 추가 시 라우팅 수정 필요

5. 이번 실습의 핵심 정리

• 이번 실습에서 가장 중요했던 점:

메시지 큐를 활용하면
각 서비스를 비동기적으로 분리하여
부하를 분산시킬 수 있다는 것

• 현재는 메모리 기반 테스트라 TPS가 높게 나왔지만, 실제 DB/JPA 연동 이후에는 성능 차이가 훨씬 크게 드러날 수 있다.

Order가 모든 작업을 직접 처리하는 구조보다,
Queue를 통해 역할을 분산한 구조가
확장성과 안정성 측면에서 훨씬 유리하다.