활성/비활성 사용자 구분 후 비활성 사용자 웹소켓 연결 끊김 처리

 

1. 개요

- 콘서트 티켓 프로젝트에서는 사용자에게 웹소켓 연결을 제공합니다.

  그런데 사용자의 웹소켓 연결이 끊어지게 되는 경우가 있는데,

  이 경우는 2가지 케이스로 나누어집니다.

  (1) 사용자가 일시적인 네트워크 이상으로 연결이 끊어지는 경우

  (2) 사용자가 서비스에서 이탈한 경우  

 

- 그리고 서버는 이 두 가지 케이스를 구분해서 서로 다른 조치를 취해야 합니다.

  (1)의 경우에는 사용자 세션 정보나 대기열 정보, 5분 좌석 선점 예약 등을 그대로 유지해야 하고,

  (2)의 경우에는 사용자 세션 정보나 대기열 정보, 5분 좌석 선점 예약 등을 삭제해야 합니다.   

- 그리고 이 구분을 위해서는 사용자가 웹소켓으로 연결되는 경우,

  헬스체크 데이터를 지속적으로 수집하여 서버에서 관리해야 합니다. 

2. 기술 스택

- 이 시스템은 WebSocket과 Redis를 활용하여 구현되었습니다.

3. 동작 방식

1. 사용자 웹소켓 연결 
   • 사용자가 웹소켓 서버와 웹소켓 연결을 맺습니다.  .
2. 사용자(웹소켓 클라이언트)가 헬스체크 데이터 발송
   
   • 사용자는 일정 주기(예: 15초)마다 헬스체크 데이터를 웹소켓 서버로 발송합니다. 
3. 스케줄러를 통한 비활성 사용자 제거 
   • 스프링 서버에서 스케줄러를 통해 비활성 사용자(ex) 헬스체크 데이터가 30초 초과)의
     세션 정보와 대기열(or 활성화열) 정보를 제거합니다.
4. 비활성 사용자의 5분 좌석 선점 예약 제거
   
   • 해당 비활성 사용자가 5분 좌석 선점 예약을 했었다면,
     해당 5분 좌석 선점 예약 정보도 제거합니다. 
     
     

이 글은 이러한 활성/비활성 사용자 구분 후 비활성 사용자 웹소켓 연결 끊김 처리를 다룹니다. 

 

4. 구현 상세 

 

(1) 웹소켓 헬스체크 설정

@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic")
               .setTaskScheduler(heartBeatScheduler())
               .setHeartbeatValue(new long[]{15, 15});
        config.setApplicationDestinationPrefixes("/api");
        config.setUserDestinationPrefix("/user");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/gs-guide-websocket").setHandshakeHandler(new CustomHandshakeHandler());
    }

    @Bean
    public TaskScheduler heartBeatScheduler() {
        return new ThreadPoolTaskScheduler();
    }
}

 

stompClient.onConnect = (frame) => {
    setConnected(true);
    console.log('Connected: ' + frame);

    const userName = frame.headers["user-name"]; // user-name 추출
    const heartBeat = frame.headers["heart-beat"];
    const [clientHeartbeat, serverHeartbeat] = heartBeat.split(",").map(Number);

    console.log(`Client heartbeat interval: ${clientHeartbeat}s, Server heartbeat interval: ${serverHeartbeat}s`);

    stompClient.subscribe('/topic/token', (response) => {
        console.log('Response body: ', response.body);  // 추가: 응답 내용 확인
        const token = JSON.parse(response.body).token;
        showToken(token); // token을 화면에 출력
        window.token = token
    });

    stompClient.subscribe('/topic/waitingQueue/status', (response) => {
        console.log('Response body: ', response.body);  // 추가: 응답 내용 확인
        const status = JSON.parse(response.body).status; // 응답에서 status 값을 추출
    });

    stompClient.subscribe('/user/topic/token', (response) => {
        console.log('Response body: ', response.body);  // 추가: 응답 내용 확인
        const token = JSON.parse(response.body).token; // 응답에서 token 값을 추출
        showToken(token); // token을 화면에 출력
    });

    stompClient.subscribe('/user/topic/reconnect', (response) => {
        console.log('Response body: ', response.body);  // 추가: 응답 내용 확인
        const result = JSON.parse(response.body).result; // 응답에서 result 값을 추출
    });

     if (clientHeartbeat) {
           setInterval(() => {
           if(window.token){
                sendHeartbeat(window.token);
           }
       }, clientHeartbeat * 1000);  // 클라이언트의 heartbeat 간격에 맞춰 실행
     }
};

- 웹소켓 연결 시 Broker를 통해 헬스체크 정보를 웹소켓 클라이언트에 전달합니다.

  웹소켓 클라이언트는 해당 정보를 수신 후, Heartbeat 주기(ex 15초)에 맞춰서
  헬스체크 정보를 서버로 발송합니다. 

 

 

(2) 헬스체크 정보 업데이트

package com.example.concertTicket_websocket.waitingqueue.service;

import com.example.concertTicket_websocket.websocket.infrastructure.HeartbeatDAO;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class HeartbeatService {

    private final HeartbeatDAO heartbeatDAO;
    // 사용자의 Heartbeat 정보를 최신화하기 위한 기능
    public void updateUserHealthStatus(String token, String timestamp) {
        heartbeatDAO.updateUserHealthStatus(token, timestamp);
    }
}

// 사용자의 Heartbeat 정보를 최신화하기 위한 기능
public void updateUserHealthStatus(String token, String timestamp) {
     RMapCache<String, String> heartbeatMap = redissonClient.getMapCache(RedisKey.HEARTBEAT_HASH_KEY);
     heartbeatMap.put(token, timestamp, 30, TimeUnit.MINUTES);
 }

 

- 웹소켓 서버는 해당 헬스체크 정보를 수신한 후, 

   Redis에 사용자의 토큰을 key로  헬스체크 정보를 value로 업데이트합니다.   

 

 

(3) 스케줄러를 통한 비활성 사용자 제거

package concert.application.waitingqueue.scheduler;

import concert.domain.waitingqueue.entities.dao.TokenHeartbeatDAO;
import concert.domain.waitingqueue.services.WaitingQueueService;
import concert.infrastructure.distributedlock.DistributedLockAop;
import concert.infrastructure.distributedlock.dao.TokenConcertScheduleSeatDAO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RMapCache;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

@Component
@RequiredArgsConstructor
@Slf4j
public class InactiveUserCleanerScheduler {

    private final WaitingQueueService waitingQueueService;
    private final TokenHeartbeatDAO tokenHeartbeatDAO;
    private final TokenConcertScheduleSeatDAO tokenConcertScheduleSeatDAO;
    private final DistributedLockAop distributedLockAop;
    private static final long INACTIVITY_THRESHOLD = 30 * 1000; // 30초 (밀리초 단위)


    @Scheduled(fixedRate = 10000)  // 10초마다 실행
    public void detectInactiveUsers() {
        RMapCache<String, String> heartbeatMap = tokenHeartbeatDAO.getTokenHeartbeatMap();

        // 현재 시간을 밀리초 단위로 구함
        long currentTime = System.currentTimeMillis();

        // Map을 순회하면서 이탈한 사용자 감지
        Iterator<String> iterator = heartbeatMap.keySet().iterator();

        List<String> tokensToRemove = new ArrayList<>();

        while (iterator.hasNext()) {
            String token = iterator.next();
            String timestampStr = heartbeatMap.get(token);

            if (timestampStr != null) {
                try {
                    // 타임스탬프를 밀리초로 변환
                    long timestamp = Long.parseLong(timestampStr);

                    // 만약 30초 이상 경과한 사용자라면 이탈한 사용자로 간주하고 삭제
                    if (currentTime - timestamp > INACTIVITY_THRESHOLD) {
                        // 이탈한 사용자 처리 (예: 대기열에서 제거)
                        tokensToRemove.add(token);
                    }
                } catch (NumberFormatException e) {
                    log.error("Wrong timestamp number format", e);
                }
            }
        }

        for (String token : tokensToRemove) {
            log.info("token remove, {}", token);
            heartbeatMap.remove(token);
            waitingQueueService.removeTokenFromQueues(token);

            List<String> concertScheduleSeatIds = tokenConcertScheduleSeatDAO.findConcertScheduleSeatIdsByToken(token);

            for(String concertScheduleSeatId: concertScheduleSeatIds){
                distributedLockAop.unlockConcertScheduleSeat(concertScheduleSeatId);
                tokenConcertScheduleSeatDAO.removeTokenScheduleSeat(concertScheduleSeatId);
            }
        }
    }
}

 

- 스프링 서버에서는 일정 주기(ex) 10초)마다 스케줄러를 실행하면서, 비활성 사용자를 제거합니다. 

  비활성 사용자의 기준은 헬스체크 데이터가 업데이트된지 임계점(ex)30초)을 초과한 사용자를 의미합니다. 

 

- 비활성 사용자를 제거한다는 의미는,

  해당 사용자의 토큰을 기준으로 헬스체크, 대기열, 활성화열, 5분 선점 예약과 관련한 정보를 제거한다는 의미입니다. 

 

 

(4) 스케줄러를 통한 비활성 사용자 제거 상세

 

(4-1) 헬스체크 제거

RMapCache<String, String> heartbeatMap = tokenHeartbeatDAO.getTokenHeartbeatMap();


for (String token : tokensToRemove) {
        log.info("token remove, {}", token);
        heartbeatMap.remove(token);
        waitingQueueService.removeTokenFromQueues(token);

        List<String> concertScheduleSeatIds = tokenConcertScheduleSeatDAO.findConcertScheduleSeatIdsByToken(token);

        for(String concertScheduleSeatId: concertScheduleSeatIds){
           distributedLockAop.unlockConcertScheduleSeat(concertScheduleSeatId);
           tokenConcertScheduleSeatDAO.removeTokenScheduleSeat(concertScheduleSeatId);
        }
 }

 

- Redis에 저장된 HeartbeatMap에서 사용자 토큰에 해당하는 헬스체크 정보를 삭제합니다. 

 

 

(4-1) 대기열&활성화열 제거

  public void removeTokenFromQueues(String token) {
    boolean waitingQueueExists = waitingQueueDAO.isTokenExistsInWaitingQueue(token);

    if(waitingQueueExists){
      waitingQueueDAO.deleteWaitingQueueToken(token);
      return;
    }

    WaitingDTO waitingDTO = WaitingDTO.parse(token);
    boolean activeQueueExists = activeQueueDAO.isTokenExistsInActiveQueue(waitingDTO);

    if(activeQueueExists){
      activeQueueDAO.deleteActiveQueueToken(waitingDTO.getUuid());
    }
  }

 

- 사용자 토큰을 기준으로 대기열 혹은 활성화열에 있는 토큰 정보를 제거합니다. 

 

 

(4-2) 5분 좌석 선점 예약 제거

   public List<String> findConcertScheduleSeatIdsByToken(String token) {
        RMapCache<String, String> tokenScheduleSeat = redisson.getMapCache(RedisKey.TOKEN_CONCERTSCHEDULE_SEAT);

        List<String> concertScheduleSeatIds = new ArrayList<>();

        for (String concertScheduleSeatId : tokenScheduleSeat.keySet()) {
            if (token.equals(tokenScheduleSeat.get(concertScheduleSeatId))) {
                concertScheduleSeatIds.add(concertScheduleSeatId); // 해당 token이 매핑된 concertScheduleSeatId 반환
            }
        }
        return concertScheduleSeatIds; // 없으면 null 반환
  }

for(String concertScheduleSeatId: concertScheduleSeatIds){
     distributedLockAop.unlockConcertScheduleSeat(concertScheduleSeatId);
     tokenConcertScheduleSeatDAO.removeTokenScheduleSeat(concertScheduleSeatId);
}

  public void unlockConcertScheduleSeat(String concertScheduleSeatId) {
     String key = REDISSON_LOCK_PREFIX + "CONCERT_SCHEDULE_SEAT_RESERVATION:" + concertScheduleSeatId;
     RLock rLock = redissonClient.getLock(key);

     if (rLock.isLocked() && rLock.isHeldByCurrentThread()) {
         rLock.unlock();
         log.info("Unlocked seat reservation for concertScheduleSeatId: {}", concertScheduleSeatId);
     } else {
            log.warn("No active lock found for concertScheduleSeatId: {}", concertScheduleSeatId);
     }
 }

 public void removeTokenScheduleSeat(String concertScheduleSeatId) {
     RMapCache<String, String> tokenScheduleSeat = redisson.getMapCache(RedisKey.TOKEN_CONCERTSCHEDULE_SEAT);
     tokenScheduleSeat.remove(concertScheduleSeatId);
     log.info("Removed token mapping for concertScheduleSeatId: {}", concertScheduleSeatId);
 }

 

- 사용자가 좌석을 선택하면 5분 좌석 선점 예약이 시작됩니다. 

  이 때, 비활성화된 사용자의 토큰에 해당하는 모든 5분 좌석 선점 예약을 삭제합니다.

 

- 5분 좌석 선점 예약은 Redis 분산락으로 구현되었기 때문에, 해당 Redis 분산락을 모두 unlock 처리합니다. 

  한 사용자는 여러 개의 좌석을 5분 선점 예약할 수 있기 때문에, 삭제되는 Redis 분산락도 복수 개 일 수 있습니다.