작업열 토큰의 만료 이벤트 처리하기 2

1. 스케줄러의 문제점

스케줄링은 간편하게 토큰의 만료 상태를 확인할 수 있는 좋은 방법이지만, 대기열이 존재하지 않을 때에도 동작하는 만큼 비효율적인 부분이 존재한다. 이에 기존의 스케줄러를 사용하지 않고, 하나의 작업열 토큰이 만료될 때에만 토큰을 즉시 이동시킬 방법이 필요했다.

 

---

2. Redis Keyspace Notifications

Redis의 Keyspace Notifications를 사용하면 Redis에서 발생하는 이벤트를 클라이언트에게 알릴 수 있다. 다양한 이벤트 유형을 지원하며 나의 경우 작업열 토큰의 만료 이벤트를 구독할 것이다.

 

먼저 redis-cli에 해당 설정을 입력하자.

 

CONFIG SET notify-keyspace-events Ex

 

RedisConfig

만료 이벤트 발생 시에 메시지를 수신할 리스너를 등록하는 설정이다.
RedisMessageListenerContainer 빈 등록시에는 Executor를 등록하지 않으면 이벤트 수신마다 새로운 스레드를 만들기 때문에 꼭 지정해주자.

 

@Configuration
public class RedisConfig {

    @Bean(name = "redisMessageTaskExecutor")
    public Executor redisMessageTaskExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(2);
        threadPoolTaskExecutor.setMaxPoolSize(4);
        return threadPoolTaskExecutor;
    }

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, RedisExpirationListener redisExpirationListener) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(redisExpirationListener, new ChannelTopic("__keyevent@0__:expired"));
        container.setTaskExecutor(redisMessageTaskExecutor());
        return container;
    }

}

 

RedisExpirationListener

리스너는 다음과 같이 구현해주었다. 이때 받게되는 메시지는 토큰의 키를 의미한다.

 

@Slf4j
@Component
@RequiredArgsConstructor
public class RedisExpirationListener implements MessageListener {

    private static final String TOKEN_PREFIX = "Token:";

    private final WorkingQueueService workingQueueService;

    @Override
    public void onMessage(final Message tokenValue, final byte[] pattern) {
        // 작업열 토큰 확인
        if (tokenValue.toString().startsWith(TOKEN_PREFIX)) {
            log.info("WorkingQueueToken has expired: {}", tokenValue);
            workingQueueService.processQueueTransfer(tokenValue.toString());
        }
    }
}

 

---

3. Distribuited Lock 도입하기

Keyspace Notifications는 이벤트를 구독 중인 모든 인스턴스에서 메시지를 수신하기 때문에 분산 환경에서는 한 인스턴스만 동작하게 분산락을 설정해주어야 한다.

 

RLock은 Redisson 라이브러리에서 제공하는 분산 락 구현체로 락을 획득한 프로세스만 자원에 접근할 수 있게 한다. 다음과 같이 처음 락을 획득한 인스턴스만 메서드를 실행할 수 있게 구현하였다.

 

RedisExpirationListener

@Slf4j
@Component
@RequiredArgsConstructor
public class RedisExpirationListener implements MessageListener {

    private static final String TOKEN_PREFIX = "Token:";

    private final WorkingQueueService workingQueueService;
    private final RedissonClient redissonClient;

    @Override
    public void onMessage(final Message tokenValue, final byte[] pattern) {
        // 작업열 토큰 확인
        if (tokenValue.toString().startsWith(TOKEN_PREFIX)) {
            log.info("WorkingQueueToken has expired: {}", tokenValue);

            String leaderKey = "LeaderKey:" + tokenValue;
            RLock leaderLock = redissonClient.getLock(leaderKey);

            tryToLeader(tokenValue, leaderLock);
        }
    }

    // 리더 선출 시도
    private void tryToLeader(Message tokenValue, RLock leaderLock) {
        try {
            if (leaderLock.tryLock(0, 5, TimeUnit.SECONDS)) {
                try {
                    // 리더로 선출된 인스턴스만 작업 수행
                    log.info("This instance is the leader, processing job..");
                    workingQueueService.processQueueTransfer(tokenValue.toString());
                } finally {
                    if (leaderLock.isHeldByCurrentThread()) {
                        leaderLock.unlock();
                    }
                }
            } else {
                log.info("Failed to acquire lock for key!");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

}

 

tryLock()의 인자는 다음과 같다.

 

- timeout: 락을 획득하기 위해 대기할 최대 시간
- leaseTime: 락을 획득한 후 자동으로 해제될 때까지의 최대 시간
- unit: 시간 값을 설정

 

결과

2개의 인스턴스 실행 시 한쪽에서만 메서드를 실행하는 것을 확인할 수 있다.