Skip to content

Commit 3ae774c

Browse files
SimoKiiSimoKii
committed
Feat: Upbit WebSocket 다중 prefix 병렬 구독 적용
Made-with: Cursor
1 parent 17ae860 commit 3ae774c

2 files changed

Lines changed: 71 additions & 47 deletions

File tree

‎src/main/java/com/rabbittick/streamer/connector/UpbitConnector.java‎

Lines changed: 66 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -41,9 +41,10 @@
4141
* <p>주요 책임:
4242
* <ul>
4343
* <li>Upbit WebSocket 연결 생명주기 관리</li>
44-
* <li>실시간 ticker 및 trade 데이터 수신 및 파싱</li>
44+
* <li>실시간 ticker, trade, orderbook 데이터 수신 및 파싱</li>
4545
* <li>연결 실패 시 자동 재연결</li>
4646
* <li>60초 주기 Ping/Pong 메커니즘으로 연결 유지</li>
47+
* <li>마켓코드를 100개씩 청크로 나눠 병렬 WebSocket 연결 유지</li>
4748
* </ul>
4849
*
4950
* <p>애플리케이션 시작 완료 후 자동으로 WebSocket 연결을 시작하며,
@@ -61,12 +62,19 @@ public class UpbitConnector implements DisposableBean {
6162

6263
private static final URI UPBIT_WEBSOCKET_URI = URI.create("wss://api.upbit.com/websocket/v1");
6364
private static final Duration PING_INTERVAL = Duration.ofSeconds(60);
65+
66+
/** Upbit WebSocket 연결당 최대 구독 가능 마켓코드 수 */
67+
private static final int MAX_MARKETS_PER_CONNECTION = 100;
6468

6569
/** 블로킹 처리 시 동시에 구독할 내부 Mono 개수 상한 (boundedElastic 스레드 풀 사용). */
6670
private static final int PROCESSING_CONCURRENCY = 32;
6771

72+
/** 구독할 마켓 prefix 목록 (KRW, BTC, USDT 등) */
73+
@Value("${upbit.websocket.market-prefixes:KRW}")
74+
private List<String> marketPrefixes;
75+
6876
/**
69-
* 사용할 마켓코드 환경 설정 (development, production, full)
77+
* 사용할 마켓코드 환경 설정 (development, production, full, fetched)
7078
*/
7179
@Value("${upbit.websocket.environment:development}")
7280
private String marketEnvironment;
@@ -111,12 +119,12 @@ public void startWebSocketConnection(ApplicationReadyEvent event) {
111119
log.debug("설정 로드 확인 - tickerEnabled: {}, tradeEnabled: {}, orderbookEnabled: {}",
112120
tickerEnabled, tradeEnabled, orderbookEnabled);
113121
List<String> marketCodes = loadMarketCodes();
114-
log.info("애플리케이션 준비 완료, {} 환경으로 {} 개 마켓코드 WebSocket 연결을 시작합니다.",
115-
marketEnvironment, marketCodes.size());
122+
int chunkCount = (int) Math.ceil((double) marketCodes.size() / MAX_MARKETS_PER_CONNECTION);
123+
log.info("애플리케이션 준비 완료, {} 환경으로 {} 개 마켓코드 → {} 개 WebSocket 연결을 시작합니다.",
124+
marketEnvironment, marketCodes.size(), chunkCount);
116125
log.info("활성화된 데이터 타입 - Ticker: {}, Trade: {}, OrderBook: {}",
117126
tickerEnabled, tradeEnabled, orderbookEnabled);
118-
log.debug("구독할 마켓코드: {}", marketCodes);
119-
connectToUpbit();
127+
connectToUpbit(marketCodes);
120128
}
121129

122130
/**
@@ -127,13 +135,15 @@ public void startWebSocketConnection(ApplicationReadyEvent event) {
127135
*/
128136
private List<String> loadMarketCodes() {
129137
List<String> marketCodes = new ArrayList<>();
130-
List<String> enabledTiers = getEnabledTiers();
131138

132-
for (String tier : enabledTiers) {
133-
List<String> tierMarkets = loadTierMarkets(tier);
134-
if (!tierMarkets.isEmpty()) {
135-
marketCodes.addAll(tierMarkets);
136-
log.debug("티어 '{}' 마켓코드 {} 개 추가: {}", tier, tierMarkets.size(), tierMarkets);
139+
for (String prefix : marketPrefixes) {
140+
List<String> enabledTiers = getEnabledTiers();
141+
for (String tier : enabledTiers) {
142+
List<String> tierMarkets = loadTierMarkets(prefix.toLowerCase(), tier);
143+
if (!tierMarkets.isEmpty()) {
144+
marketCodes.addAll(tierMarkets);
145+
log.debug("마켓 '{}' 티어 '{}' 마켓코드 {}개 추가", prefix, tier, tierMarkets.size());
146+
}
137147
}
138148
}
139149

@@ -142,21 +152,23 @@ private List<String> loadMarketCodes() {
142152
log.warn("설정에서 마켓코드를 로드할 수 없어 기본값 사용: {}", marketCodes);
143153
}
144154

155+
log.info("전체 마켓코드 {}개 로드 완료 (prefix: {})", marketCodes.size(), marketPrefixes);
145156
return marketCodes;
146157
}
147158

148159
/**
149160
* Environment를 통해 특정 티어의 마켓코드를 동적으로 로드한다.
150161
*
162+
* @param marketPrefix 마켓 prefix (krw, btc, usdt)
151163
* @param tier 로드할 티어 (tier1, tier2, tier3)
152164
* @return 해당 티어의 마켓코드 목록
153165
*/
154-
private List<String> loadTierMarkets(String tier) {
166+
private List<String> loadTierMarkets(String marketPrefix, String tier) {
155167
List<String> markets = new ArrayList<>();
156168
int index = 0;
157169

158170
while (true) {
159-
String propertyKey = String.format("markets.krw.%s[%d]", tier, index);
171+
String propertyKey = String.format("markets.%s.%s[%d]", marketPrefix, tier, index);
160172
String market = env.getProperty(propertyKey);
161173

162174
if (market == null) {
@@ -167,7 +179,7 @@ private List<String> loadTierMarkets(String tier) {
167179
index++;
168180
}
169181

170-
log.debug("티어 '{}' 로드 완료: {} 개 마켓코드", tier, markets.size());
182+
log.debug("마켓 '{}' 티어 '{}' 로드 완료: {} 개 마켓코드", marketPrefix, tier, markets.size());
171183
return markets;
172184
}
173185

@@ -208,33 +220,46 @@ private List<String> getEnabledTiers() {
208220
* <p>연결 실패 시 지수 백오프 방식으로 자동 재연결을 시도한다.
209221
* 최대 백오프 시간은 1분으로 제한된다.
210222
*/
211-
private void connectToUpbit() {
212-
this.connectionDisposable = webSocketClient
213-
.execute(UPBIT_WEBSOCKET_URI, this::handleWebSocketSession)
214-
.retryWhen(Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(5))
215-
.maxBackoff(Duration.ofMinutes(1))
216-
.doBeforeRetry(retrySignal ->
217-
log.warn("Upbit WebSocket 연결 재시도 중... (시도 횟수: {})",
218-
retrySignal.totalRetries() + 1)))
223+
private void connectToUpbit(List<String> marketCodes) {
224+
List<List<String>> chunks = partition(marketCodes, MAX_MARKETS_PER_CONNECTION);
225+
log.info("총 {} 개 청크로 병렬 연결 시작", chunks.size());
226+
227+
this.connectionDisposable = Flux.fromIterable(chunks)
228+
.flatMap(chunk -> webSocketClient
229+
.execute(UPBIT_WEBSOCKET_URI, session -> handleWebSocketSession(session, chunk))
230+
.retryWhen(Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(5))
231+
.maxBackoff(Duration.ofMinutes(1))
232+
.doBeforeRetry(retrySignal ->
233+
log.warn("WebSocket 연결 재시도 중 (마켓코드 {}개, 시도: {})",
234+
chunk.size(), retrySignal.totalRetries() + 1)))
235+
.doOnError(error -> log.error("WebSocket 연결 복구 불가능한 오류 (마켓코드 {}개)", chunk.size(), error)))
219236
.subscribe(
220237
null,
221-
error -> log.error("Upbit WebSocket 연결에 복구 불가능한 오류 발생", error),
222-
() -> log.info("Upbit WebSocket 연결이 정상적으로 종료되었습니다.")
238+
error -> log.error("전체 WebSocket 연결에 복구 불가능한 오류 발생", error),
239+
() -> log.info("전체 WebSocket 연결이 정상적으로 종료되었습니다.")
223240
);
224241
}
225242

243+
private <T> List<List<T>> partition(List<T> list, int chunkSize) {
244+
List<List<T>> chunks = new ArrayList<>();
245+
for (int i = 0; i < list.size(); i += chunkSize) {
246+
chunks.add(list.subList(i, Math.min(i + chunkSize, list.size())));
247+
}
248+
return chunks;
249+
}
250+
226251
/**
227252
* WebSocket 세션을 처리한다.
228253
* 구독 메시지 전송, Ping/Pong 메커니즘, 데이터 수신을 담당한다.
229254
*
230255
* @param session WebSocket 세션
256+
* @param marketCodes 구독할 마켓코드 목록
231257
* @return 세션 처리 완료를 나타내는 Mono
232258
*/
233-
private Mono<Void> handleWebSocketSession(WebSocketSession session) {
234-
log.info("Upbit WebSocket 연결 성공, 데이터 구독을 시작합니다.");
259+
private Mono<Void> handleWebSocketSession(WebSocketSession session, List<String> marketCodes) {
260+
log.info("Upbit WebSocket 연결 성공 (마켓코드 {}개), 데이터 구독을 시작합니다.", marketCodes.size());
235261

236-
// 구독 메시지 전송
237-
Mono<Void> sendSubscription = sendSubscriptionMessage(session);
262+
Mono<Void> sendSubscription = sendSubscriptionMessage(session, marketCodes);
238263

239264
// 60초 주기 Ping 메시지 전송
240265
Mono<Void> keepAlive = sendPeriodicPing(session);
@@ -250,12 +275,13 @@ private Mono<Void> handleWebSocketSession(WebSocketSession session) {
250275
* Upbit WebSocket에 데이터 구독 메시지를 전송한다.
251276
*
252277
* @param session WebSocket 세션
278+
* @param marketCodes 구독할 마켓코드 목록
253279
* @return 메시지 전송 완료를 나타내는 Mono
254280
*/
255-
private Mono<Void> sendSubscriptionMessage(WebSocketSession session) {
281+
private Mono<Void> sendSubscriptionMessage(WebSocketSession session, List<String> marketCodes) {
256282
try {
257-
String subscriptionMessage = createSubscriptionMessage();
258-
log.debug("구독 메시지 전송: {}", subscriptionMessage);
283+
String subscriptionMessage = createSubscriptionMessage(marketCodes);
284+
log.debug("구독 메시지 전송 (마켓코드 {}개)", marketCodes.size());
259285

260286
return session.send(Mono.just(session.textMessage(subscriptionMessage)));
261287
} catch (Exception e) {
@@ -296,11 +322,13 @@ private Mono<Void> receiveAndProcessMessages(WebSocketSession session) {
296322
return session.receive()
297323
.map(WebSocketMessage::getPayloadAsText)
298324
.flatMap(jsonMessage ->
299-
parseMessage(jsonMessage).subscribeOn(Schedulers.boundedElastic()),
325+
parseMessage(jsonMessage)
326+
.subscribeOn(Schedulers.boundedElastic())
327+
.onErrorResume(error -> {
328+
log.warn("메시지 처리 실패, 다음 메시지 계속 처리: {}", error.getMessage());
329+
return Mono.empty();
330+
}),
300331
PROCESSING_CONCURRENCY)
301-
.doOnError(error -> log.error("메시지 처리 중 오류 발생", error))
302-
.onErrorContinue((error, obj) ->
303-
log.warn("메시지 처리 실패, 다음 메시지 계속 처리: {}", error.getMessage()))
304332
.then();
305333
}
306334

@@ -314,7 +342,6 @@ private Mono<Void> parseMessage(String jsonMessage) {
314342
log.debug("WebSocket 메시지 수신: {}", jsonMessage);
315343

316344
try {
317-
// 먼저 메시지에서 type 필드를 확인
318345
JsonNode messageNode = objectMapper.readTree(jsonMessage);
319346
JsonNode typeNode = messageNode.get("ty");
320347

@@ -498,18 +525,14 @@ private Mono<Void> processOrderBookData(UpbitOrderBookDto upbitDto) {
498525
* <li>orderbook: 호가 정보</li>
499526
* </ul>
500527
*
528+
* @param marketCodes 구독할 마켓코드 목록
501529
* @return JSON 형식의 구독 메시지
502530
* @throws JsonProcessingException JSON 직렬화 실패 시
503531
* @throws IllegalStateException 활성화된 데이터 타입이 없는 경우
504532
*/
505-
private String createSubscriptionMessage() throws JsonProcessingException {
506-
List<String> marketCodes = loadMarketCodes();
533+
private String createSubscriptionMessage(List<String> marketCodes) throws JsonProcessingException {
507534
List<Object> request = new ArrayList<>();
508-
509-
// Ticket 정보
510535
request.add(new Ticket(UUID.randomUUID().toString()));
511-
512-
// 활성화된 데이터 타입별로 구독 요청 추가
513536
if (tickerEnabled) {
514537
request.add(new Type("ticker", marketCodes));
515538
log.info("Ticker 구독 활성화: {} 개 마켓코드", marketCodes.size());
@@ -525,12 +548,9 @@ private String createSubscriptionMessage() throws JsonProcessingException {
525548
log.info("OrderBook 구독 활성화: {} 개 마켓코드", marketCodes.size());
526549
}
527550

528-
// 구독할 데이터 타입이 없는 경우 예외 발생
529551
if (!tickerEnabled && !tradeEnabled && !orderbookEnabled) {
530552
throw new IllegalStateException("최소 하나의 데이터 타입은 활성화되어야 합니다");
531553
}
532-
533-
// Format 정보
534554
request.add(new Format("SIMPLE"));
535555

536556
return objectMapper.writeValueAsString(request);

‎src/main/resources/application.yml‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,13 +19,17 @@ upbit:
1919
websocket:
2020
uri: wss://api.upbit.com/websocket/v1
2121
environment: fetched
22+
market-prefixes:
23+
- KRW
24+
- BTC
25+
- USDT
2226
data-types:
2327
ticker:
2428
enabled: true
2529
trade:
2630
enabled: true
2731
orderbook:
28-
enabled: false
32+
enabled: true
2933

3034
rabbitmq:
3135
exchange:

0 commit comments

Comments
 (0)