내가 이해하려고 적는 카프카

카프카란

카프카는 대규모 실시간 데이터 스트리밍을 위해 설계된 분산 이벤트 스트리밍 플랫폼이다.
대략적인 구조는 시스템에서는 프로듀서가 메시지를 생성하고, 이를 카프카 클러스터의 토픽으로 전송한다.
토픽은 여러 파티션으로 나뉘어 있어 병렬 처리가 가능하고, 컨슈머는 만들어진 파티션에서 메시지를 읽어 처리한다.
이런 구조로 인해 카프카는 높은 처리량과 확장성을 제공하며, 데이터의 영속성도 보장한다.

코드로 가보자
온라인 쇼핑몰에서 주문이 들어올 때마다 이를 처리하고, 재고를 업데이트하는 시스템을 Kafka를 이용해 예시를 들겠다.

Producer(주문 접수) 코드: 주문 정보를 생성하고 ‘sinsang-orders’ 토픽으로 전송한다.

require 'kafka'

# Kafka 브로커 주소 설정
# 브로커란? --> 카프카 클러스터를 구성하는 서버로 기본단위이자 
# 메시지를 저장하고 관리하는 역할을 한다. 
# 여러 브로커가 클러스터를 이루어 고가용성과 확장성을 제공한다.
kafka = Kafka.new(["localhost:9092"])

# 주문 정보 생성 (웹 요청이 왔다고 가정)
order = {
  id: "12345",
  product_id: "ABC123",
  quantity: 2,
  customer_id: "CUST001"
}.to_json

# 'sinsang-orders' 토픽으로 주문 정보 전송
# 토픽이란? 카프카에서 메시지를 카테고리화하는 논리적 단위다. 각 토픽은 여러 파티션으로 나뉘어 
# 병렬 처리와 확장성을 제공한다.
kafka.deliver_message(order, topic: "sinsang-orders")

puts "주문이 접수되었습니다: #{order}"

Consumer(주문 처리) 코드

require 'kafka'

kafka = Kafka.new(["localhost:9092"])

# 컨슈머 그룹 생성 
# 컨슈머란? --> 토픽에서 메시지를 읽고 처리하는 애플리케이션이다.
# 컨슈머 그룹을 통해 여러 컨슈머가 협력하여 메시지를 병렬로 처리할 수 있다.
consumer = kafka.consumer(group_id: "order-processing-group")

# 'new-orders' 토픽 구독
consumer.subscribe("sinsang-orders")

# 메시지 처리
consumer.each_message do |message|
  order = JSON.parse(message.value)
  
  puts "새로운 주문을 처리합니다: #{order}"
  
 
  process_order(order)  # 주문이 처리되고 DB에 저장하는 로직
  
  # 재고 업데이트를 위한 새로운 메시지 발행
  update_stock_message = {
    product_id: order['product_id'],
    quantity: order['quantity']
  }.to_json

  # 재고업데이트를 위한 새 메시지 토픽을 발행
  kafka.deliver_message(update_stock_message, topic: "stock-updates")
end

이 컨슈머 코드는 ‘sinsang-orders’ 토픽에서 메시지를 읽어 처리하고, 재고 업데이트를 위한 새 메시지를 ‘stock-updates’ 토픽으로 발행한다.
카프카는 주문 정보를 안정적으로 전달하고, 여러 컨슈머가 병렬로 주문을 처리할 수 있게 해준다.
시스템 간 느슨한 결합으로 주문저장과 재고 업데이트와 같은 후속 작업을 위한 새로운 메시지를 생성하여 서버의 부담을 줄인다.

추가적인 설명

• 파티션: 각 토픽은 여러 파티션으로 나뉘어 있다. 이를 통해 데이터를 병렬로 처리할 수 있고, 시스템의 처리량을 높일 수 있다.
• 오프셋: 각 메시지는 파티션 내에서 고유한 오프셋 값을 가진다. 컨슈머는 이 오프셋을 이용해 자신이 어디까지 메시지를 처리했는지 추적한다.
• 복제: 데이터 안정성을 위해 각 파티션은 여러 브로커에 복제될 수 있다. 이를 통해 일부 브로커에 장애가 발생해도 시스템이 계속 작동할 수 있다.

더 디테일한 내용은 이 블로그 글을 참고하면 좋을듯하다.

그럼 이걸 왜 다안쓰냐…?
내가 경험해본 바로는 분산시스템이기에 완벽한 무결성을 보장하기 어렵다. 예시로 들었지만 재고같은 숫자를 다루는 데이터보다 상대적으로 인스타의 좋아요나 대규모 로그처리 실시간 행동분석 등에서는 적합하나 금전거래나 의료기록같은 1원까지 틀어지면 안되는 곳에서는 사용을 지양하는것이 좋은 것 같다.