Page History
MVC WEBSOCKET 을 설정하는 방법은 이전장에서 소개가 되었으며
여기서는 WebFlux가 지원하는 ReactiveWebSocket을 살펴보겠습니다WebFlux가 지원하는 ReactiveWebSocket을 Spring Boot 기본 Websocket(MVC)과 구현방식과 작동방식의 차이를 알아보겠습니다.
다음과 같은 큰 차이가 있기때문에, 적은 자원으로 웹소켓 커넥션 고성능 효율을 가지려면 Reactive-WebFlux모드가 권장됩니다.
...
| Code Block | ||
|---|---|---|
| ||
package org.example.kotlinbootreactivelabs.ws
import org.example.kotlinbootreactivelabs.ws.model.EventTextMessage
import org.example.kotlinbootreactivelabs.ws.model.MessageFrom
import org.example.kotlinbootreactivelabs.ws.model.MessageType
import org.example.kotlinbootreactivelabs.ws.model.sendReactiveEventTextMessage
import org.springframework.web.reactive.socket.WebSocketSession as ReactiveWebSocketSession
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Component
import java.util.concurrent.ConcurrentHashMap
@Component
class WebSocketSessionManager {
private val logger = LoggerFactory.getLogger(WebSocketSessionManager::class.java)
val reactiveSessions = ConcurrentHashMap<String, ReactiveWebSocketSession>()
val topicSubscriptions = ConcurrentHashMap<String, MutableSet<String>>()
fun addSession(session: ReactiveWebSocketSession) {
reactiveSessions[session.id] = session
logger.info("[SessionManager] Connected: ${session.id}")
sendReactiveEventTextMessage(session, EventTextMessage(
type = MessageType.INFO,
message = "You are connected - ${session.id}",
from = MessageFrom.SYSTEM,
id = null,
jsondata = null,
))
}
fun removeSession(session: ReactiveWebSocketSession) {
reactiveSessions.remove(session.id)
logger.info("[SessionManager] Disconnected: ${session.id}")
}
fun subscribeReactiveToTopic(sessionId: String, topic: String) {
topicSubscriptions.computeIfAbsent(topic) { mutableSetOf() }.add(sessionId)
logger.info("Session $sessionId subscribed to topic $topic")
}
fun unsubscribeReactiveFromTopic(sessionId: String, topic: String) {
topicSubscriptions[topic]?.remove(sessionId)
logger.info("Session $sessionId unsubscribed from topic $topic")
}
fun sendReactiveMessageToSession(sessionId: String, message: String) {
reactiveSessions[sessionId]?.let {
sendReactiveEventTextMessage(
it, EventTextMessage(
type = MessageType.PUSH,
message = message,
from = MessageFrom.SYSTEM,
id = null,
jsondata = null,
)
)
}
}
fun sendReactiveMessageToTopic(topic: String, message: String) {
topicSubscriptions[topic]?.forEach { sessionId ->
reactiveSessions[sessionId]?.let {
sendReactiveEventTextMessage(
it, EventTextMessage(
type = MessageType.PUSH,
message = message,
from = MessageFrom.SYSTEM,
id = null,
jsondata = null,
)
)
}
}
}
} |
- https://docs.spring.io/spring-integration/reference/reactive-streams.html#reactive-channel-adapters
- reactive가 지원하는 reactive adapter를 이용해 pub/sub stream 처리를 더 우아하게 할것으로 기대해봅니다.
- https://docs.spring.io/spring-integration/reference/amqp.html
Reactive 모드로 작성된 소켓 pub/sub/sentoSession 위한 TEST 툴도 준비되어있으며 전체코드도 포함되어있습니다.
- 인증모드는 제외되었으며 API를 통해 pub/sub/sendToSession 기능을 모두 수행할수 있습니다.
- 전체 코드 : https://github.com/psmon/java-labs/tree/master/KotlinBootReactiveLabs/src/main/kotlin/org/example/kotlinbootreactivelabs/ws
- MCV Websocket을 다양하게 이용하기 : MVC WEBSOCKET SocketHandeler 처리방식만 다르며 다양한 변종기능을 적용할수 있습니다.