RabbitMQ & Kafka

대규모 시스템

• 방대한 양의 데이터를 처리하고 수많은 사용자의 요청을 동시에 처리할 수 있는 시스템
• RabbitMQ, Kafka와 같은 메시징 시스템을 활용하여 각 서비스간의 효율적인 데이터 통신과 확장성 보장

메시징 시스템

• Queue 형태로 메시지 저장
• Producer가 전송하고자 하는 메시지를 큐에 넣으면 Consumer는 자신의 속도에 맞춰 Queue에서 메시지 처리
• Message Broker를 사이에 두고 Producer와 Consumer가 간접적으로 데이터를 주고받음

기능

• 비동기 처리
  • Producer가 메시지를 발행한 후 consumer가 메시지를 처리할 때까지 기다리지 않아도 됨
• 데이터 손실 방지
  • 데이터를 메시지 큐에 안전하게 보관
• 부하 분산
  • 여러 consumer가 큐의 메세지를 가져가서 처리 → 시스템의 부하를 분산
  • 효율적인 데이터 처리 가능
• 스케일링
  • 수평적 확장 용이

RabbitMQ vs Kafka

차이점	RabbitMQ	Kafka
상호작용방식	Producer가 메시지를 전송하고 메시지가 의도한 Consumer에게 도착했는지 모니터링	Producer는 Consumer 수령 여부와 상관없이 Queue에 저장
아키텍쳐	- 복잡한 메시지 라우팅 - 메시지 큐 기반 설계 - push 모델 사용	- 더 복잡한 아키텍쳐(zookeeper) - 파티션 기반 설계 (하나의 주제안에 여러가지 queue들이 들어가있음) - pull 모델 사용 (처리 가능할 때 가져가는 모델)
메시징 처리 방식	- 우선순위 대기열 지원 - 순서대로 메시지 처리 - consumer 가 메시지 처리 → 브로커가 확인 응답 (ack) 전송 → 브로커가 대기열에서 메시지 삭제	- 우선순위 대기열 X - 파티션 내에선 순서 보장 (offset) - 메시지를 로그 파일에 추가 - 보존기간이 만료될 때까지 보관
성능	초당 수천개 메시지 처리 지연시간 짧음	초당 수백만개 메시지 처리 대용량 데이터 처리 RabbitMQ에 비해 실시간 처리량은 떨어짐 (consume할 수 있는 메시지를 가져오기 때문)

rabbit MQ가 지연시간이 짧은데 Kafka가 처리량이 많은 이유?

rabbitMQ는 하나씩 처리, Kafka는 벌크로 처리가 가능 → 초당 처리할 수 있는 데이터 양이 많음

MSA에서 쓰이는 이유?

• 서비스간 비동기 통신
  • 비즈니스 로직 안에서 비동기로 처리할 수 있는 부분을 분리하고 kafka/rabbitMQ로 처리 → 성능 향상
• 확장성
  • 증가하는 트래픽이나 데이터 양에 대응하여 성능을 유지시키거나 향상시킬 수 있는 환경이 중요
  • kafka → 파티셔닝, 클러스터에 브로커 추가
  • Rabbit MQ → 클러스터링, 분산 큐
• 데이터 일관성, 신뢰성
  • 각 MSA는 독립적으로 데이터 관리 → 서비스간 데이터 일관성 중요
  • Kafka → 로그, 메시지 저장 기능으로 데이터 일관성 유지
  • RabbitMQ → 메시지 디스크에 저장, 실패한 메시지 재처리 지원
• 서비스 간 결합도 ⬇️
  • Kafka → 이벤트 기반 아케틱쳐 지원, 서비스 간 독립적 동작 가능
  • Rabbit MQ → 큐 기반 메시징을 통해 서비스 간 메시지 전달

Kafka를 MSA에 적용하기

• Event-Driven Architecture
  • 각 MSA간 통신을 이벤트 기반으로 처리, 통신 주체는 Kafka
• 실시간 데이터 처리
  • 대규모 데이터 스트리밍 처리에 용이
  • 행동 로그, 모니터링 데이터 실시간 분석
• 모니터링과 로깅
  • 중앙 집중식 로그 수집

Event-Driven Architecture

코드

kafka: 
	bootstrap-servers: localhost:9092 
	consumer: # TODO kafka consumer, producer에 대한 설정값 수정은 여기에서 
		group-id: ${spring.application.name}-group 
		auto-offset-reset: earliest 
		key-deserializer: org.apache.kafka.common.serialization.StringDeserializer 
		value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer 
		properties: 
			spring.json.trusted.packages: '*' 
	producer: 
		key-serializer: org.apache.kafka.common.serialization.StringSerializer 
		value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

이벤트 발신

• KafkaTemplate을 사용해 이벤트 발신
• 토픽과 보내고자 하는 이벤트 내용을 담아 전송

@Service
@RequiredArgsConstructor
public class DeliveryService {
    private final DeliveryRepository deliveryRepository;
		private final KafkaTemplate<String, Object> kafkaTemplate; // 자동 주입
 
    @Transactional
    public void createDelivery(...) {
				Delivery delivery = new Delivery(...);
				deliveryRepository.save(delivery);
				
				DeliveryCreatedEvent event = new DeliveryCreatedEvent(...); //이벤트에 필요한 값을 넣어주세요
        kafkaTemplate.send("delivery-created", EventSerializer.serialize(event));
    }
}

이벤트 수신

• @KafkaListener를 사용해 메시지 수신
• groupId = yml에 설정한 group id

@Transactional
@KafkaListener(topics = "delivery-created", groupId = "notification-group")
public void handleDeliveryCreatedEvent(String message) {
    DeliveryCreatedEvent event = EventSerializer.deserialize(message, DeliveryCreatedEvent.class);
    
    // 이벤트로 해야하는 후 처리 진행 (비즈니스 로직)
    // e.g. 배달완료 이벤트 발행 
}