Server-Sent Events (SSE) 알람 : 활용편
다국어 알람 메세지 구현
실제 업무에서 Server-Sent Event(SSE) 를 이용하여 알람을 구현했던 기능은 아래와 같습니다.
- 주기적으로 DB 를 조회한 이전 조회 시간 이후, 새로운 데이터가 있는 경우 알람을 보내는 기능
- 추가적으로 클라이언트가 선택한 언어에 따라, 알람 메세지도 각 언어로 데이터를 전송
전체적인 흐름과 구성
클라이언트 화면에서 언제든 언어를 변경할 수 있습니다. 선택한 언어에 따라서 SSE 연결 요청하고, 그때마다 요청 HEADER 에 Accept-Language
을 사용하여 선택한 언어를 전달하게 구현돼 있습니다.
> 언어값을 HEADER - Accept-Language 로 전달 받자
DB 에는 primary key 로 id
와 lang
을 사용하고 같은 id
에 각각 ko
, en-US
, jp
등의 키 값으로 알람 메세지 데이터가 각각 들어 있는 상태입니다.
id* | lang* | message | isn_dt |
---|---|---|---|
5b48ff24 | ko | 예약에 성공했습니다! | 2023-12-01 11:10:04 |
5b48ff24 | US-en | I have successfully made a reservation. | 2023-12-01 11:10:04 |
5b48ff24 | ja | 予約に成功しました | 2023-12-01 11:10:04 |
서버 구현
클라이언트가 SSE 요청 하면서 보내주는 lang
값을 이용해서 DB 데이터를 조회하면 됩니다. Flask 에서는 request
객체에 전달되는 accept_languages
중에서 best
속성값을 이용해 손쉽게 정보를 가져올 수 있습니다.
SSE 연결하면서 전달 받은 언어 정보를 제너레이터 함수의 인자값으로 넣어주면 해당 값을 이용해서 DB 를 조회하면 사용자가 선택한 언어에 알맞은 알람이 전달 됩니다.
from flask import request
from sse import get_user_alarm_by_lang
@app.get("/connection/<user_id>")
def connection(user_id: str):
lang = request.accept_languages.best
return Response(
get_user_alarm_by_lang(user_id, lang), content_type="text/event-stream"
)
from time import sleep
import json
def select_alarm(user_id, lang):
# DB 조회 로직
return { "message": "예약에 성공했습니다!" }
def get_user_alarm_by_lang(user_id, lang):
PENDING_TIME = 5
while True:
data = select_alarm(user_id, lang)
yield f"""event: alarm\ndata: {json.dumps(data)}\n\n"""
sleep(PENDING_TIME)
사용자가 언어를 변경한다면?
사용자가 이미 SSE 요청을 한 이후, 화면에서 언어를 변경한다고 가정해 봅시다. 이때 이전 언어로 호출한 제너레이터는 더 이상 필요하지 않기 때문에 종료가 필요합니다. 그리고 변경된 언어의 lang
파라미터로 다시 새로운 제너레이터를 호출 합니다.
- 클라이언트가 선택한 언어로 다시 SSE 연결 요청
- 이전의 연결 존재하는 경우는 해당 연결을 종료
- 새로운 연결만 존재하고 이를 통해 알람 전달
SSE 문제점
하지만 문제는 SSE 는 서버에서 클라이언트로만 통신이 가능한 단방향 통신이라는 점입니다. 서버에서 클라이언트로 데이터를 보내는 것만 가능하며 클라이언트로부터의 응답이나 상태 확인은 기본적으로 제공되지 않습니다. 물론 기본적인 연결 오류 감지 기능을 통해 연결을 종료 할 수 있습니다. 하지만 올바른 방식으로 여러 번 SSE 연결을 요청하는 경우 중복된 연결을 확인 할 수 없기에 추가적인 보완이 필요합니다.
connection 관리 구현
connection 관리는 간단하게 global 변수를 이용해서 구현 가능합니다. 우선은 connection 을 관리 할 global 변수를 dict
형태로 생성해 둡니다. 가장 최근의 요청의 임의값으로 global 변수를 최신화 해두고 이를 비교하는 로직을 작성하면 됩니다.
- 각 SSE 요청마다 임의값 생성
- connection 관리하는 global 변수에 임의값을 덮어쓰기
- 제너레이터 호출 하면서 각 요청마다 생성한 임의값을 전달
- 제너레이터 내부에서 전달받은 임의값과 connection 관리하는 global 변수를 비교
- 최신 연결이 아닌 경우,
break
를 통해서 제너레이터를 종료하고 연결을 종료
from flask import request
from sse import get_user_alarm_by_lang, client_connections
from uuid import uuid4
@app.get("/connection/<user_id>")
def connection(user_id: str):
current_conn_id = uuid4().hex
client_connections[user_id] = current_conn_id # 가장 최신값으로 덮어씀
lang = request.accept_languages.best
return Response(
get_user_alarm_by_lang(user_id, lang, current_conn_id),
content_type="text/event-stream"
)
from time import sleep
import json
client_connections = {}
def get_user_alarm_by_lang(user_id, lang, current_conn_id):
PENDING_TIME = 5
while True:
# global 변수의 최신값과 호출시 전달 받은 값을 비교
if client_connections.get(user_id) != current_conn_id:
print(f"BYE BYE conn_id:{current_conn_id}")
break
data = select_alarm_by_lang(user_id, lang)
yield f"""event: alarm\ndata: {json.dumps(data)}\n\n"""
sleep(PENDING_TIME)
클라이언트 구현
SSE 는 기본적으로 클라이언트에서 연결이 끊어진 경우, 자동 재연결을 하기 때문에 클라이언트에서도 서버가 연결을 끊음에 따라 알맞은 구현이 필요합니다.
const eventSource = new EventSource('http://localhost:5000/connection/tiaz');
eventSource.onerror = function() {
eventSource.close()
// 필요에 따라 여기에서 재연결 로직을 구현할 수 있습니다.
};
동시성 문제
여기서 하나 더 생각해야 할것이 있습니다. Flask 자체는 멀티스레딩을 지원하는 웹 프레임워크입니다. 때문에 경쟁 상태(race conditions), 데드락(deadlocks)과 같은 동시성 문제를 야기할 수 있습니다. 이를 방지하기 위해 적절한 동기화 메커니즘을 사용해야 합니다. 여기서는 스레드락을 사용하겠습니다.
from threading import Lock
connections_lock = Lock()
@app.get("/connection/<user_id>")
def connection(user_id: str):
current_conn_id = uuid4().hex
with connections_lock:
client_connections[user_id] = current_conn_id
lang = request.accept_languages.best
return Response(
get_user_alarm_by_lang(user_id, lang, current_conn_id),
content_type="text/event-stream",
)
에필로그 : 더 생각해야 하는 것
추가적으로 Flask 는 멀티 프로레싱 또한 지원합니다. 따라서 flask 애플리케이션을 구동하는 환경이 멀티프로세스 인 경우, 각 프로세스는 자체 메모리 공간을 가집니다. 따라서 위에서 구현한 전역 변수는 프로세스 간에 공유되지 않습니다.
또한 전역 변수를 사용하면 서버의 상태가 유지되므로, 서버의 무상태(stateless) 특성이 손상될 수 있습니다. 이는 확장성과 유지보수성에 영향을 줄 수 있습니다.
따라서 이러한 환경에서는 Redis와 같은 외부 데이터 저장소를 사용하여 데이터를 공유하게 개선이 필요합니다. 😊