[IT 잡썰] 간단한 Event Broker 구현 – 바람부는 자유의 블로그

이벤트 브로커를 직접 설계하는 경우 핵심 요소는 메시지의 송수신 처리, 메시지 큐 관리, 장애 대응 및 확장성입니다. 여기서는 기본적인 기능과 더불어 장애 복구 및 확장성을 고려한 이벤트 브로커 설계를 단계별로 설명하겠습니다.

이벤트 브로커 아키텍처 개요

이벤트 브로커는 서비스들 간의 중재자 역할을 하며, 발행자(Publisher)와 구독자(Subscriber)를 분리하여 이벤트 기반 통신을 가능하게 합니다. 이벤트 브로커는 다음과 같은 핵심 기능을 제공해야 합니다:

메시지 라우팅: 발행된 메시지를 구독자에게 전달합니다.
메시지 큐 관리: 메시지를 안전하게 저장하고 구독자가 처리할 수 있도록 보관합니다.
확장성 및 장애 대응: 대규모 트래픽을 처리하고 장애 발생 시 복구할 수 있도록 설계합니다.

구성 요소 설계

메시지 큐

메시지를 발행 후 큐에 저장하고 구독자가 처리할 수 있도록 하는 중심적인 요소입니다. 메시지 큐는 여러 가지 정책을 통해 안정적인 메시지 전달을 보장해야 합니다.

큐 저장소: 메시지를 임시로 저장할 저장소입니다. Redis, PostgreSQL, 또는 자체 파일 시스템 등을 사용할 수 있습니다.
메시지 TTL(Time-to-Live): 특정 시간이 지난 메시지를 삭제하는 기능을 추가하여, 처리되지 않은 오래된 메시지가 시스템을 차지하지 않도록 관리합니다.

토픽 관리

메시지를 주제별로 그룹화하여 서비스들이 특정 토픽을 구독할 수 있도록 합니다.

토픽 생성 및 삭제: 클라이언트 요청에 따라 동적으로 토픽을 생성하고 삭제할 수 있도록 API를 설계합니다.
구독 관리: 구독자가 원하는 토픽을 선택할 수 있도록 하고, 토픽별로 구독자를 관리하여 구독자 목록을 유지합니다.

메시지 라우터

발행된 메시지를 구독자에게 전달하는 역할을 담당합니다. 이를 통해 분산형 아키텍처에서 확장성을 유지하면서 구독자에게 메시지를 전달할 수 있습니다.

라우팅 규칙: 특정 조건에 따라 구독자에게 메시지를 전달하는 규칙을 설정합니다. 예를 들어, 특정 구독자에게만 특정 유형의 메시지를 전달하는 방식입니다.
로드 밸런싱: 여러 구독자가 동일한 토픽을 구독할 때, 메시지를 균등하게 배분하여 처리 속도를 높일 수 있도록 합니다.

내결함성 및 복구 시스템

서비스 중단이나 장애 상황에서도 브로커가 정상적으로 작동할 수 있도록 장애를 감지하고 복구하는 기능을 구현합니다.

재전송 메커니즘: 구독자가 메시지를 제대로 수신하지 못한 경우, 일정 횟수 재전송하거나 큐에 남겨두는 방식으로 복구를 지원합니다.
로그 저장소: 메시지의 상태를 추적할 수 있는 로그를 유지하여, 장애 상황에서도 메시지를 다시 처리할 수 있도록 합니다.

세부 기능 설계

메시지 발행 API

클라이언트가 브로커에 메시지를 발행할 때 사용하는 API입니다.

API 엔드포인트: /publish
HTTP 메서드: POST
입력 데이터: { "topic": "order_updates", "message": { "orderId": "123", "status": "completed" } }
기능: API 요청을 통해 들어온 메시지를 큐에 추가하고, 메시지 라우터가 구독자에게 전달하도록 큐에 저장합니다.

구독 API

구독자가 특정 토픽을 구독할 수 있도록 하는 API입니다.

API 엔드포인트: /subscribe
HTTP 메서드: POST
입력 데이터: { "topic": "order_updates", "subscriberUrl": "http://subscriber-service-url" }
기능: 구독자를 토픽에 등록하여 구독자 목록에 추가합니다. 새로운 메시지가 발행될 때 라우터가 해당 URL로 메시지를 전송할 수 있도록 구독자 정보가 큐에 등록됩니다.

메시지 배달 시스템

메시지가 발행되면, 이벤트 브로커는 해당 메시지를 구독자에게 비동기로 전송합니다.

라우팅 방식: 메시지 큐에서 메시지를 읽어 해당 토픽에 구독된 모든 구독자에게 전송합니다.
Retry 전략: 구독자가 일시적으로 다운되거나 메시지를 받을 수 없는 경우 재전송을 시도합니다. 예를 들어, 3회 재시도 후에도 전송 실패 시 로그를 남기고 이후 복구 시스템을 통해 다시 전송을 시도합니다.

로깅과 모니터링

메시지의 상태와 큐 상태를 모니터링하여 장애 발생 시 신속히 대처할 수 있도록 설계합니다.

로그 저장: 메시지 발행 시점, 배달 시점, 성공/실패 상태를 기록합니다.
모니터링 지표: 메시지 대기 시간, 성공/실패 비율, 재시도 횟수 등을 기록하여 시스템 상태를 추적합니다.

메시지 브로커 확장성 설계

샤딩(Sharding)과 파티셔닝

메시지 큐의 샤딩이나 파티셔닝을 통해 트래픽이 증가해도 확장 가능한 구조로 설계할 수 있습니다. 각 파티션은 독립된 큐로 운영되며, 특정 규칙에 따라 메시지를 적절한 파티션에 저장합니다.

로드 밸런싱

구독자 수가 많아질 경우 로드 밸런서를 통해 각 구독자가 균등하게 메시지를 받을 수 있도록 하고, 구독자 간 메시지 처리가 균등하게 이루어지도록 합니다.

장애 복구 및 보안 강화

메시지 중복 방지: 동일한 메시지가 중복 전송되지 않도록 메시지 ID로 중복 방지 로직을 추가합니다.
보안 인증: 발행자와 구독자 간의 인증을 통해 외부 침입을 방지합니다. API Key나 OAuth를 통해 인증을 관리할 수 있습니다.

간단한 코드 예시: 메시지 발행 및 구독 로직

다음은 메시지 발행 및 구독을 위한 파이썬 예시입니다.

# Publisher (발행자)
import requests
import json

def publish_message(topic, message):
    url = "http://localhost:5000/publish"
    data = { "topic": topic, "message": message }
    response = requests.post(url, json=data)
    return response.json()

publish_message("order_updates", { "orderId": "123", "status": "completed" })

# Subscriber (구독자)
from flask import Flask, request

app = Flask(__name__)

@app.route('/notify', methods=['POST'])
def notify():
    message = request.json.get("message")
    print("Received message:", message)
    return "Message processed", 200

if __name__ == "__main__":
    app.run(port=6000)

요약

이 설계는 발행자와 구독자가 서비스 브로커를 통해 간접적으로 통신하게 함으로써 서비스의 독립성과 확장성을 보장합니다. 메시지 큐, 토픽, 장애 복구 메커니즘, 모니터링과 로깅을 통해 서비스 브로커가 원활하게 운영되도록 설계되었습니다.