Event Broker

Event Broker

이벤트 브로커는 이벤트 기반 아키텍처에서 이벤트 발행자 (Producer) 와 구독자 (Consumer) 사이의 이벤트 메시지 흐름을 중개하는 미들웨어이다.
분산 마이크로서비스 환경에서 데이터 흐름의 실시간성, 비동기성, 확장성을 확보할 수 있도록 지원한다. 대용량 실시간 데이터 처리와 시스템 간의 결합도 감소, 장애 격리 및 이벤트 아카이빙 등 다양한 장점으로 인해 오늘날 IT 인프라에서 매우 중요한 역할을 차지한다.

이벤트 브로커는 Pub/Sub, 스트리밍, 다양한 QoS, 보안, 트랜잭션, 확장성, 장애 복구 등 다양한 기능을 제공하며, 시스템 간 결합도를 낮추고, 실시간 데이터 분배와 비동기 처리를 지원한다.

대표 솔루션으로는 Solace, Kafka, AWS EventBridge, Azure Event Grid 등이 있으며, 다양한 산업에서 실시간 분석, 알림, IoT, 마이크로서비스 통합 등 광범위하게 활용된다.

배경

Event Broker 는 다음과 같은 기술적 요구사항과 발전 과정을 통해 등장했다:

1. 분산 시스템의 복잡성 증가

   • 마이크로서비스 아키텍처의 확산
   • 시스템 간 상호 의존성 증가
   • 동기식 통신의 한계 극복 필요성

2. 실시간 데이터 처리 요구 증가

   • IoT 디바이스의 폭발적 증가
   • 실시간 분석 및 대응 시스템 구축 필요
   • 스트리밍 데이터 처리 패러다임 변화

3. 기존 메시지 브로커의 한계

   • 지점 간 통신 중심의 설계
   • 메시지 지속성과 재처리 어려움
   • 확장성과 처리량의 제약

목적 및 필요성

1. 시스템 분리 (Decoupling)

   • 이벤트 생산자와 소비자의 완전한 분리
   • 시간적, 공간적 분리를 통한 유연성 확보
   • 의존성 최소화로 독립적인 개발 및 배포 가능

2. 확장성 향상 (Scalability)

   • 수평적 확장을 통한 처리 능력 증대
   • 동적 확장 및 축소 지원
   • 워크로드 분산을 통한 효율성 개선

3. 내결함성 보장 (Fault Tolerance)

   • 시스템 일부 장애가 전체에 미치는 영향 최소화
   • 자동 복구 및 재시도 메커니즘
   • 데이터 손실 방지를 위한 지속성 보장

4. 실시간 반응성 (Real-time Responsiveness)

   • 이벤트 발생 즉시 관련 시스템들의 반응
   • 배치 처리 대비 현저한 지연시간 단축
   • 동적 비즈니스 요구사항에 신속한 대응

핵심 개념

Event Broker는 이벤트를 발행 (Publish) 하는 서비스와 이를 구독 (Subscribe) 하는 소비자 사이에서 중계자 역할을 하는 시스템 구성 요소다.

기본 개념

1. 이벤트 (Event)

   • 시스템에서 발생하는 상태 변화를 나타내는 메시지
   • 불변성 (Immutable) 특성을 가지며 과거에 발생한 사실을 기록
   • 이벤트 헤더 (메타데이터) 와 이벤트 페이로드 (실제 데이터) 로 구성

2. 발행 - 구독 패턴 (Publish-Subscribe Pattern)

   • 이벤트 발행자 (Publisher) 와 구독자 (Subscriber) 간의 비동기 통신 모델
   • 발행자는 구독자의 존재를 알 필요가 없는 완전한 분리
   • 토픽 (Topic) 기반의 이벤트 분류 및 라우팅

3. 토픽 (Topic)

   • 이벤트를 분류하고 라우팅하는 논리적 채널
   • 계층적 구조를 통한 효율적인 이벤트 관리
   • 와일드카드를 활용한 구독 패턴 지원

4. 이벤트 스트림 (Event Stream)

   • 시간 순서로 정렬된 이벤트들의 연속된 흐름
   • 내구성과 재처리 가능성을 제공
   • 실시간 및 배치 처리 모두 지원

실무 구현을 위한 핵심 연관성

확장성 측면:

• 수평적 확장을 통한 처리량 증대
• 파티셔닝을 통한 병렬 처리 지원
• 로드 밸런싱과 백프레셔 관리

신뢰성 측면:

• 이벤트 지속성과 복제를 통한 데이터 안정성
• 정확히 한 번 처리 (Exactly-once processing) 보장
• 장애 복구 및 재시작 메커니즘

성능 측면:

• 비동기 처리를 통한 응답성 향상
• 배치 처리와 압축을 통한 효율성 개선
• 메모리 및 디스크 최적화

주요 기능 및 역할

Event Broker 는 다음과 같은 핵심 기능들을 수행한다:

1. 이벤트 수집 및 수신 (Event Ingestion)

   • 다양한 소스로부터 이벤트 수집
   • 이벤트 검증 및 스키마 확인
   • 처리량 조절 및 백프레셔 관리

2. 이벤트 라우팅 및 분배 (Event Routing & Distribution)

   • 토픽 기반의 이벤트 분류
   • 구독 패턴에 따른 선택적 배포
   • 다중 소비자 지원

3. 이벤트 지속성 (Event Persistence)

   • 내구성 있는 이벤트 저장
   • 복제를 통한 데이터 안정성
   • 히스토리 데이터 관리

4. 이벤트 순서 보장 (Event Ordering)

   • 파티션 내 순서 보장
   • 글로벌 순서 제어 옵션
   • 타임스탬프 기반 정렬

특징

1. 비동기 통신 (Asynchronous Communication)

   • 논블로킹 이벤트 전송
   • 생산자와 소비자의 독립적 처리
   • 시스템 응답성 향상

2. 높은 처리량 (High Throughput)

   • 대용량 이벤트 스트림 처리
   • 배치 처리 및 압축 최적화
   • 병렬 처리 지원

3. 내결함성 (Fault Tolerance)

   • 자동 장애 감지 및 복구
   • 다중 복제본 관리
   • 네트워크 파티션 내성

4. 확장성 (Scalability)

   • 수평적 확장 지원
   • 동적 파티션 관리
   • 로드 밸런싱

핵심 원칙

1. 이벤트 불변성 (Event Immutability)

   • 한 번 기록된 이벤트는 변경되지 않음
   • 히스토리 추적 및 감사 가능
   • 데이터 일관성 보장

2. 최소한의 브로커 (Dumb Broker, Smart Client)

   • 브로커는 단순한 전달 역할에 집중
   • 클라이언트가 복잡한 로직 처리
   • 성능과 확장성 최적화

3. 토픽 기반 라우팅 (Topic-based Routing)

   • 논리적 채널을 통한 이벤트 분류
   • 계층적 토픽 구조 지원
   • 와일드카드 구독 패턴

주요 원리 및 작동 방식

다음은 Event Broker 의 핵심 작동 원리를 보여주는 다이어그램입니다:

graph TB
    subgraph "Event Producers"
        P1[Producer 1<br/>Web Application]
        P2[Producer 2<br/>Mobile App]
        P3[Producer 3<br/>IoT Sensor]
    end
    
    subgraph "Event Broker Cluster"
        EB1[Event Broker 1<br/>Partition A]
        EB2[Event Broker 2<br/>Partition B]
        EB3[Event Broker 3<br/>Partition C]
        
        EB1 -.->|Replication| EB2
        EB2 -.->|Replication| EB3
        EB3 -.->|Replication| EB1
    end
    
    subgraph "Topics"
        T1[user.events]
        T2[order.events]
        T3[sensor.data]
    end
    
    subgraph "Event Consumers"
        C1[Consumer 1<br/>Analytics Service]
        C2[Consumer 2<br/>Notification Service]
        C3[Consumer 3<br/>Inventory Service]
        C4[Consumer 4<br/>ML Pipeline]
    end
    
    P1 -->|Publish| EB1
    P2 -->|Publish| EB2
    P3 -->|Publish| EB3
    
    EB1 --> T1
    EB2 --> T2
    EB3 --> T3
    
    T1 -->|Subscribe| C1
    T1 -->|Subscribe| C2
    T2 -->|Subscribe| C3
    T3 -->|Subscribe| C4
    
    style EB1 fill:#f9f,stroke:#333,stroke-width:2px
    style EB2 fill:#f9f,stroke:#333,stroke-width:2px
    style EB3 fill:#f9f,stroke:#333,stroke-width:2px

작동 원리 설명:

1. 이벤트 발행: 프로듀서들이 다양한 토픽으로 이벤트를 발행
2. 파티셔닝: 이벤트가 파티션별로 분산 저장
3. 복제: 내결함성을 위한 데이터 복제
4. 구독 및 소비: 컨슈머들이 관심 있는 토픽을 구독하여 이벤트 소비

구조 및 아키텍처

graph TB
    subgraph "Client Layer"
        PC[Producer Clients]
        CC[Consumer Clients]
    end
    
    subgraph "Event Broker Infrastructure"
        subgraph "Broker Cluster"
            B1[Broker Node 1]
            B2[Broker Node 2]
            B3[Broker Node 3]
        end
        
        subgraph "Storage Layer"
            S1[Segment 1]
            S2[Segment 2]
            S3[Segment 3]
        end
        
        subgraph "Coordination Service"
            ZK[ZooKeeper/Raft<br/>Metadata Management]
        end
    end
    
    subgraph "Monitoring & Management"
        M1[Metrics Collection]
        M2[Health Monitoring]
        M3[Configuration Management]
    end
    
    PC -.->|Produce Events| B1
    PC -.->|Produce Events| B2
    PC -.->|Produce Events| B3
    
    B1 -.->|Consume Events| CC
    B2 -.->|Consume Events| CC
    B3 -.->|Consume Events| CC
    
    B1 <--> S1
    B2 <--> S2
    B3 <--> S3
    
    B1 <--> ZK
    B2 <--> ZK
    B3 <--> ZK
    
    B1 --> M1
    B2 --> M2
    B3 --> M3

구성요소

구현 기법

장점

단점과 문제점 그리고 해결방안

단점

문제점

도전 과제

분류 기준에 따른 종류 및 유형

Event Broker 기술별 비교

요약

Event Broker 구축 예제

Kafka 를 사용하여 Event Broker 환경을 구축하고 간단한 Producer-Consumer 이벤트 흐름을 구현한다.

구성:

• Apache Kafka
• Python (kafka-python 라이브러리)

1. 설치 준비

   1 2 3 4 5 6 7 8 9

   # Kafka & Zookeeper 실행 (Docker 기반 예시) docker run -d --name zookeeper -p 2181:2181 zookeeper docker run -d --name kafka -p 9092:9092 \ -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \ -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \ -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT \ -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \ --network=host \ confluentinc/cp-kafka

2. Python Producer 코드

   1 2 3 4 5 6 7 8 9 10 11 12 13 14

   # producer.py from kafka import KafkaProducer import json producer = KafkaProducer( bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8') ) event = {'event_type': 'OrderCreated', 'order_id': 1234, 'user_id': 5678} producer.send('order-events', value=event) producer.flush() print("✅ OrderCreated 이벤트 전송 완료")

3. Python Consumer 코드

   1 2 3 4 5 6 7 8 9 10 11 12 13 14

   # consumer.py from kafka import KafkaConsumer import json consumer = KafkaConsumer( 'order-events', bootstrap_servers='localhost:9092', auto_offset_reset='earliest', group_id='order-service', value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) for message in consumer: print(f"✅ 수신된 이벤트: {message.value}")

Kafka 모니터링 지표 및 실전 대시보드 구성

Tool: Prometheus + Grafana
주요 지표:

실전 대시보드 구성

1. Exporter 설치
   Kafka 용 JMX Exporter 사용:

   1 2	java -javaagent:/opt/jmx_prometheus_javaagent.jar=7071:/opt/kafka.yml \ -jar kafka-server-start.sh …

2. Prometheus 설정 예시

   1 2 3 4

   scrape_configs: - job_name: 'kafka' static_configs: - targets: ['localhost:7071']

3. Grafana 대시보드 템플릿

   • ID: 7589 (confluentinc 제공 Kafka Dashboard)
   • 대시보드 포함 항목:
     • 브로커별 처리량
     • 파티션 상태
     • lag 상태
     • consumer group health
     • 메시지 지연 현황

실무에서 효과적으로 적용하기 위한 고려사항 및 주의할 점

아키텍처 설계 시 고려 요소

배포 전략

테스트 및 검증 전략

운영 자동화 및 모니터링

운영 항목

자동화 요소

최적화하기 위한 고려사항 및 주의할 점

실무 사용 예시

활용 사례

사례 1: IoT 센서 로그와 웹 서버 로그를 실시간으로 수집·분석

시스템 구성도

flowchart LR
    IoT[IoT/웹서버] --> Broker[Event Broker]
    Broker --> Analyzer[Log Analyzer/Consumer]
    Broker --> Storage[Raw Log 저장소]
    Analyzer --> BI[DashBoard/Alert]

Workflow

1. IoT/웹서버에서 로그 이벤트 발생 → 브로커로 발행
2. 브로커가 Analyzer(실시간 처리) 및 Storage(저장) 에 동시에 브로드캐스트
3. Analyzer 는 실시간 알림 또는 대시보드에 결과 전달

주제의 역할: 서로 다른 데이터 소스를 빠짐 없이 실시간 브로드캐스팅, 대용량 처리, 장애 시 재처리 가능

유무 차이: 이벤트 브로커 없을 시, 다수 대상 동시 전달 힘들고 확장/유지보수 취약, 장애 전파 쉬움

구현 예시:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# Kafka를 통한 이벤트 브로커 실습 예시
from kafka import KafkaProducer, KafkaConsumer

# Producer - 이벤트 전송
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
producer.send('log-events', b'{"msg":"로그 수집 이벤트"}')

# Consumer - 이벤트 수신
consumer = KafkaConsumer(
    'log-events',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    group_id='log-analyzer'
)
for message in consumer:
    print(f"수신된 이벤트: {message.value.decode()}")
    # 이곳에 실시간 분석 등 처리 준비

• Producer 는 ’log-events’ 토픽에 메시지를 보내고, Consumer 는 같은 토픽을 구독하여 메시지를 실시간으로 소비/처리

사례 2: 이커머스 주문 시스템에서의 Event Broker 활용

시스템 구성:

• Producer: 주문 마이크로서비스
• Broker: Apache Kafka
• Consumer:
  • 재고 시스템
  • 배송 시스템
  • 알림 서비스
  • 이벤트 로깅 (ELK)

flowchart TD
    A[Order Service] --> B[Kafka Topic: order.created]
    B --> C[Inventory Service]
    B --> D[Shipping Service]
    B --> E[Notification Service]
    B --> F[Analytics Dashboard]

Workflow:

1. order.created 이벤트가 Kafka 에 발행됨
2. Kafka 는 이를 다수 Consumer 에 브로드캐스트
3. 각 Consumer 는 독립적으로 이벤트를 처리 (예: 배송 트리거, 알림 전송, 재고 차감)

역할 분석:

• Event Broker는 Producer-Consumer 간 결합을 제거
• 복수 시스템에서 동시에 이벤트 소비 가능
• Kafka 는 이벤트 로그를 유지하여 리플레이/장애 복구도 지원

유무 비교:

구현 예시:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# Kafka Producer - 주문 이벤트 발행
from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

order_event = {
    "order_id": 1001,
    "status": "created",
    "timestamp": "2025-07-18T09:00:00Z"
}

producer.send('order.created', value=order_event)
producer.flush()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Kafka Consumer - 배송 서비스
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'order.created',
    bootstrap_servers='localhost:9092',
    group_id='shipping-service',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for msg in consumer:
    order = msg.value
    print(f"배송 처리 시작 - 주문번호: {order['order_id']}")
    # 배송 처리 로직

사례 3: 글로벌 전자상거래 플랫폼의 실시간 주문 처리

대형 전자상거래 플랫폼에서 Event Broker 를 활용하여 주문부터 배송까지의 전체 워크플로우를 실시간으로 처리하는 시스템을 구축한 사례.

시스템 구성

graph TB
    subgraph "Frontend Layer"
        WEB[Web Application]
        MOBILE[Mobile App]
        API[API Gateway]
    end
    
    subgraph "Event Broker Infrastructure"
        KAFKA[Apache Kafka Cluster]
    end
    
    subgraph "Microservices"
        ORDER[Order Service]
        PAYMENT[Payment Service]
        INVENTORY[Inventory Service]
        SHIPPING[Shipping Service]
        NOTIFICATION[Notification Service]
        ANALYTICS[Analytics Service]
    end
    
    subgraph "Data Storage"
        ORDERDB[(Order DB)]
        INVENTORYDB[(Inventory DB)]
        USERDB[(User DB)]
        CACHE[(Redis Cache)]
    end
    
    WEB --> API
    MOBILE --> API
    API --> ORDER
    
    ORDER --> KAFKA
    PAYMENT --> KAFKA
    INVENTORY --> KAFKA
    SHIPPING --> KAFKA
    
    KAFKA --> ORDER
    KAFKA --> PAYMENT
    KAFKA --> INVENTORY
    KAFKA --> SHIPPING
    KAFKA --> NOTIFICATION
    KAFKA --> ANALYTICS
    
    ORDER --> ORDERDB
    INVENTORY --> INVENTORYDB
    PAYMENT --> USERDB
    NOTIFICATION --> CACHE

워크플로우

sequenceDiagram
    participant Customer
    participant OrderService
    participant EventBroker
    participant PaymentService
    participant InventoryService
    participant ShippingService
    participant NotificationService
    
    Customer->>OrderService: 주문 요청
    OrderService->>EventBroker: order.created 이벤트 발행
    
    EventBroker->>PaymentService: 결제 처리 트리거
    EventBroker->>InventoryService: 재고 확인 트리거
    
    PaymentService->>EventBroker: payment.processed 이벤트 발행
    InventoryService->>EventBroker: inventory.reserved 이벤트 발행
    
    EventBroker->>ShippingService: 배송 준비 트리거
    EventBroker->>NotificationService: 고객 알림 트리거
    
    ShippingService->>EventBroker: shipping.prepared 이벤트 발행
    EventBroker->>NotificationService: 배송 알림 트리거
    NotificationService->>Customer: 주문 완료 알림

Event Broker 의 역할

• 실시간 이벤트 라우팅: 주문 생성 시 관련된 모든 서비스에 즉시 알림
• 비동기 처리: 각 서비스가 독립적으로 작업을 수행하여 전체 응답 시간 단축
• 장애 격리: 한 서비스의 장애가 다른 서비스에 미치는 영향 최소화
• 확장성: 트래픽 증가 시 개별 서비스 확장 가능

Event Broker 유무에 따른 차이점

• Event Broker 사용 시:
  • 평균 주문 처리 시간: 2-3 초
  • 시스템 간 결합도: 낮음
  • 확장성: 개별 서비스 단위로 수평 확장 가능
  • 장애 영향 범위: 제한적
• Event Broker 미사용 시 (동기식 API 호출):
  • 평균 주문 처리 시간: 8-12 초
  • 시스템 간 결합도: 높음
  • 확장성: 전체 시스템 단위로만 확장 가능
  • 장애 영향 범위: 연쇄적 장애 발생 가능

구현 예시:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267

import json
import asyncio
from typing import Dict, List, Callable, Any
from dataclasses import dataclass, asdict
from datetime import datetime
import logging

# 이벤트 클래스 정의
@dataclass
class Event:
    """
    이벤트의 기본 구조를 정의하는 클래스
    - topic: 이벤트가 발행될 토픽
    - data: 이벤트 페이로드 데이터
    - timestamp: 이벤트 발생 시각
    - event_id: 이벤트 고유 식별자
    """
    topic: str
    data: Dict[str, Any]
    timestamp: datetime = None
    event_id: str = None
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now()
        if self.event_id is None:
            import uuid
            self.event_id = str(uuid.uuid4())

# Event Broker 핵심 클래스
class EventBroker:
    """
    이벤트 브로커의 핵심 기능을 구현하는 클래스
    - 토픽 기반 이벤트 라우팅
    - 비동기 이벤트 처리
    - 구독자 관리
    """
    
    def __init__(self):
        self.subscribers: Dict[str, List[Callable]] = {}  # 토픽별 구독자 목록
        self.event_history: List[Event] = []  # 이벤트 히스토리 (재처리용)
        self.logger = logging.getLogger(__name__)
        
    async def publish(self, event: Event) -> None:
        """
        이벤트를 지정된 토픽에 발행
        - 이벤트 히스토리에 저장 (재처리 지원)
        - 해당 토픽의 모든 구독자에게 비동기 전달
        """
        self.logger.info(f"Publishing event {event.event_id} to topic '{event.topic}'")
        
        # 이벤트 히스토리에 저장
        self.event_history.append(event)
        
        # 토픽의 구독자들에게 이벤트 전달
        if event.topic in self.subscribers:
            tasks = []
            for subscriber in self.subscribers[event.topic]:
                task = asyncio.create_task(self._deliver_event(subscriber, event))
                tasks.append(task)
            
            # 모든 구독자에게 병렬로 이벤트 전달
            await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _deliver_event(self, subscriber: Callable, event: Event) -> None:
        """
        개별 구독자에게 이벤트 전달 (내부 메서드)
        - 에러 핸들링 포함
        - 재시도 로직 구현 가능 지점
        """
        try:
            if asyncio.iscoroutinefunction(subscriber):
                await subscriber(event)
            else:
                subscriber(event)
        except Exception as e:
            self.logger.error(f"Error delivering event {event.event_id}: {e}")
            # 실제 환경에서는 Dead Letter Queue로 전송
    
    def subscribe(self, topic: str, callback: Callable) -> None:
        """
        특정 토픽에 구독자 등록
        - 와일드카드 패턴 지원 가능
        - 구독자별 필터링 옵션 추가 가능
        """
        if topic not in self.subscribers:
            self.subscribers[topic] = []
        
        self.subscribers[topic].append(callback)
        self.logger.info(f"Subscriber registered for topic '{topic}'")
    
    def unsubscribe(self, topic: str, callback: Callable) -> None:
        """
        구독자 해제
        """
        if topic in self.subscribers:
            try:
                self.subscribers[topic].remove(callback)
                self.logger.info(f"Subscriber unregistered from topic '{topic}'")
            except ValueError:
                self.logger.warning(f"Subscriber not found for topic '{topic}'")
    
    async def replay_events(self, topic: str, from_timestamp: datetime = None) -> None:
        """
        과거 이벤트 재처리 기능
        - 특정 시점 이후의 이벤트 재처리
        - A/B 테스트, 버그 수정 후 재처리 등에 활용
        """
        filtered_events = [
            event for event in self.event_history
            if event.topic == topic and (
                from_timestamp is None or event.timestamp >= from_timestamp
            )
        ]
        
        self.logger.info(f"Replaying {len(filtered_events)} events for topic '{topic}'")
        
        for event in filtered_events:
            await self.publish(event)

# 실제 사용 예시: 전자상거래 주문 처리
class OrderService:
    """
    주문 서비스 구현 예시
    - 주문 생성 시 이벤트 발행
    """
    
    def __init__(self, event_broker: EventBroker):
        self.event_broker = event_broker
        self.logger = logging.getLogger(self.__class__.__name__)
    
    async def create_order(self, order_data: Dict[str, Any]) -> str:
        """
        주문 생성 및 이벤트 발행
        """
        order_id = f"order_{datetime.now().strftime('%Y%m%d%H%M%S')}"
        
        # 주문 데이터 저장 (DB 저장 로직)
        self.logger.info(f"Order {order_id} created")
        
        # 주문 생성 이벤트 발행
        event = Event(
            topic="order.created",
            data={
                "order_id": order_id,
                "customer_id": order_data["customer_id"],
                "items": order_data["items"],
                "total_amount": order_data["total_amount"]
            }
        )
        
        await self.event_broker.publish(event)
        return order_id

class PaymentService:
    """
    결제 서비스 구현 예시
    - 주문 생성 이벤트 구독
    - 결제 처리 후 이벤트 발행
    """
    
    def __init__(self, event_broker: EventBroker):
        self.event_broker = event_broker
        self.logger = logging.getLogger(self.__class__.__name__)
        
        # 주문 생성 이벤트 구독
        self.event_broker.subscribe("order.created", self.handle_order_created)
    
    async def handle_order_created(self, event: Event) -> None:
        """
        주문 생성 이벤트 처리
        """
        order_data = event.data
        self.logger.info(f"Processing payment for order {order_data['order_id']}")
        
        # 결제 처리 로직 (실제로는 외부 결제 게이트웨이 연동)
        await asyncio.sleep(0.5)  # 결제 처리 시뮬레이션
        
        # 결제 완료 이벤트 발행
        payment_event = Event(
            topic="payment.completed",
            data={
                "order_id": order_data["order_id"],
                "payment_status": "success",
                "transaction_id": f"txn_{datetime.now().strftime('%Y%m%d%H%M%S')}"
            }
        )
        
        await self.event_broker.publish(payment_event)

class InventoryService:
    """
    재고 서비스 구현 예시
    """
    
    def __init__(self, event_broker: EventBroker):
        self.event_broker = event_broker
        self.logger = logging.getLogger(self.__class__.__name__)
        
        # 주문 생성 이벤트 구독
        self.event_broker.subscribe("order.created", self.handle_order_created)
    
    async def handle_order_created(self, event: Event) -> None:
        """
        주문 생성 시 재고 확인 및 예약
        """
        order_data = event.data
        self.logger.info(f"Checking inventory for order {order_data['order_id']}")
        
        # 재고 확인 및 예약 로직
        await asyncio.sleep(0.3)  # 재고 처리 시뮬레이션
        
        # 재고 예약 완료 이벤트 발행
        inventory_event = Event(
            topic="inventory.reserved",
            data={
                "order_id": order_data["order_id"],
                "items": order_data["items"],
                "reservation_status": "confirmed"
            }
        )
        
        await self.event_broker.publish(inventory_event)

# 사용 예시 및 테스트
async def main():
    """
    Event Broker 사용 예시
    """
    # 로깅 설정
    logging.basicConfig(level=logging.INFO)
    
    # Event Broker 초기화
    broker = EventBroker()
    
    # 서비스 인스턴스 생성
    order_service = OrderService(broker)
    payment_service = PaymentService(broker)
    inventory_service = InventoryService(broker)
    
    # 주문 생성 테스트
    order_data = {
        "customer_id": "customer_123",
        "items": [
            {"product_id": "prod_1", "quantity": 2, "price": 29.99},
            {"product_id": "prod_2", "quantity": 1, "price": 49.99}
        ],
        "total_amount": 109.97
    }
    
    print("=== 주문 처리 시작 ===")
    order_id = await order_service.create_order(order_data)
    
    # 이벤트 처리 완료를 위한 대기
    await asyncio.sleep(2)
    
    print(f"\n=== 주문 {order_id} 처리 완료 ===")
    print(f"총 처리된 이벤트 수: {len(broker.event_history)}")
    
    # 이벤트 히스토리 출력
    print("\n=== 이벤트 히스토리 ===")
    for event in broker.event_history:
        print(f"[{event.timestamp}] {event.topic}: {event.data}")

# 실행
if __name__ == "__main__":
    asyncio.run(main())

이 구현 예시는 Event Broker 의 핵심 개념들을 실제 코드로 보여줍니다:

1. 비동기 이벤트 처리: asyncio 를 활용한 논블로킹 처리
2. 토픽 기반 라우팅: 토픽별 구독자 관리
3. 느슨한 결합: 서비스 간 직접 호출 없이 이벤트를 통한 통신
4. 이벤트 재처리: 히스토리 기반 재처리 기능
5. 에러 핸들링: 구독자별 예외 처리

사례 4: 실시간 알림 서비스

SNS, 커뮤니티, 커머스 등 거의 모든 서비스에서 사용되며, 일반적으로 Event-Driven Architecture + Pub/Sub 기반 알림 브로커 + WebSocket/Push API 조합으로 구현된다.

시스템 구성:

graph TD
  A[사용자] -->|댓글 작성| B(API Gateway)
  B --> C[Comment Service]
  C -->|Event 생성| D(Kafka Topic: comment.events)
  D --> E[Notification Service]
  E -->|WebSocket 전송| F[WebSocket Gateway]
  F --> G[실시간 연결 사용자]
  E -->|DB 저장| H[Notification DB]

Workflow:

구현 코드:

1. Kafka 이벤트 발행–Comment Service

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

   # comment_service.py from kafka import KafkaProducer import json producer = KafkaProducer( bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8') ) def publish_comment_event(comment_data): event = { "type": "CommentCreated", "post_id": comment_data["post_id"], "commenter_id": comment_data["user_id"], "receiver_id": comment_data["post_owner_id"], "message": comment_data["text"] } producer.send('comment.events', event) producer.flush()

2. Kafka 이벤트 소비 + WebSocket 연동–Notification Service

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40

   # notification_service.py from kafka import KafkaConsumer import json from fastapi import FastAPI, WebSocket from collections import defaultdict import asyncio app = FastAPI() active_connections = defaultdict(list) # user_id -> list of WebSockets @app.websocket("/ws/{user_id}") async def websocket_endpoint(websocket: WebSocket, user_id: str): await websocket.accept() active_connections[user_id].append(websocket) try: while True: await websocket.receive_text() # Keep connection alive finally: active_connections[user_id].remove(websocket) def send_realtime_notification(user_id, message): for ws in active_connections[user_id]: asyncio.create_task(ws.send_text(message)) def run_kafka_consumer(): consumer = KafkaConsumer( 'comment.events', bootstrap_servers='localhost:9092', group_id='notification-group', value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) for msg in consumer: data = msg.value receiver_id = data['receiver_id'] text = f"새 댓글이 달렸습니다: {data['message']}" send_realtime_notification(receiver_id, text) # 비동기 Kafka consumer 실행 import threading threading.Thread(target=run_kafka_consumer, daemon=True).start()

3. 클라이언트 (WebSocket JS 예시)

   1 2 3 4 5

   const socket = new WebSocket("ws://localhost:8000/ws/USER_ID"); socket.onmessage = function(event) { alert("🔔 알림: " + event.data); };

사례 5: 마이크로서비스 아키텍처에서 서비스 간 통합

주문 처리 시스템

• 사용자가 주문을 생성하면 Order Service 가 OrderCreated 이벤트를 발행하고, Inventory Service 와 Payment Service 가 이를 구독해 각각 재고 확인과 결제 처리를 수행한다.

시스템 구성:

flowchart TD
  A[Client] -->|POST /orders| B[Order Service]
  B -->|OrderCreated Event| C[Kafka Topic: order.events]
  C --> D[Inventory Service]
  C --> E[Payment Service]
  D -->|InventoryReserved| F[Kafka Topic: inventory.events]
  E -->|PaymentProcessed| G[Kafka Topic: payment.events]

Workflow:

구현 예시:

1. Order Service–이벤트 발행

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27

   # order_service.py from kafka import KafkaProducer import json from fastapi import FastAPI, Request app = FastAPI() producer = KafkaProducer( bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8') ) @app.post("/orders") async def create_order(req: Request): body = await req.json() order = { "order_id": body["order_id"], "user_id": body["user_id"], "items": body["items"], "total": body["total"] } # 가상 DB 저장 생략 producer.send('order.events', { "type": "OrderCreated", "data": order }) producer.flush() return {"status": "Order created and event published"}

2. Inventory Service–이벤트 소비자

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# inventory_service.py
from kafka import KafkaConsumer, KafkaProducer
import json

consumer = KafkaConsumer(
    'order.events',
    bootstrap_servers='localhost:9092',
    group_id='inventory-service',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def handle_order_created(event_data):
    items = event_data['data']['items']
    # 재고 확인 로직 (가정)
    all_available = True  # 여기서는 항상 성공 가정
    if all_available:
        result = {
            "type": "InventoryReserved",
            "data": {
                "order_id": event_data['data']['order_id'],
                "status": "reserved"
            }
        }
        producer.send('inventory.events', result)
        producer.flush()

for msg in consumer:
    event = msg.value
    if event['type'] == "OrderCreated":
        handle_order_created(event)

1. Payment Service–이벤트 소비자

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35

   # payment_service.py from kafka import KafkaConsumer, KafkaProducer import json consumer = KafkaConsumer( 'order.events', bootstrap_servers='localhost:9092', group_id='payment-service', value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) producer = KafkaProducer( bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8') ) def handle_order_created(event_data): amount = event_data['data']['total'] # 결제 처리 로직 (가정) is_paid = True if is_paid: result = { "type": "PaymentProcessed", "data": { "order_id": event_data['data']['order_id'], "status": "paid" } } producer.send('payment.events', result) producer.flush() for msg in consumer: event = msg.value if event['type'] == "OrderCreated": handle_order_created(event)

주목할 내용

반드시 학습해야 할 내용

용어 정리

참고 및 출처

Message Broker & Event Broker

• 메시지 브로커란 무엇인가요? | IBM
• 메시지 브로커 - 위키백과
• Event Broker 개념 및 기능
• What is an Event Broker - Confluent

Message Broker Vs Event Broker 비교

• 메시지 브로커와 이벤트 브로커의 이해와 활용
• 메시지 브로커와 이벤트 브로커: 분산 시스템의 핵심 기술
• Message Broker vs Event Broker 비교 분석 - Riskified Tech
• 메시지 브로커 vs 이벤트 처리 도구 비교 - Upsolver

RabbitMQ Vs Kafka

• RabbitMQ와 Kafka - 메시지 대기열 시스템 간의 차이점 - AWS
• Kafka(이벤트 브로커) vs RabbitMQ(메시지 브로커)
• 카프카(kafka) vs RabbitMQ

Event-Driven Architecture & Event Mesh

• 이벤트 기반 아키텍처란? | Red Hat
• 이벤트 메쉬(Event Mesh)의 개념, 기능, 작동 방식, 사용 방법 | Red Hat
• Event-Driven Architecture 패턴 가이드 - Microsoft Azure
• 솔라스 Event Broker 아키텍처 예시

실무 적용 & 최적화 사례

• EDA 실무 적용 사례 - Estuary
• EDA 성능 최적화 및 브로커 튜닝
• Event-Driven Architecture 실무 사례 - Estuary

플랫폼별 공식 문서

• Apache Kafka 공식 문서 (Event Broker 구조)
• AWS EventBridge 공식 페이지
• Azure Event Grid 개요
• Google Cloud Eventarc 소개
• AsyncAPI 공식 웹사이트

추가 자료

• Message Broker(메시지 브로커)란 무엇일까?
• Message Broker 활용