Skip to content

Commit 7b2d100

Browse files
thezz9thezz9
committed
refactor: Application idle 상태 시 Consumer 사라지는 현상 해결
1 parent 837c5d3 commit 7b2d100

File tree

4 files changed

+144
-31
lines changed

4 files changed

+144
-31
lines changed

‎src/main/java/org/ezcode/codetest/infrastructure/event/config/RedisStreamConfig.java‎

Lines changed: 14 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,7 @@
11
package org.ezcode.codetest.infrastructure.event.config;
22

3-
import java.net.InetAddress;
4-
import java.net.UnknownHostException;
53
import java.time.Duration;
64
import java.util.Map;
7-
import java.util.UUID;
85
import java.util.concurrent.Executor;
96

107
import org.ezcode.codetest.infrastructure.event.listener.RedisJudgeQueueConsumer;
@@ -14,10 +11,8 @@
1411
import org.springframework.data.redis.connection.RedisConnectionFactory;
1512
import org.springframework.data.redis.connection.stream.MapRecord;
1613
import org.springframework.data.redis.connection.stream.ReadOffset;
17-
import org.springframework.data.redis.connection.stream.StreamOffset;
1814
import org.springframework.data.redis.core.RedisTemplate;
1915
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
20-
import org.springframework.data.redis.connection.stream.Consumer;
2116

2217
import jakarta.annotation.PostConstruct;
2318
import lombok.RequiredArgsConstructor;
@@ -30,6 +25,9 @@ public class RedisStreamConfig {
3025

3126
private final RedisTemplate<String, String> redisTemplate;
3227
private final Executor consumerExecutor;
28+
private final RedisStreamConsumerRegistrar consumerRegistrar;
29+
30+
private StreamMessageListenerContainer<String, MapRecord<String, String, String>> container;
3331

3432
@PostConstruct
3533
public void initConsumerGroup() {
@@ -39,7 +37,7 @@ public void initConsumerGroup() {
3937
if (Boolean.FALSE.equals(exists)) {
4038
log.info("Redis Stream 'judge-queue'를 생성합니다.");
4139
redisTemplate.opsForStream().add("judge-queue", Map.of(
42-
"emitterKey", "dummy",
40+
"sessionKey", "dummy",
4341
"problemId", "0",
4442
"languageId", "0",
4543
"userId", "0",
@@ -53,21 +51,20 @@ public void initConsumerGroup() {
5351
connection.xGroupDelConsumer(
5452
"judge-queue".getBytes(),
5553
"judge-group",
56-
getConsumerName().replace("consumer-", "")
54+
consumerRegistrar.getConsumerName()
5755
);
5856
return null;
5957
});
6058
} catch (Exception e) {
61-
log.warn("DELCONSUMER 중 오류 발생: {}", e.getMessage());
59+
log.warn("기존 컨슈머 삭제 중 오류 발생: {}", e.getMessage());
6260
}
6361

6462
redisTemplate.opsForStream().createGroup("judge-queue", ReadOffset.latest(), "judge-group");
6563

6664
log.info("Redis Stream 'judge-queue'에 대한 Consumer Group 'judge-group'을 생성했습니다.");
6765
} catch (Exception e) {
68-
log.info("예외 발생: {}, 메시지: {}", e.getClass(), e.getMessage());
6966
if (e.getCause() instanceof io.lettuce.core.RedisBusyException) {
70-
log.info("Redis Consumer Group 'judge-group'이 이미 존재하여 생성을 건너뜁니다.");
67+
log.info("이미 존재하는 Consumer Group이므로 생성을 생략합니다.");
7168
} else {
7269
log.error("Redis Consumer Group 초기화에 실패했습니다.", e);
7370
throw e;
@@ -80,34 +77,20 @@ public StreamMessageListenerContainer<String, MapRecord<String, String, String>>
8077
RedisConnectionFactory factory,
8178
RedisJudgeQueueConsumer consumer
8279
) {
83-
StreamMessageListenerContainer
84-
.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
85-
StreamMessageListenerContainer
80+
var options = StreamMessageListenerContainer
8681
.StreamMessageListenerContainerOptions
8782
.builder()
8883
.executor(consumerExecutor)
8984
.pollTimeout(Duration.ofSeconds(2))
85+
.errorHandler(e ->
86+
log.error("[Redis Listener] 예외 발생 - 컨테이너가 죽었을 수 있음", e))
9087
.build();
9188

92-
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
93-
StreamMessageListenerContainer.create(factory, options);
94-
95-
container.receive(
96-
Consumer.from("judge-group", getConsumerName()),
97-
StreamOffset.create("judge-queue", ReadOffset.lastConsumed()),
98-
consumer
99-
);
89+
this.container = StreamMessageListenerContainer.create(factory, options);
10090

101-
container.start();
102-
return container;
103-
}
91+
consumerRegistrar.registerConsumer(this.container, consumer);
92+
this.container.start();
10493

105-
private String getConsumerName() {
106-
try {
107-
return "consumer-" + InetAddress.getLocalHost().getHostName();
108-
} catch (UnknownHostException e) {
109-
log.warn("호스트명 확인 실패, UUID 사용: {}", e.getMessage());
110-
return "consumer-" + UUID.randomUUID().toString().substring(0, 8);
111-
}
94+
return this.container;
11295
}
11396
}

‎src/main/java/org/ezcode/codetest/infrastructure/event/config/RedisStreamConsumerRegistrar.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
package org.ezcode.codetest.infrastructure.event.config;
2+
3+
import java.net.InetAddress;
4+
import java.net.UnknownHostException;
5+
import java.util.UUID;
6+
7+
import org.ezcode.codetest.infrastructure.event.listener.RedisJudgeQueueConsumer;
8+
import org.springframework.data.redis.connection.stream.Consumer;
9+
import org.springframework.data.redis.connection.stream.MapRecord;
10+
import org.springframework.data.redis.connection.stream.ReadOffset;
11+
import org.springframework.data.redis.connection.stream.StreamOffset;
12+
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
13+
import org.springframework.stereotype.Component;
14+
15+
import lombok.Getter;
16+
import lombok.extern.slf4j.Slf4j;
17+
18+
@Slf4j
19+
@Getter
20+
@Component
21+
public class RedisStreamConsumerRegistrar {
22+
23+
private final String consumerName;
24+
25+
public RedisStreamConsumerRegistrar() {
26+
this.consumerName = initConsumerName();
27+
}
28+
29+
public void registerConsumer(StreamMessageListenerContainer<String, MapRecord<String, String, String>> container,
30+
RedisJudgeQueueConsumer consumer) {
31+
container.receive(
32+
Consumer.from("judge-group", consumerName),
33+
StreamOffset.create("judge-queue", ReadOffset.lastConsumed()),
34+
consumer
35+
);
36+
}
37+
38+
private String initConsumerName() {
39+
try {
40+
return "consumer-" + InetAddress.getLocalHost().getHostName();
41+
} catch (UnknownHostException e) {
42+
log.warn("호스트명 확인 실패, UUID 사용: {}", e.getMessage());
43+
return "consumer-" + UUID.randomUUID().toString().substring(0, 8);
44+
}
45+
}
46+
}

‎src/main/java/org/ezcode/codetest/infrastructure/event/scheduler/RedisStreamMonitor.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
package org.ezcode.codetest.infrastructure.event.scheduler;
2+
3+
import org.springframework.data.redis.connection.stream.MapRecord;
4+
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
5+
import org.springframework.scheduling.annotation.Scheduled;
6+
import org.springframework.stereotype.Component;
7+
8+
import lombok.RequiredArgsConstructor;
9+
import lombok.extern.slf4j.Slf4j;
10+
11+
@Slf4j
12+
@Component
13+
@RequiredArgsConstructor
14+
public class RedisStreamMonitor {
15+
16+
private final StreamMessageListenerContainer<String, MapRecord<String, String, String>> container;
17+
18+
// Redis Stream 컨테이너가 죽었는지 감시하고, 죽어 있으면 재시작 + 컨슈머 재등록
19+
@Scheduled(fixedDelay = 300_000)
20+
public void monitor() {
21+
if (!container.isRunning()) {
22+
log.warn("[Redis Listener] 죽어 있음 -> 컨테이너 재시작 시도");
23+
24+
try {
25+
container.start();
26+
log.info("[Redis Listener] 컨테이너 재시작 완료");
27+
} catch (Exception e) {
28+
log.error("[Redis Listener] 재시작 실패", e);
29+
}
30+
}
31+
}
32+
}

‎src/main/java/org/ezcode/codetest/infrastructure/redis/CommonRedisConfig.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
package org.ezcode.codetest.infrastructure.redis;
2+
3+
import org.springframework.beans.factory.annotation.Value;
4+
import org.springframework.context.annotation.Bean;
5+
import org.springframework.context.annotation.Configuration;
6+
import org.springframework.context.annotation.Primary;
7+
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
8+
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
9+
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
10+
11+
import io.lettuce.core.ClientOptions;
12+
import io.lettuce.core.SocketOptions;
13+
import io.lettuce.core.resource.DefaultClientResources;
14+
15+
@Configuration
16+
public class CommonRedisConfig {
17+
18+
@Value("${spring.data.redis.host}")
19+
private String redisHost;
20+
21+
@Value("${spring.data.redis.port}")
22+
private int redisPort;
23+
24+
@Value("${spring.data.redis.password}")
25+
private String redisPassword;
26+
27+
@Bean
28+
@Primary
29+
public LettuceConnectionFactory redisConnectionFactory() {
30+
DefaultClientResources resources = DefaultClientResources.create();
31+
ClientOptions clientOptions = ClientOptions.builder()
32+
.autoReconnect(true)
33+
.pingBeforeActivateConnection(true)
34+
.socketOptions(SocketOptions.builder()
35+
.keepAlive(true)
36+
.build())
37+
.build();
38+
39+
LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
40+
.clientResources(resources)
41+
.clientOptions(clientOptions)
42+
.build();
43+
44+
RedisStandaloneConfiguration redisConfig =
45+
new RedisStandaloneConfiguration(redisHost, redisPort);
46+
47+
redisConfig.setPassword(redisPassword);
48+
49+
return new LettuceConnectionFactory(redisConfig, clientConfig);
50+
}
51+
52+
}

0 commit comments

Comments
 (0)