면접 준비 : MIRIDI
소켓 아키텍쳐 기반 EDA/MSA 중점 면접 준비
본문
메시지 저장 및 이벤트 발행
Transactional Outbox pattern을 통한 메시지 저장 및 이벤트 발행 동기화
- 처음 시스템을 구축할 때는 DB에 데이터가 쌓이면 서버가 계속해서 조회(Polling)하며 이벤트를 발행하는 방식
- 하지만 실시간 메시지 처리 시스템이라는 점과 메시지 트래픽을 고려했을 때 이는 너무 비효율적이라고 판단
- 관점을 바꿔 데이터베이스가 변경사항이 발생했을 때 직접 이벤트를 push하는 CDC 구조를 구현
MongoDB의 Change Streams를 이용해 outbox 테이블의 Insert 로그를 실시간으로 구독했고,
이 스트림을 Redis에 이벤트를 발행하는 DispatcherService와 Direct Connect 하였습니다.
이렇게 폴링 방식에서 Push 기반으로 전환하면서 서버의 불필요한 대기 시간을 없앴으며,
WebFlux의 retryWhen 같은 방어 로직을 스트림 내부에 추가하여, 순간적인 네트워크 단절이 발생하더라도
메시지 유실 없이 이벤트를 발행할 수 있는 견고한 구조를 구현하였습니다
소켓 서버의 이벤트 읽기 방식 최적화
서버 부하 및 Pooling 최적화
- 중계 서버에서 발행한 이벤트 처리 실패 재시도 로직을 위해서 유저별 큐를 생성하고 이를 실제 소켓 서버와 연결을 하는 구조
- 유저별 큐를 관리하고, 해당 큐를 계속해서 pooling하는 부하가, 유저가 늘어남에 따라 기하급수적으로 증가
- scale out을 고려했을 때, 유저의 중복 세션 관리 등의 어려움이 발생하여 소켓 서버 당 1개의 통합 큐를 배치하고, 메시지 분기 로직 구현
서버는 단일 큐를 구독하도록 하여, pooling 부하를 낮추고, WebFlux Sinks를 활용해 인메모리 상에서 유저와 웹소켓 세션을 매핑하고 관리하도록 하였습니다.
유저가 웹소켓 연결을 실행할 때, User- Sink 연결 정보를 기록하고, 후에 event를 받았을 때, event 내부의 목적지 정보를 기반으로,
Sink를 찾아 해당 위치로 Emit하여 실제 데이터 삽입이 완벽히 성공(EmitResult.OK)한 경우에만 Redis Streams에 xACK를 전달하는 구조로
서버 부하를 줄일 뿐 아니라 큐를 읽은 시점이 아닌, 실제 메시지가 웹소켓으로 전달된 시점을 성공으로 설정하여
중간 유실까지 억제할 수 있는 아키텍쳐를 고안하였습니다.
장애 상황별 맞춤형 장애 복구 파이프라인 구축
미승인 메시지 처리 및 장애 서버 PEL 재전달 로직 구현
- 메시지의 특성 상 상대방이 정해져 있기 때문에, 단순히 큐에서 전송 실패한 메시지에 대한 재시도 로직으로는 불가능함
- 소켓 서버가 다운될 경우, 해당 서버의 큐에 있는 모든 이벤트를 다시 중계 서버에서 실시간으로 유저를 찾아 재분배하는 로직 필요
메시지가 전달 안되는 장애 상황이 발생했을 때는 중계 서버에서 데이터를 재처리하도록 구현했습니다.
중계 서버에서 실시간으로 소켓 서버의 Heartbeat 모니터링을 통해 다운 감지 시스템을 통해 죽은 서버의 PEL을 전범위 xPENDING하도록 설정하고,
event의 수신자 정보를 확인해서 해당 유저의 실시간 접속 정보를 확인하여, 해당 서버 큐에 재 할당하고,
미 접속유저의 경우 알림을 보내도록 구현하였습니다.
[ 1. CS 공부 ]
심심할 때 마다 보기
https://velog.io/@youngjun_10/BackEnd-기술-면접-질문-정리
https://dev-coco.tistory.com/163
MSA / EDA
[ 2. 프로젝트 소개 : Socket Server ]
1. Web Socket Part
WebSocket vs SSE vs Rsocket (+Stomp를 왜 안썼나)
- WebSocket: 클라이언트와 서버 간의 양방향 실시간 통신을 지원하는 웹 표준 프로토콜
- SSE: 서버가 클라이언트로 데이터를 일방적으로 밀어넣는 단방향 통신 기술 (서버의 데이터 갱신)
- RSocket: 진보된 양방향 프로토콜 + Backpressure
- STOMP: WebSocket과 함께 사용 + 목적지 등의 정보를 기반으로 분기(for Pub/Sub)
1. 왜 순수 WebSocket을 선택했는가?
1-1. 왜 SSE를 쓰지 않았는가? (단방향의 한계)
클라이언트 to 서버 메시지 전송, 읽음 처리, 디바이스 정보 갱신, 연결 유지를 위한 Pong
- "수신용 SSE 채널"과 "송신용 REST 채널”
1-2. 왜 RSocket을 채택하지 않았는가? (오버엔지니어링 방지)
- 낮은 트래픽 : 사용자당 초당 수 건의 메시지 일시적인 트래픽
- Rsocket의 장점인 backpressure와 multiplexing(stream 다중화)가 불필요
- 채팅에서는 한 (사용자+domain) = 한 논리 스트림(자기 앞으로 오는 모든 이벤트)
- 외부로 나갈 경우 소켓 off = multi session이 아닌 경우 유저당 1개
- only Message 이미지 다운로드 등의 정체 구간 x
→ Event Type 단위 분기 + Coroutine Loop로 처리
2. 왜 STOMP도 채택하지 않았는가? (불필요한 추상화 제거)
- Relay 서버 : userId 기반 + 사용자의 접속 여부 확인 → 목적지에 따른 분배 불필요
- 경량화 : 가벼운 JSON 객체에
eventType필드만 추가
- redisStreams : Pub/Sub을 사용하지 않음
ChatWebSocketHandler - coroutineScope Loop
- PingPongHandler launch (25초 주기)
- 10초 동안 pong이 안오면 소켓 해제
- pong이 오면 redis ttl 초기화 → 25초 후에 ping
- MessageReceivedHandler launch : (Trigger : Flux emit = fanout)
- MESSAGE_NEW & ROOM_READ_UPDATE
- Receive Loop : pong/send message/read 등 서버가 받은 것 처리 (Trigger : 클라가 보낸 메시지)
- payload를 collect 해서 메시지 형태로
중요 : messageFlux.asFlow().collect {}형식이기 때문에
- suspend 즉 : emit 올 때까지 스레드 놓고 대기 (CPU 점유 x)
- Flux에 값이 들어오면 → lambda 함수 실행
멱등성 처리 (Message Send)
- 손실된 message = ACK가 오지 않음 = DB에 저장 실패
- 클라이언트가 재시도
- 클라이언트가 발급한 requestId를 Unique Key → catch DuplicatedKeyException
- 동일한 msgId, seq 정보 재 전송
2. CDC Part
kafka를 어떻게 대체 했나?
- Kafka : 초당 백만 건이상의 TPS가 필요한 환경 + 여러 데이터 소스에서 발생하는 대규모 데이터를 수집 및 처리하는 분산 스트리밍 플랫폼
- 단순 메시지큐(휘발성)가 아님
- 디스크에 log를 append only
- Offeset 기반 재처리 기능
- Retry topic / Dead Letter Queue
- 극단적 병렬성 (파티션 기반)
CDC인 Debezium과 같은 것이 필수! + zookeeper 같은 관리 Broker
1. 어떻게 kafka를 대체했나? (신뢰성 이벤트 전파와 에러 복구)
1-1. Mongo Change Stream (영속성 + CDC)
- outbox event table을 통해 영속성 보장
- oplog(append-only)를 push 방식으로 확인
1-2. Consumer Group & PEL (파티셔닝, Retry Topic)
- redisStreams의 Consumer Group은 파티셔닝과 동일한 역할
- PEL에 미처리 메시지가 저장되므로, 이를 xpending을 통해 재처리 로직으로
1-3. Relay & Sink (Zookeeper cluster) → 장애 감지 및 복구
- 어떤 kafka brocker 살아있는지 → 서버당 stream할당하고 서버가 살아있는지
- heartbeat check를 통한 60초 redis timestamp 갱신
- 토픽에 관한 연결 정보를 → 유저별 Sink연결 정보를 redis로 소유
- sub에서 바로 목적지인 해당 유저의 websocket으로 다이렉트
- Transcation Outbox Pattern : 메시지 저장 + outbox 저장을 하나의 transaction으로
- MongoChageStreams를 사용해서 insert 기준으로 oplog 인식 → 즉각적 푸시
- Oplog push (tail -f 방식)
- ResumeToken 방식 (redis 저장)
- 실제 Flux 형태
3. EDA Part
- fanout : RedisStreams
- 채팅 시스템에서는 roomId 기반 (유저의 실시간 접속 서버를 확인)
- stream-serverInstanceId
- 방안의 유저를 .forEach
- 각 유저를 xadd(socket or noti) 하는 블록을 launch
- 모든 유저가 launch 완료된 경우만
- dispatch 함수 return
coroutineScope (suspend = 동기적 블록)
- 동적 파티셔닝 : dispatcher = relay server에서 분기
- 세션 큐 → 서버 단위 큐 : pooling 방식인데 서버가 유저수 만큼 pooling?
- 인스턴스 다운 시 그 큐의 미처리 메시지를 다른 인스턴스가 인계
- PendingMessageService
- consumer Group + PEL
- 다음 round-trip에서 PEL을 우선 재처리
- SinkRegistry
- 유저 id 기반 멀티 세션의 두 웹소켓 모두 하나의 sink로
- pingpong 시에 sink caffeine ttl update
- Sink emit 성공 시점에 ACK
장애 대응
- 서버 heartbeat check by RelayServer
- death
- xpending → 자세한 메시지형태가 아닌 log형태
- 각 로그를 기반으로, range로 msg 재조회
- 현재 접속 instance id로 다시 xadd
- alive
- val messages = if (!pending.isNullOrEmpty()) pending else read()
→ read 이전에 pending 먼저
서버 간 통신
- 모듈화된 redisStreams를 이용해서 (직렬화된 node)를 이용한 통신
- 서버별 의존 X
- 각 서버는 직렬화된 node의 형태를 필요한 이벤트 형태로 convert
Flux.defer { ... }는 "매 구독 시점마다 새 flux = 새 메시지의 lazy publisher 생성자
retryWhen으로 재 호출 시 새롭게 사이클마다 새로운 메시지 read() 하기 위해
Stream 동작
Flux : Mongo Stream
→ reactiveMongoTemplate.changeStream("outbox", ...)
Reactor : 이후에 outbox 이벤트 처리 파이프라인(Mono)
→ mono { dispatch(outbox) }.then(saveResumeToken(token))
- Mono : 1-element stream
Coroutine : Dispatcher : 동기 코드 블럭
- 실제 redisStreams에 xadd
dispatch return → saveResumeToken → 다음 outbox (concateMap으로 강제로 순차적)
[ 3. 프로젝트 설계 : 동시 편집]
동시 편집
기존 문제점 : 초당 수백번의 이벤트 처리 불가
- Outbox에 저장하면 DB가 버티지 못하고
- Change Stream을 거쳐가는 과정에서 실시간성이 떨어져 화면이 버벅
CRDT (충돌 없는 복제 데이터 삽입)
- 수학의 교환 법칙 기반
- 순서가 꼬이고, 중복되도, 결국 모든 클라이언트가 수식 계산시 동일한 상태에 도달
OT : 순차 병합
대신
2.5번 index 사용
아주 짧은 버퍼 유지 MQ or PUBSUB
- 몇초전의 커서 움직임 보다 현재 상태(실시간성)가 더 중요
- 약간의 손실 + 순서보장은 CRDT 위임 이면 → 실시간성 latency에 중점
저장 : 쉴때 (Debouncing)
변경사항이 많을 때는 저장 x
사용자가 타이핑을 멈추고 3초가 지나면 DB에 스냅샷을 저장한다.
단, 사용자가 쉬지 않고 타이핑하더라도 데이터 유실 방지를 위해 최대 30초(Throttle/Max Interval)마다 한 번씩은 무조건 DB에 강제로 플러시(Flush)한다.
보내는 쪽 설계
각 웹소켓 입력을
1. CRDT Handler를 이용해서 mono로 coroutineLoop를 Launch
- asFlux() 형태로 받아서
2. Start pipeLine (Flux)
- CRDT니까 순서보장이 중요치 않으므로 Flux
- CRDT 로직이 하나의 이벤트로
- publish로 event를 읽을때, debounce와 flush를 활용해서 Flux 데이터 merge
- 1번 case 미입력 3초
- 2번 case 주기 30초
- 이를 통해 생성된 concateMap을 snapshot 형태로 저장
- snapshot에 고유 id 부여 → 중복 작업 방지 및 1case/2case 동시 실행 방지
- snapshot에 해당하는 CRDT 이벤트들을 병합
- 아직 DB에 저장하지 않은 채로
- Redis Pub/sub Publish (문서 id를 topic)
- 이후 DB 저장 (백업용)
받는 쪽 설계
해당 파이프라인을 sub하는 서버
- 병합된 스냅샷(CRDT 묶음 → 하나의 CRDT)
- CRDT의 일부가 오지 않았다
- CRDT의 시간 비교를 통해 이후의 CRDT 동작에 없는 모르는 값만 delta 요청
→ publish
