[백엔드] 이벤트 버스

# event_bus.py from typing import Dict, List, Type, Callable, Any import asyncio import logging import inspect logger = logging.getLogger(__name__) class EventBus: """비동기 지원 이벤트 버스 구현""" def __init__(self): # 이벤트 타입별 핸들러를 저장할 딕셔너리 self._handlers: Dict[Type, List[Callable]] = {} # 이벤트 내역 저장용 리스트 (디버깅 용도) self._event_history: List[Any] = [] self._max_history = 100 # 현재 실행 중인 비동기 태스크 추적 self._running_tasks: List[asyncio.Task] = [] def subscribe(self, event_type: Type, handler: Callable) -> None: """이벤트 핸들러를 등록합니다.""" if event_type not in self._handlers: self._handlers[event_type] = [] if handler not in self._handlers[event_type]: self._handlers[event_type].append(handler) logger.debug(f"핸들러 등록됨: {event_type.__name__} -> {handler.__name__}") async def publish(self, event: Any) -> None: """이벤트를 발행하고, 등록된 핸들러를 호출합니다.""" event_type = type(event) # 이벤트 내역 기록 self._event_history.append(event) if len(self._event_history) > self._max_history: self._event_history.pop(0) logger.debug(f"이벤트 발행됨: {event_type.__name__}") if event_type not in self._handlers: logger.debug(f"등록된 핸들러 없음: {event_type.__name__}") return # 핸들러 실행을 백그라운드에서 시작 task = asyncio.create_task(self._process_handlers(event, event_type)) self._running_tasks.append(task) task.add_done_callback(lambda t: self._running_tasks.remove(t)) async def _process_handlers(self, event: Any, event_type: Type) -> None: """핸들러를 비동기적으로 처리하는 내부 메서드""" tasks = [] for handler in self._handlers[event_type]: if inspect.iscoroutinefunction(handler): tasks.append(handler(event)) else: tasks.append(asyncio.to_thread(handler, event)) if tasks: await asyncio.gather(*tasks, return_exceptions=True) def get_running_tasks_count(self) -> int: """실행 중인 핸들러 수 반환 (디버깅용)""" return len(self._running_tasks)

# __init__.py from typing import Type, Callable, Any from common.arch.event_bus.event_bus import EventBus # 이벤트 버스 싱글톤 인스턴스 event_bus = EventBus() def on_event(event_type: Type): """동기식 이벤트 핸들러 데코레이터""" def decorator(handler: Callable): event_bus.subscribe(event_type, handler) return handler return decorator def on_event_async(event_type: Type): """비동기식 이벤트 핸들러 데코레이터""" def decorator(handler: Callable): event_bus.subscribe(event_type, handler) return handler return decorator __all__ = ["EventBus", "event_bus", "on_event", "on_event_async"]

from dataclasses import dataclass @dataclass class UserCreatedEvent: user_id: str email: str

from common.arch.event_bus import on_event @on_event(UserCreatedEvent) def handle_user_created(event: UserCreatedEvent): print(f"새로운 사용자 생성됨: {event.user_id}") # 동기 로직 처리

from common.arch.event_bus import on_event_async @on_event_async(UserCreatedEvent) async def send_welcome_email(event: UserCreatedEvent): await email_service.send_email( to=event.email, subject="환영합니다!", content="가입해 주셔서 감사합니다!" )

from common.arch.event_bus import event_bus async def create_user(user_data): # 사용자 생성 로직... user = {"user_id": "123", "email": "user@example.com"} # 이벤트 발행 await event_bus.publish(UserCreatedEvent(**user)) # 이벤트는 비동기적으로 처리되고 즉시 다음 코드 실행 return user

# event_bus_test.py import asyncio import time from datetime import datetime from dataclasses import dataclass import logging # 타임스탬프에 밀리초를 포함하는 커스텀 로그 포매터 class MillisecondFormatter(logging.Formatter): def formatTime(self, record, datefmt=None): # datetime 객체 생성 dt = datetime.fromtimestamp(record.created) # 밀리초 포함하여 포맷팅 return dt.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] # 로깅 설정 handler = logging.StreamHandler() formatter = MillisecondFormatter("%(asctime)s [%(levelname)s] %(message)s") handler.setFormatter(formatter) logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) # 기존 핸들러 제거 후 새 핸들러 추가 logger.handlers = [] logger.addHandler(handler) # 상위 로거로 전파 방지 logger.propagate = False # 이벤트 버스를 직접 import (독립적인 테스트를 위해) if __name__ == "__main__": logger.info("현재 디렉토리로 import") # 현재 디렉토리의 event_bus.py 직접 import from event_bus import EventBus # 이벤트 버스 인스턴스 생성 event_bus = EventBus() # 데코레이터 함수 정의 def on_event(event_type): def decorator(handler): event_bus.subscribe(event_type, handler) return handler return decorator def on_event_async(event_type): def decorator(handler): event_bus.subscribe(event_type, handler) return handler return decorator else: # 모듈로 임포트 시에는 일반 경로 사용 logger.info("절대 경로로 import") from common.arch.event_bus import event_bus, on_event, on_event_async # 테스트용 이벤트 정의 @dataclass class TestEvent: message: str timestamp: str = None def __post_init__(self): if self.timestamp is None: # 현재 시간을 읽기 쉬운 형식으로 저장 self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] # 이벤트 처리 결과 저장소 (테스트 검증용) results = [] # 동기 이벤트 핸들러 등록 @on_event(TestEvent) def sync_event_handler(event: TestEvent): """동기 이벤트 핸들러""" logger.info( f"[동기 핸들러] 시작: 메시지='{event.message}', 이벤트시간={event.timestamp}" ) results.append(f"동기:{event.message}") # 의도적으로 약간의 지연 추가 (처리 시간 시뮬레이션) time.sleep(3) completion_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] logger.info( f"[동기 핸들러] 완료: 메시지='{event.message}', 완료시간={completion_time}" ) # 비동기 이벤트 핸들러 등록 @on_event_async(TestEvent) async def async_event_handler(event: TestEvent): """비동기 이벤트 핸들러""" logger.info( f"[비동기 핸들러] 시작: 메시지='{event.message}', 이벤트시간={event.timestamp}" ) results.append(f"비동기:{event.message}") # 의도적으로 약간의 지연 추가 (비동기 처리 시간 시뮬레이션) await asyncio.sleep(3) completion_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] logger.info( f"[비동기 핸들러] 완료: 메시지='{event.message}', 완료시간={completion_time}" ) # 다른 비동기 이벤트 핸들러 등록 (멀티 핸들러 테스트) @on_event_async(TestEvent) async def another_async_handler(event: TestEvent): """추가 비동기 이벤트 핸들러""" logger.info( f"[다른 비동기 핸들러] 시작: 메시지='{event.message}', 이벤트시간={event.timestamp}" ) results.append(f"다른비동기:{event.message}") await asyncio.sleep(0.2) completion_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] logger.info( f"[다른 비동기 핸들러] 완료: 메시지='{event.message}', 완료시간={completion_time}" ) async def publish_events(): """여러 이벤트를 발행하는 비동기 함수""" logger.info("===== 이벤트 버스 테스트 시작 =====") # 첫 번째 이벤트 발행 logger.info("----- 첫 번째 이벤트 발행 -----") event1 = TestEvent(message="사용자_생성_이벤트") await event_bus.publish(event1) # 비동기 처리 확인을 위한 로그 (병렬 처리 시각화) logger.info(f"[시스템] 첫 번째 이벤트 발행 후 메인 스레드는 계속 실행됨") await asyncio.sleep(1) logger.info(f"[시스템] 첫 번째 이벤트는 백그라운드에서 계속 처리 중") # 두 번째 이벤트 발행 logger.info("----- 두 번째 이벤트 발행 -----") event2 = TestEvent(message="주문_완료_이벤트") await event_bus.publish(event2) # 비동기 처리 확인을 위한 로그 (병렬 처리 시각화) logger.info(f"[시스템] 두 번째 이벤트 발행 후 메인 스레드는 계속 실행됨") await asyncio.sleep(1) logger.info(f"[시스템] 첫/두 번째 이벤트는 백그라운드에서 병렬 처리 중") # 세 번째 이벤트 발행 (연속 발행 테스트) logger.info("----- 여러 이벤트 연속 발행 -----") events = [ TestEvent(message=f"결제_처리_이벤트"), TestEvent(message=f"알림_발송_이벤트"), TestEvent(message=f"재고_업데이트_이벤트"), ] for i, event in enumerate(events, 1): event_types = ["결제_처리", "알림_발송", "재고_업데이트"] # 이벤트 발행 시작 로깅 current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] logger.info( f"[시스템] {event_types[i-1]} 이벤트 발행 시작: 시간={current_time}" ) # 이벤트 발행 await event_bus.publish(event) # 이벤트 발행 완료 로깅 current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] logger.info( f"[시스템] {event_types[i-1]} 이벤트 발행 완료: 시간={current_time}" ) # 약간의 간격을 두고 발행 (로그 확인용) await asyncio.sleep(0.1) # 모든 이벤트 처리 완료 대기 current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] logger.info( f"[시스템] 모든 이벤트 발행 완료, 백그라운드 처리 확인중: 시간={current_time}" ) await asyncio.sleep(2) # 실행 중인 모든 이벤트 핸들러가 완료될 때까지 대기 wait_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] logger.info(f"[시스템] 모든 이벤트 핸들러 완료 대기 시작: 시간={wait_start_time}") # 실행 중인 태스크 수 확인 (최대 30초 대기) timeout = 30 # 최대 대기 시간(초) start_time = time.time() while event_bus.get_running_tasks_count() > 0: # 현재 실행 중인 태스크 수 로깅 if event_bus.get_running_tasks_count() > 0: logger.info( f"[시스템] 실행 중인 이벤트 핸들러: {event_bus.get_running_tasks_count()}개" ) # 제한 시간 확인 if time.time() - start_time > timeout: logger.warning( f"[시스템] 최대 대기 시간({timeout}초)이 초과되었습니다. 일부 핸들러는 여전히 실행 중입니다." ) break # 잠시 대기 await asyncio.sleep(1) wait_end_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] logger.info(f"[시스템] 모든 이벤트 핸들러 처리 완료: 시간={wait_end_time}") # 결과 출력 logger.info("===== 테스트 결과 =====") logger.info(f"수집된 결과: {results}") logger.info(f"발행된 이벤트 수: {len(event_bus.get_event_history())}") logger.info( f"이벤트 히스토리: {[e.message for e in event_bus.get_event_history()]}" ) logger.info("===== 테스트 완료 =====") # 메인 실행 함수 async def main(): try: await publish_events() except Exception as e: logger.error(f"테스트 중 오류 발생: {e}", exc_info=True) # 직접 실행 시 테스트 수행 if __name__ == "__main__": logger.info("이벤트 버스 테스트 시작...") asyncio.run(main())