분산락에서 비동기 스레드풀 실행 거부 예외 식별과 해결
서론
캐시 스탬피드를 발견하고 많은 스레드가 DB를 동시 접근 하는 것을 방지하기 위해 다양한 방식의 해결법을 적용하던 과정에서 분산락을 적용했을 때 게시글을 조회하지 못하는 오류가 발생했습니다.
요약
| 캐시 | 평균 Throughput | 캐시 갱신 시 레디스 CPU 사용량 | 캐시 갱신 시 실행거부 오류 |
| 분산락 개선 전 | 235 | 0.06% | RejectedExecutionExeption 466건 |
| 분산락 개선 후 | 190 | 0.03% | 오류 0건 |
개선 이전보다 ThroughPut이 하락했습니다.
이는 비동기 스레드에서 락 획득 시도를 하는 것을 동기 스레드로 변경했기 때문입니다.
비동기 스레드에서 락 획득 시도를 하면 비동기 스레드 풀 안에 있는 스레드만이 Redis와 통신하지만 현재는 모든 스레드가 Redis와 통신을 시도합니다. 하지만 비동기 스레드에서 락 획득을 시도할시 어느정도 트래픽까지는 속도가 더 빠르겠지만 비동기 스레드풀의 대기열이 포화될정도의 트래픽이 일어나면 나머지는 전부 연결이 거부되는 현상이 발생합니다. 두번째 캐시스탬피드 때 개선 이전은 순간적으로 TPS가 30까지 떨어졌습니다.
현재 두번의 캐시스탬피드까지만 테스트를 실행했지만 만약 테스트를 더 길게 시행했거나 더 많은 트래픽을 시행했다면 개선 이후가 더 안정적인 성능을 자랑합니다.
애초에 락 프리한 전략을 이용하면 이러한 고민을 모두 해결할 수 있습니다. 아래의 글에서 PER전략을 채택 했습니다.
https://jaeiktech.tistory.com/79
증상 요약
1. 캐시 스탬피드 해결
2. 커넥션 획득 시간 50배 감소
3. 레디스 메모리 사용량 1.8배 감소
4. TTL 만료시 비동기 스레드로 캐시 갱신을 할 때 RejectedExecutionExeption 발생
결론 요약
1. 비동기 로직 내부에서 락을 취득하면 비동기 스레드 풀 병목이 발생 가능하다. 락 획득 시도를 비동기 스레드에게 맡기면 안된다.
2. 많은 스레드들이 순차적으로 락이 필요한 상황이 아니라면 waittime을 설정하지 않는 것이 Redis의 CPU리소스를 아끼고 조금 더 빠른 응답 반환이 가능하다.
깨달은 점
1. 캐시 갱신시 비동기 스레드 풀에서 연결 거부가 발생하면 락 획득의 위치를 다시 살펴보아야한다.
2. waittime을 설정하면 RLock에서 채널을 개설한다. 이는 Redis의 CPU에 영향이 가므로 순차적인 락 경합 (추첨 이벤트 등)이 필요한 상황이 아니라면 waittime을 설정하지 않는 것이 유리하다.
구현 코드
Redisson과 비동기 설정
@Configuration
public class RedissonConfig {
@Bean
public RedissonClient redissonClient() {
config.setThreads(8); // 애플리케이션 스레드
config.setNettyThreads(8); // Pub/Sub 발행을 생각하여 Netty 설정
return Redisson.create(config);
}
}
@Bean(name = "cacheRefreshExecutor")
public ThreadPoolTaskExecutor cacheRefreshExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2); // 기본 스레드 수
executor.setMaxPoolSize(8); // 최대 스레드 수
executor.setQueueCapacity(50); // 대기열 크기
executor.setThreadNamePrefix("cache-refresh-");
executor.initialize();
return executor;
}
아래는 캐시 조회 로직입니다. TTL 만료시 캐시 갱신 로직에 접근합니다.
비동기 캐시 갱신 로직에서는 락을 획득하면 2차캐시로부터 글 ID목록을 받아 DB 조회 후 1차캐시에 업데이트 합니다.
public List<PostSimpleDetail> getWeeklyPosts() {
try {
Long ttl = TTL확인(PostCacheFlag.WEEKLY);
List<PostSimpleDetail> currentCache = 캐시반환(PostCacheFlag.WEEKLY);
// TTL이 없거나 만료되었으면 캐시 갱신
if (ttl == null || ttl <= 0) {
cacheRefreshService.asyncRefreshCache(PostCacheFlag.WEEKLY);
}
// 캐시 반환
// 갱신 이후 다음 스레드 부터 갱신된 캐시를 반환 받습니다.
return currentCache;
} catch (Exception e) {
throw new CustomException(ErrorCode.POST_REDIS_WEEKLY_ERROR, e);
}
}
// 비동기 캐시 갱신 로직
@Async("cacheRefreshExecutor")
public void asyncRefreshCache(PostCacheFlag type) {
boolean acquired = 락 획득(PostCacheFlag.WEEKLY);
if (!acquired) {
log.info("다른 스레드가 캐시 갱신 중: type={}", PostCacheFlag.WEEKLY);
return;
}
try {
List<Long> storedPostIds;
storedPostIds = 2차캐시로부터 조회할 게시글 ID 반환(type);
List<PostSimpleDetail> refreshed = DB조회(storedPostIds)
캐시 갱신(type, refreshed);
} catch (Exception e) {
log.error("캐시 갱신 에러: 타입={}", type, e);
} finally {
락 해제(type);
}
}
// RLOCK의 락 획득 로직입니다.
public boolean 락획득(PostCacheFlag type) {
String lockKey = "lock:cache:refresh:" + type.name();
RLock lock = redissonClient.getLock(lockKey);
try {
// waittime 3초 leasetime 5초로 설정했습니다.
boolean acquired = lock.tryLock(3, 5, TimeUnit.SECONDS);
return acquired;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("캐시 갱신 락 획득 중 인터럽트 발생: type={}", type, e);
return false;
}
}
Redisson에 대해서
Redisson의 tryLock()
@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();
long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
// 락 획득 시도
if (ttl == null) {
return true;
}
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(waitTime, unit, threadId);
return false;
}
current = System.currentTimeMillis();
// 락 획득 실패 시 락 채널 구독
// subscribe 내부에서는 내부적으로 세마포어로 동시성제어를 합니다.
// 하나의 채널에 너무 많은 구독을 생성하지 않도록,
// 한번에 하나의 스레드만 subscribe를 할수있도록 기존 구독이 있다면 재사용하는 역할 입니다.
CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
try {
subscribeFuture.get(time, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
if (!subscribeFuture.completeExceptionally(new RedisTimeoutException(
"Unable to acquire subscription lock after " + time + "ms. " +
"Try to increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."))) {
subscribeFuture.whenComplete((res, ex) -> {
if (ex == null) {
unsubscribe(res, threadId);
}
});
}
acquireFailed(waitTime, unit, threadId);
return false;
} catch (ExecutionException e) {
LOGGER.error(e.getMessage(), e);
acquireFailed(waitTime, unit, threadId);
return false;
}
try {
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(waitTime, unit, threadId);
return false;
}
while (true) {
long currentTime = System.currentTimeMillis();
ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
// 락 획득 시도
if (ttl == null) {
return true;
}
// 실패시 락의 TTL 만큼 스레드를 timed_wating으로 만듭니다.
time -= System.currentTimeMillis() - currentTime;
// 대기 시간이 끝나면 밖으로 나옵니다.
if (time <= 0) {
acquireFailed(waitTime, unit, threadId);
return false;
}
// waiting for message
currentTime = System.currentTimeMillis();
if (ttl >= 0 && ttl < time) {
commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(waitTime, unit, threadId);
return false;
}
}
} finally {
// 마지막에는 채널을 구독해제 합니다.
unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
}
// return get(tryLockAsync(waitTime, leaseTime, unit));
}락 획득 실패시 락 채널을 구독합니다. 락의 TTL 만큼 스레드를 timed_waiting으로 만들고 대기시간이 끝날때까지 반복합니다.
대기시간이 끝나면 채널을 구독해제하고 밖으로 나갑니다. 그리고 waittime이 없을 시 구독 자체를 안하는 것을 확인할 수 있습니다.
waittime이란 락 획득에 실패한 스레드가 다음 기회를 노리기 위해 채널을 구독하는 것이고 캐시 스탬피드 상황에서는 waittime을 설정하지 않고 subscribe에 자원을 낭비하는 것을 막는 것이 좋습니다.
Redisson의 unlock()
protected CompletableFuture<Void> unlockInnerAsync(long threadId) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"... Lua script ...",
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; ",
Arrays.asList(getRawName(), getChannelName()),
LockPubSub.UNLOCK_MESSAGE
);
} Lua Script로 채널에 언락 메시지를 보내는것을 알 수 있습니다.
캐시 스탬피드의 상황 확인
커넥션 풀 지표
TTL 만료 시간에도 커넥션 요청이 몰리지 않았습니다.
사실 DB커넥션이 요청이 많아지지 않을 뿐이지 스탬피드 상황에서 에러가 반환된 것을 보면 해결한 것으로 볼 수 없습니다.
레디스의 Hit / Miss 지표
락 적용 이전과 다르게 TTL 만료시간에 Hit가 급 감소하지 않았습니다.
Pub/Sub의 작동
펍섭 명령이발생하여 채널이 생성되어 제대로 락을 수행하고 있다는 것을 알 수 있습니다.
원인 추적
2025-12-25 15:53:05.823 Caused by: java.util.concurrent.RejectedExecutionException:
Task java.util.concurrent.FutureTask@5c211e3a[Not completed,
task = org.springframework.aop.interceptor.AsyncExecutionInterceptor
$$Lambda/0x0000724c090af8a8@254eb0ad] rejected from
org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor$1@2aeef1ca
[Running, pool size = 8, active threads = 8, queued tasks = 50, completed tasks = 120]위는 스택트레이스의 일부입니다. 비동기 스레드 풀이 가득차고 대기열이 꽉차서 에러가 난 것을 확인 할 수 있습니다.
에러를 모니터링하기 위해서 10초마다 캐시 갱신 비동기 스레드 풀의 상태를 로깅하였습니다.
@Component
@Slf4j
public class AsyncExecutorStatusLogger {
private final ThreadPoolTaskExecutor cacheRefreshExecutor;
public AsyncExecutorStatusLogger(@Qualifier("cacheRefreshExecutor") ThreadPoolTaskExecutor cacheRefreshExecutor) {
this.cacheRefreshExecutor = cacheRefreshExecutor;
}
@Scheduled(fixedRate = 10000) // 10초마다 실행
public void monitorExecutor() {
int activeCount = cacheRefreshExecutor.getActiveCount();
int poolSize = cacheRefreshExecutor.getPoolSize();
int queueSize = cacheRefreshExecutor.getThreadPoolExecutor().getQueue().size();
log.info("[Monitor] 활성: {}, 풀크기: {}, 대기열: {}", activeCount, poolSize, queueSize);
}
}
비동기 캐시 갱신 스레드 풀의 로깅
// 테스트 시작 5분 후 첫번째 캐시스탬피드 상황의 비동기 스레드 풀
2025-12-25 06:47:44.753 [scheduling-1] INFO j.b.i.monitor.AsyncExecutorStatusLogger
[trace=none user=anonymous ip=] - [Monitor] 활성: 8, 풀크기: 8, 대기열: 29
// 테스트 시작 10분 후 (끝나기 직전) 두번째 캐시 스탬피드 상황의 비동기 스레드 풀
2025-12-25 06:53:02.623 [scheduling-1] INFO j.b.i.monitor.AsyncExecutorStatusLogger
[trace=none user=anonymous ip=] - [Monitor] 활성: 8, 풀크기: 8, 대기열: 50
2025-12-25 06:52:52.623 [scheduling-1] INFO j.b.i.monitor.AsyncExecutorStatusLogger
[trace=none user=anonymous ip=] - [Monitor] 활성: 8, 풀크기: 8, 대기열: 50이 스레드풀은 8개의 스레드가 동작하며 50개의 스레드까지 대기 할 수 있게 설정했습니다.
첫번째 스탬피드 상황에 29개의 스레드가 대기하는것을 확인할수 있습니다.
두번째 스탬피드 상황에서는 50개의 스레드가 대기하여 대기열이 가득찬 것을 확인할 수 있습니다.
비동기 스레드풀의 병목으로 에러가 반환된 것을 확인하였습니다.
10초마다 로깅하여 정확하지는 않지만
첫번째 스탬피드 상황은 아직 부하가 완벽하게 램프업되지않았고 (초당 요청 150건) 다음 10초에는 병목이 해소된것을 확인하였습니다.
두번째 스탬피드 상황에는 부하가 완벽하게 램프업대어 (초당 요청 300건) 10초가 지나도 대기큐가 줄어들지 않았습니다.
실제로 그래프를 보면 두번째 스탬피드를 감당하지 못하고 TPS가 30으로 곤두박칠 친것을 볼 수 있습니다.
만약 테스트를 더 길게 잡았거나 더 많은 부하가 일어났다면 병목이 아주 심해지고 전체 시스템의 성능도 급격히 하락했을 겁니다.
해결 방안
이 증상의 문제는 캐시갱신을 위한 비동기 큐가 포화되는 것이고 그렇다면 원인은 락을 획득하는 시점에 있습니다.
현재 코드를 다시 보면 비동기 로직 안에서 락을 획득하는 것을 볼 수 있습니다.
그렇기에 락을 획득한 스레드만 비동기 큐에 들어가야해야하고 락은 동기적으로 획득하게 해야합니다.
// 비동기 캐시 갱신 로직
@Async("cacheRefreshExecutor")
public void asyncRefreshCache(PostCacheFlag type) {
boolean acquired = 락 획득(PostCacheFlag.WEEKLY);
if (!acquired) {
log.info("다른 스레드가 캐시 갱신 중: type={}", PostCacheFlag.WEEKLY);
return;
}
try {
List<Long> storedPostIds;
storedPostIds = 2차캐시로부터 조회할 게시글 ID 반환(type);
List<PostSimpleDetail> refreshed = DB조회(storedPostIds)
캐시 갱신(type, refreshed);
} catch (Exception e) {
log.error("캐시 갱신 에러: 타입={}", type, e);
} finally {
락 해제(type);
}
}개선
개선 이전 로직
public List<PostSimpleDetail> getWeeklyPosts() {
try {
Long ttl = TTL확인(PostCacheFlag.WEEKLY);
List<PostSimpleDetail> currentCache = 캐시반환(PostCacheFlag.WEEKLY);
// TTL이 없거나 만료되었으면 캐시 갱신
if (ttl == null || ttl <= 0) {
cacheRefreshService.asyncRefreshCache(PostCacheFlag.WEEKLY);
}
// 캐시 반환
// 갱신 이후 다음 스레드 부터 갱신된 캐시를 반환 받습니다.
return currentCache;
} catch (Exception e) {
throw new CustomException(ErrorCode.POST_REDIS_WEEKLY_ERROR, e);
}
}
개선 이후 로직
public List<PostSimpleDetail> getWeeklyPosts() {
try {
Long ttl = TTL확인(PostCacheFlag.WEEKLY);
List<PostSimpleDetail> currentCache = 캐시반환(PostCacheFlag.WEEKLY);
// TTL이 없거나 만료되었으면 캐시 갱신
if (ttl == null || ttl <= 0) {
boolean acquired = 락 획득(PostCacheFlag.WEEKLY);
if (acquired) {
비동기갱신(PostCacheFlag.WEEKLY);
} else {
log.info("다른 스레드가 캐시 갱신 중: type={}", PostCacheFlag.REALTIME);
}
}
// 캐시 반환
// 갱신 이후 다음 스레드 부터 갱신된 캐시를 반환 받습니다.
return currentCache;
} catch (Exception e) {
throw new CustomException(ErrorCode.POST_REDIS_WEEKLY_ERROR, e);
}
}메인 로직 락을 취득하게하고 비동기 로직을 호출했습니다.
// 비동기 캐시 갱신 로직
@Async("cacheRefreshExecutor")
public void asyncRefreshCache(PostCacheFlag type) {
// 락 획득 로직 삭제
try {
List<Long> storedPostIds;
storedPostIds = 2차캐시로부터 조회할 게시글 ID 반환(type);
List<PostSimpleDetail> refreshed = DB조회(storedPostIds)
캐시 갱신(type, refreshed);
} catch (Exception e) {
log.error("캐시 갱신 에러: 타입={}", type, e);
} finally {
락 해제(type);
}
}
비동기 로직에서는 락 획득 로직을 삭제했습니다.
// RLOCK의 락 획득 로직입니다.
public boolean 락획득(PostCacheFlag type) {
String lockKey = "lock:cache:refresh:" + type.name();
RLock lock = redissonClient.getLock(lockKey);
try {
// waittime을 0초로 변경했습니다.
boolean acquired = lock.tryLock(0, 5, TimeUnit.SECONDS);
return acquired;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("캐시 갱신 락 획득 중 인터럽트 발생: type={}", type, e);
return false;
}
}또한 현재 상황에서는 단 하나의 스레드만 락을 얻으면 되기에 waittime을 삭제하여 채널의 구독을 막아 리소스를 절약하고 락을 획득하지 못한 스레드들은 즉시 이전 캐시를 반환하게 하여 응답속도를 높였습니다.
결과
개선 이후에 펍섭이 실제로 생성되지 않았습니다.
CPU에 부하가 있는것은 아니었지만 펍섭의 삭제로 CPU 사용량을 0.06%에서 0.03%로 절약했습니다.
개선 이전 두번째 스탬피드에서는 비동기 스레드 풀의 병목으로 연결이 거부되어 아이러니하게 Redis의 CPU가 할일을 덜어준 모습니다.
개선 이후에는 두번째 스탬피드에도 병목이 없어 정상적으로 레디스가 CPU를 사용하는 모습을 볼 수 있습니다.
| 캐시 | 평균 Throughput | 캐시 갱신 시 레디스 CPU 사용량 | 캐시 갱신 시 실행거부 오류 |
| 분산락 개선 전 | 235 | 0.06% | RejectedExecutionExeption 466건 |
| 분산락 개선 후 | 190 | 0.03% | 오류 0건 |