거래소 연동 개발을 하다 보면 반드시 마주치는 벽이 있습니다. REST API의 폴링 방식으로는 실시간 시세 변동에 대응할 수 없기 때문입니다. 2024년 초, 저는 고빈도 알트코인 스캘핑 봇을 개발하면서 Binance WebSocket의 연결 끊김 문제를 해결하지 못해 2시간 만에 340달러相当의 거래 기회를 놓친 경험이 있습니다.

이번 튜토리얼에서는 Binance WebSocket Stream의 실제 연결 방법부터 데이터 파싱, 재연결 로직, 그리고 프로덕션 환경에서 자주 발생하는 오류들을 체계적으로 다룹니다. Python과 JavaScript 양쪽 예제를 제공하므로 어떤 언어를 사용하시든 바로 적용하실 수 있습니다.

Binance WebSocket 개요

Binance는 세계 최대 암호화폐 거래소로서 초당 수천 건의 메시지를 처리하는 WebSocket 스트림을 제공합니다. REST API의 요청-응답 방식과 달리 WebSocket은 서버가 먼저 클라이언트에게 데이터를 푸시하므로 지연 시간을 최소화할 수 있습니다.

주요 스트림 유형

연결 엔드포인트

용도 URL 프로토콜
실시간 거래 데이터 wss://stream.binance.com:9443/ws WSS
콤바인드 스트림 wss://stream.binance.com:9443/stream WSS
테스트넷 wss://testnet.binance.vision/ws WSS

Python으로 Binance WebSocket 연동하기

Python에서는 websockets 라이브러리 또는 binance-connector 라이브러리를 주로 사용합니다. 저는 두 가지 방법을 모두 보여드리겠습니다.

방법 1: websockets 라이브러리 (저수준 제어)

import asyncio
import json
import websockets
from datetime import datetime

class BinanceWebSocketClient:
    def __init__(self):
        self.url = "wss://stream.binance.com:9443/ws"
        self.streams = ["btcusdt@trade", "ethusdt@trade", "bnbusdt@trade"]
        self.last_ping = datetime.now()
        self.reconnect_delay = 1
        self.max_reconnect_delay = 60
        
    def build_subscribe_message(self):
        """구독 메시지 생성"""
        return json.dumps({
            "method": "SUBSCRIBE",
            "params": self.streams,
            "id": 1
        })
    
    async def on_message(self, message):
        """수신된 메시지 처리"""
        try:
            data = json.loads(message)
            
            # Trade 데이터 파싱
            if "e" in data and data["e"] == "trade":
                symbol = data["s"]
                price = float(data["p"])
                quantity = float(data["q"])
                trade_time = datetime.fromtimestamp(data["T"] / 1000)
                
                print(f"[{trade_time.strftime('%H:%M:%S.%f')[:-3]}] "
                      f"{symbol}: ${price:,.2f} | 수량: {quantity}")
                
            # 심볼 티커 데이터
            elif "e" in data and data["e"] == "24hrTicker":
                symbol = data["s"]
                price_change = float(data["p"])
                price_change_percent = float(data["P"])
                print(f"{symbol}: 변동 ${price_change:,.2f} ({price_change_percent:+.2f}%)")
                
            # Pong 응답 처리
            elif data.get("result") is None and "id" not in data:
                pass  # 구독 확인 메시지
                
        except json.JSONDecodeError:
            print(f"JSON 파싱 오류: {message[:100]}")
        except KeyError as e:
            print(f"필드 누락 오류: {e}")
    
    async def ping_pong_handler(self, websocket):
        """핑퐁 하트비트 유지"""
        while True:
            try:
                await asyncio.sleep(30)  # 30초마다 ping
                await websocket.send(json.dumps({"method": "ping"}))
            except Exception:
                break
    
    async def connect(self):
        """WebSocket 연결 및 재연결 로직"""
        while True:
            try:
                async with websockets.connect(self.url) as ws:
                    print(f"✅ 연결 성공: {self.url}")
                    self.reconnect_delay = 1  # 지수 백오프 초기화
                    
                    # 구독 요청 전송
                    await ws.send(self.build_subscribe_message())
                    print(f"📡 구독 완료: {self.streams}")
                    
                    # 핑퐁 핸들러 동시 실행
                    ping_task = asyncio.create_task(self.ping_pong_handler(ws))
                    
                    # 메시지 수신 루프
                    async for message in ws:
                        await self.on_message(message)
                        
            except websockets.exceptions.ConnectionClosed as e:
                print(f"❌ 연결 끊김 (코드: {e.code}): {e.reason}")
            except OSError as e:
                print(f"🌐 네트워크 오류: {e}")
            except Exception as e:
                print(f"⚠️ 예상치 못한 오류: {type(e).__name__}: {e}")
            
            # 지수 백오프 방식으로 재연결
            print(f"🔄 {self.reconnect_delay}초 후 재연결 시도...")
            await asyncio.sleep(self.reconnect_delay)
            self.reconnect_delay = min(self.reconnect_delay * 2, self.max_reconnect_delay)

async def main():
    client = BinanceWebSocketClient()
    await client.connect()

if __name__ == "__main__":
    asyncio.run(main())

방법 2: binance-connector 라이브러리 (높은 수준 추상화)

# pip install binance-connector

from binance.websocket.websocket_api import BinanceWebsocketAPI
from binance.websocket.websocket_client import WebsocketStreamClient
import time

class AdvancedBinanceClient:
    def __init__(self, api_key=None, api_secret=None):
        self.api_key = api_key
        self.api_secret = api_secret
        self.ws_client = None
        self.callback_count = 0
        
    def message_handler(self, _, message):
        """콜백 기반 메시지 처리"""
        self.callback_count += 1
        message_type = message.get("e", "unknown")
        
        if message_type == "trade":
            # 실시간 거래 데이터
            data = {
                "symbol": message["s"],
                "price": float(message["p"]),
                "quantity": float(message["q"]),
                "time": message["T"],
                "is_buyer_maker": message["m"]
            }
            print(f"거래 #{self.callback_count}: {data}")
            
        elif message_type == "24hrTicker":
            # 24시간 통계
            data = {
                "symbol": message["s"],
                "last_price": float(message["c"]),
                "high": float(message["h"]),
                "low": float(message["l"]),
                "volume": float(message["v"]),
                "quote_volume": float(message["q"])
            }
            print(f"티커: {data}")
            
        elif message_type == "depthUpdate":
            # 호가창 업데이트
            bids = message.get("b", [])
            asks = message.get("a", [])
            print(f"호가창 업데이트 - 매수: {len(bids)} | 매도: {len(asks)}")
            
        elif "result" in message:
            # 구독 확인
            print(f"구독 확인: {message['result']}")
            
    def start_trade_stream(self, symbols):
        """거래 스트림 시작"""
        self.ws_client = WebsocketStreamClient()
        
        for symbol in symbols:
            # 개별 심볼 거래 스트림 구독
            self.ws_client.instant_subscribe(
                symbol=f"{symbol.lower()}@trade",
                callback=self.message_handler
            )
            # 1초봉(kline) 스트림 추가 구독
            self.ws_client.instant_subscribe(
                symbol=f"{symbol.lower()}@kline_1s",
                callback=self.message_handler
            )
            
        print(f"✅ {len(symbols)}개 심볼 구독 완료")
        return self.ws_client
    
    def start_combined_stream(self):
        """콤바인드 스트림 (단일 연결 다중 구독)"""
        self.ws_client = WebsocketAPIClient()
        
        # 24시간 티커 모니터링
        self.ws_client.start()
        self.ws_client.ticker(
            symbol="bnbusdt",
            id=1,
            callback=self.message_handler
        )
        
        # 시스템 상태 모니터링
        self.ws_client.exchange_symbol_list(
            id=2,
            callback=self.message_handler
        )
        
        return self.ws_client
    
    def start_user_stream(self):
        """사용자 데이터 스트림 (API 키 필요)"""
        if not self.api_key:
            raise ValueError("API 키가 필요합니다")
            
        from binance.spot import Spot as Client
        
        # Listen Key 획득
        spot_client = Client(api_key=self.api_key, api_secret=self.api_secret)
        listen_key = spot_client.new_listen_key()
        print(f"Listen Key: {listen_key}")
        
        # 사용자 데이터 스트림 구독
        self.ws_client = WebsocketStreamClient()
        self.ws_client.instant_subscribe(
            symbol=f"{listen_key}",
            callback=self.message_handler
        )
        
        return self.ws_client
    
    def stop(self):
        """연결 종료"""
        if self.ws_client:
            self.ws_client.stop()
            print("연결 종료됨")


사용 예제

if __name__ == "__main__":
    client = AdvancedBinanceClient()
    
    # BTC, ETH, BNB 거래 데이터 모니터링
    symbols = ["BTCUSDT", "ETHUSDT", "BNBUSDT"]
    client.start_trade_stream(symbols)
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        client.stop()

JavaScript/Node.js로 Binance WebSocket 연동하기

Node.js 환경에서는 ws 라이브러리 또는 Binance 공식 SDK를 사용합니다. 실시간 트레이딩 봇이나 대시보드 연동에 특히 유용합니다.

// npm install ws axios

const WebSocket = require('ws');

class BinanceWebSocketManager {
    constructor() {
        this.streams = new Map();
        this.reconnectAttempts = new Map();
        this.maxReconnectAttempts = 10;
        this.baseDelay = 1000;
        this.pingInterval = null;
        this.messageBuffer = [];
        this.bufferSize = 100;
    }
    
    /**
     * 개별 스트림 연결
     * @param {string} symbol - 심볼 (예: btcusdt)
     * @param {string} stream - 스트림 타입 (trade, ticker, depth)
     */
    connectStream(symbol, stream = 'trade') {
        const streamName = ${symbol.toLowerCase()}@${stream};
        const url = wss://stream.binance.com:9443/ws/${streamName};
        
        this.createConnection(url, streamName);
    }
    
    /**
     * 콤바인드 스트림 연결 (다중 구독)
     * @param {Array} streams - ['btcusdt@trade', 'ethusdt@ticker']
     */
    connectCombinedStream(streams) {
        const streamsParam = streams.join('/');
        const url = wss://stream.binance.com:9443/stream?streams=${streamsParam};
        
        this.createConnection(url, 'combined');
    }
    
    createConnection(url, streamKey) {
        console.log(🔌 연결 시도: ${url.substring(0, 60)}...);
        
        const ws = new WebSocket(url);
        const reconnectDelay = this.calculateBackoff(streamKey);
        
        ws.on('open', () => {
            console.log(✅ 연결 성공: ${streamKey});
            this.reconnectAttempts.set(streamKey, 0);
            
            // 30초 핑퐁
            this.pingInterval = setInterval(() => {
                if (ws.readyState === WebSocket.OPEN) {
                    ws.ping();
                }
            }, 30000);
            
            // 버퍼 플러시
            this.flushBuffer(streamKey);
        });
        
        ws.on('message', (data) => {
            try {
                const message = JSON.parse(data.toString());
                this.handleMessage(message, streamKey);
            } catch (error) {
                console.error('메시지 파싱 오류:', error.message);
            }
        });
        
        ws.on('pong', () => {
            console.log('🏓 Pong 수신');
        });
        
        ws.on('close', (code, reason) => {
            console.log(❌ 연결 종료 (코드: ${code}): ${reason || '없음'});
            clearInterval(this.pingInterval);
            this.handleReconnect(url, streamKey);
        });
        
        ws.on('error', (error) => {
            console.error(⚠️ WebSocket 오류 [${streamKey}]:, error.message);
        });
        
        this.streams.set(streamKey, ws);
        return ws;
    }
    
    handleMessage(message, streamKey) {
        // 콤바인드 스트림인 경우 data 필드 추출
        const payload = message.stream ? message.data : message;
        const eventType = payload.e;
        
        switch (eventType) {
            case 'trade':
                this.processTrade(payload);
                break;
            case '24hrTicker':
                this.processTicker(payload);
                break;
            case 'depthUpdate':
                this.processDepth(payload);
                break;
            case 'kline':
                this.processKline(payload);
                break;
            case 'aggTrade':
                this.processAggTrade(payload);
                break;
            default:
                console.log(미처리 이벤트: ${eventType});
        }
    }
    
    processTrade(data) {
        const trade = {
            symbol: data.s,
            price: parseFloat(data.p),
            quantity: parseFloat(data.q),
            time: new Date(data.T).toISOString(),
            isBuyerMaker: data.m
        };
        
        console.log([${trade.time.split('T')[1].split('.')[0]}] 
            + ${trade.symbol}: $${trade.price.toLocaleString()} 
            + (Qty: ${trade.quantity}));
        
        // 거래량 임계값 초과 시 알림
        if (trade.quantity > 1) {
            console.log(🔔 대량 거래 감지: ${trade.symbol});
        }
    }
    
    processTicker(data) {
        const ticker = {
            symbol: data.s,
            priceChange: parseFloat(data.p),
            priceChangePercent: parseFloat(data.P),
            high: parseFloat(data.h),
            low: parseFloat(data.l),
            volume: parseFloat(data.v),
            quoteVolume: parseFloat(data.q)
        };
        
        const emoji = ticker.priceChangePercent >= 0 ? '📈' : '📉';
        console.log(${emoji} ${ticker.symbol}: ${ticker.priceChangePercent.toFixed(2)}%);
    }
    
    processDepth(data) {
        console.log(호가창 업데이트: 매수 ${data.b.length}개, 매도 ${data.a.length}개);
    }
    
    processKline(data) {
        const kline = data.k;
        console.log(${kline.s} 1분봉: O:${kline.o} H:${kline.h} L:${kline.l} C:${kline.c});
    }
    
    processAggTrade(data) {
        // 집계 거래 (중복 제거된 거래)
        console.log(집계거래: ${data.s} @ ${data.p});
    }
    
    calculateBackoff(streamKey) {
        const attempts = this.reconnectAttempts.get(streamKey) || 0;
        const delay = Math.min(this.baseDelay * Math.pow(2, attempts), 60000);
        return delay;
    }
    
    handleReconnect(url, streamKey) {
        const attempts = (this.reconnectAttempts.get(streamKey) || 0) + 1;
        
        if (attempts > this.maxReconnectAttempts) {
            console.error(❌ 최대 재연결 횟수 초과: ${streamKey});
            return;
        }
        
        this.reconnectAttempts.set(streamKey, attempts);
        const delay = this.calculateBackoff(streamKey);
        
        console.log(🔄 ${delay/1000}초 후 재연결 시도 (${attempts}/${this.maxReconnectAttempts})...);
        
        setTimeout(() => {
            this.createConnection(url, streamKey);
        }, delay);
    }
    
    flushBuffer(streamKey) {
        if (this.messageBuffer.length > 0) {
            console.log(버퍼에서 ${this.messageBuffer.length}개 메시지 플러시);
            this.messageBuffer.forEach(msg => this.handleMessage(msg, streamKey));
            this.messageBuffer = [];
        }
    }
    
    closeAll() {
        console.log('모든 연결 종료 중...');
        this.streams.forEach((ws, key) => {
            if (ws.readyState === WebSocket.OPEN) {
                ws.close(1000, 'Client closing');
            }
        });
        this.streams.clear();
        clearInterval(this.pingInterval);
    }
}

// 사용 예제
const manager = new BinanceWebSocketManager();

// 방법 1: 개별 스트림 연결
manager.connectStream('btcusdt', 'trade');
manager.connectStream('ethusdt', 'trade');

// 방법 2: 콤바인드 스트림 연결
manager.connectCombinedStream([
    'bnbusdt@trade',
    'bnbusdt@ticker',
    'bnbusdt@depth@100ms'
]);

//graceful shutdown
process.on('SIGINT', () => {
    console.log('\nGraceful shutdown...');
    manager.closeAll();
    process.exit(0);
});

Binance WebSocket 데이터 구조 이해하기

실제 개발에서 가장 흔한 실수는 데이터 구조를 제대로 이해하지 못해 필드 접근에 실패하는 것입니다. 주요 이벤트 타입별 구조를 정리합니다.

# Python으로 각 스트림 데이터 구조 확인

import json
import asyncio
import websockets

async def explore_data_structure():
    """각 스트림 타입의 데이터 구조를 출력"""
    
    streams = {
        "trade": "btcusdt@trade",
        "ticker": "btcusdt@ticker",
        "kline_1m": "btcusdt@kline_1m",
        "depth_20": "btcusdt@depth20@100ms",
        "agg_trade": "btcusdt@aggTrade"
    }
    
    async with websockets.connect("wss://stream.binance.com:9443/stream") as ws:
        # 콤바인드 스트림 구독
        subscribe_msg = {
            "method": "SUBSCRIBE",
            "params": list(streams.values()),
            "id": 1
        }
        await ws.send(json.dumps(subscribe_msg))
        print("📡 구독 요청 전송\n")
        
        # 5초간 데이터 수집
        for _ in range(20):
            message = await asyncio.wait_for(ws.recv(), timeout=5.0)
            data = json.loads(message)
            
            if "stream" in data:
                stream_name = data["stream"]
                payload = data["data"]
                
                print(f"\n{'='*60}")
                print(f"스트림: {stream_name}")
                print(f"이벤트 타입: {payload.get('e', 'N/A')}")
                print(f"모든 필드: {list(payload.keys())}")
                print(f"원본 데이터: {json.dumps(payload, indent=2)[:500]}")

if __name__ == "__main__":
    asyncio.run(explore_data_structure())

주요 데이터 필드 설명

이벤트 핵심 필드 설명
trade p, q, T, m 가격, 수량, 타임스탬프, 매도자여부
24hrTicker c, h, l, v, q 현재가, 고가, 저가, 거래량, 거래대금
kline o, h, l, c, v 시가, 고가, 저가, 종가, 거래량
depthUpdate b, a, E 매수호가, 매도호가, 이벤트시간
aggTrade p, q, f, l, m 가격, 수량, 첫매칭ID, 마지막매칭ID

자주 발생하는 오류 해결

오류 1: ConnectionError: timeout — 연결 시간 초과

증상: websockets.exceptions.ConnectionClosed: code=1006 또는 ECONNREFUSED

원인: 방화벽 차단, 프록시 설정 오류, Binance 서버 일시 장애

# 해결 방법 1: 타임아웃 및 재시도 로직 추가

import asyncio
import websockets
from websockets.exceptions import InvalidURI, ConnectionClosed

class TimeoutResilientClient:
    def __init__(self):
        self.timeout = 10  # 연결 타임아웃 (초)
        self.max_retries = 5
        
    async def connect_with_timeout(self, url):
        for attempt in range(self.max_retries):
            try:
                print(f"연결 시도 {attempt + 1}/{self.max_retries}...")
                
                # 타임아웃 설정
                ws = await asyncio.wait_for(
                    websockets.connect(
                        url,
                        ping_interval=20,
                        ping_timeout=10,
                        close_timeout=5
                    ),
                    timeout=self.timeout
                )
                return ws
                
            except asyncio.TimeoutError:
                print(f"⏱️ 타임아웃 발생 (시도 {attempt + 1})")
            except InvalidURI as e:
                print(f"❌ URI 오류: {e}")
                raise
            except ConnectionClosed as e:
                print(f"🔌 연결 끊김: {e.code} - {e.reason}")
            except Exception as e:
                print(f"⚠️ 오류: {type(e).__name__}: {e}")
                
            # 지수 백오프
            await asyncio.sleep(min(2 ** attempt, 30))
            
        raise ConnectionError("최대 재시도 횟수 초과")


해결 방법 2: 프록시 우회 (회사망 환경)

import os

PROXY_URL = os.getenv("HTTPS_PROXY") or os.getenv("HTTP_PROXY")

if PROXY_URL:
    import ssl
    ssl_context = ssl.create_default_context()
    
    async with websockets.connect(
        url,
        ssl=ssl_context,
        proxy=PROXY_URL  # websockets>=10.0 supports proxy
    ) as ws:
        print("프록시를 통해 연결됨")
else:
    print("프록시 없음 - 직접 연결")

오류 2: 401 Unauthorized — 사용자 데이터 스트림 접근 실패

증상: {"error": {"code": -2015, "msg": "Invalid API-key, IP, or permissions for action"}}

원인: API 키 권한 부족, IP 화이트리스트 미설정, 만료된 Listen Key

# 해결 방법: Listen Key 갱신 및 권한 확인

from binance.spot import Spot as Client

class AuthenticatedWebSocketClient:
    def __init__(self, api_key, api_secret):
        self.client = Client(api_key=api_key, api_secret=api_secret)
        self.listen_key = None
        self.keepalive_interval = None
        
    def get_listen_key(self):
        """Listen Key 획득 (IP 권한 확인)"""
        try:
            response = self.client.new_listen_key()
            self.listen_key = response.get("listenKey")
            print(f"✅ Listen Key 획득: {self.listen_key[:20]}...")
            return self.listen_key
        except Exception as e:
            error_code = str(e)
            if "-2015" in error_code:
                print("❌ API 키 권한 오류:")
                print("   1. Binance -> API Management에서 'Enable Spot & Margin Trading' 체크")
                print("   2. IP 화이트리스트에 현재 IP 추가")
                print("   3. API 키 비활성화 후 재발급")
            raise
            
    def keep_listen_key_alive(self):
        """Listen Key 60분 갱신 (자동 만료 방지)"""
        import threading
        
        def _refresh():
            while self.listen_key:
                try:
                    # 55분마다 갱신 (만료 60분 전에)
                    import time
                    time.sleep(55 * 60)
                    self.client.keep_listen_key(self.listen_key)
                    print("🔄 Listen Key 갱신됨")
                except Exception as e:
                    print(f"갱신 실패: {e}")
                    break
                    
        thread = threading.Thread(target=_refresh, daemon=True)
        thread.start()
        return thread
        
    def subscribe_user_data(self, callback):
        """사용자 데이터 스트림 구독"""
        listen_key = self.get_listen_key()
        
        # WebSocket 연결
        ws_url = f"wss://stream.binance.com:9443/ws/{listen_key}"
        # ... WebSocket 연결 및 callback 등록
        
        # Listen Key 자동 갱신 시작
        self.keep_listen_key_alive()
        

사용 전 체크리스트

CHECKLIST = """
API 키 설정 전 체크리스트:
□ API Management에서 Spot Trading 권한 활성화
□ IP 화이트리스트에 서버 IP 추가 (또는 unrestricted 선택)
□ API 키/시크릿 정확히 입력되었는지 확인
□ 서버 시간과 Binance 서버 시간 동기화 (ntpdate)
□tf.env.timezone 확인 (일부 서버 시간대 문제)
"""
print(CHECKLIST)

오류 3: 메모리 누수 — 장시간 실행 시 클라이언트 메모리 증가

증상: MemoryError 또는 프로세스 메모리가 계속 증가

원인: 메시지 버퍼 미청소, 콜백 클로저 메모리 누적,老了旧的 구독 해제 안함

# 해결 방법: 메모리 관리 최적화

import gc
import weakref
from collections import deque
from typing import Callable, Optional

class MemoryOptimizedClient:
    def __init__(self, max_buffer_size=1000, gc_interval=60):
        self.max_buffer_size = max_buffer_size
        self.gc_interval = gc_interval
        self.message_buffer = deque(maxlen=max_buffer_size)  # 고정 크기 버퍼
        self.callbacks = []  # 강한 참조 대신弱참조 고려
        self._subscription_ids = set()
        
    def subscribe(self, stream: str, callback: Callable):
        """구독 등록 (weak reference 사용)"""
        # weakref로 콜백 저장 (메모리 누수 방지)
        weak_callback = weakref.ref(callback)
        self.callbacks.append(weak_callback)
        self._subscription_ids.add(stream)
        
        print(f"📡 구독: {stream} (총 {len(self._subscription_ids)}개)")
        
    def on_message(self, message: dict):
        """메시지 처리 (버퍼 제한)"""
        # deque가 maxlen 초과 시 자동 오래된 항목 제거
        self.message_buffer.append(message)
        
        # weak reference 호출
        for weak_cb in self.callbacks[:]:  # 복사본 순회
            cb = weak_cb()
            if cb is not None:
                try:
                    cb(message)
                except Exception as e:
                    print(f"콜백 오류: {e}")
            else:
                # 죽은 참조 제거
                self.callbacks.remove(weak_cb)
                
    def cleanup(self):
        """명시적 메모리 정리"""
        self.message_buffer.clear()
        self.callbacks.clear()
        self._subscription_ids.clear()
        gc.collect()
        print("🧹 메모리 정리 완료")
        
    def get_memory_usage(self):
        """현재 메모리 사용량 확인"""
        import sys
        size = sys.getsizeof(self.message_buffer)
        print(f"버퍼 크기: {len(self.message_buffer)}/{self.max_buffer_size}")
        print(f"메모리: {size / 1024:.2f} KB")
        return size


장시간 실행 시 주기적 GC

import threading
import time

def start_gc_scheduler(interval=60):
    """주기적 가비지 컬렉션 스케줄러"""
    def gc_loop():
        while True:
            time.sleep(interval)
            gc.collect()
            print(f"🔄 GC 실행 - freed: {gc.mem_free_after_collection()} bytes")
            
    thread = threading.Thread(target=gc_loop, daemon=True)
    thread.start()
    return thread


사용 예제

client = MemoryOptimizedClient(max_buffer_size=500)
gc_thread = start_gc_scheduler(interval=30)


실행 중 주기적 메모리 체크

for i in range(100):
    client.on_message({"test": i})
    if i % 50 == 0:
        client.get_memory_usage()

오류 4: 구독 중복 — 동일 스트림 중복 구독

증상: 동일한 데이터가 여러 번 수신되거나 Duplicate stream name 에러

# 해결 방법: 구독 추적 및 중복 방지

import asyncio
import websockets
import json

class SubscriptionManager:
    def __init__(self):
        self.subscribed_streams = set()
        self.pending_subscriptions = set()
        self.ws = None
        
    async def subscribe(self, streams):
        """중복 없이 구독"""
        new_streams = []
        
        for stream in streams:
            if stream not in self.subscribed_streams:
                new_streams.append(stream)
                self.pending_subscriptions.add(stream)
            else:
                print(f"⚠️ 이미 구독됨: {stream}")
                
        if new_streams:
            await self._send_subscribe(new_streams)
            
    async def _send_subscribe(self, streams):
        """구독 메시지 전송"""
        subscribe_msg = {
            "method": "SUBSCRIBE",
            "params": streams,
            "id": int(asyncio.time.time() * 1000)
        }
        
        await self.ws.send(json.dumps(subscribe_msg))
        print(f"📡 구독 요청: {streams}")
        
        # 구독 확인 대기
        try:
            response = await asyncio.wait_for(
                self.ws.recv(),
                timeout=5.0
            )
            data = json.loads(response)
            
            if "result" in data and data["result"] is None:
                # 구독 성공 - pending -> subscribed 이동
                for stream in streams:
                    if stream in self.pending_subscriptions:
                        self.pending_subscriptions.remove(stream)
                        self.subscribed_streams.add(stream)
                print(f"✅ 구독 완료: {len(streams)}개")
                
        except asyncio.TimeoutError:
            print("❌ 구독 확인 타임아웃")
            for stream in streams:
                self.pending_subscriptions.discard(stream)
                
    async def unsubscribe(self, streams):
        """구독 해제"""
        unsubscribe_msg = {
            "method": "UNSUBSCRIBE",
            "params": streams,
            "id": int(asyncio.time.time() * 1000)
        }
        
        await self.ws.send(json.dumps(unsubscribe_msg))
        
        for stream in streams:
            self.subscribed_streams.discard(stream)
            print(f"🔓 구독 해제: {stream}")
            
    def get_subscribed_count(self):
        """현재 구독 수 반환"""
        return len(self.subscribed_streams)
    
    def list_subscriptions(self):
        """구독 목록 출력"""
        print(f"\n📋 구독 목록 ({len(self.subscribed_streams)}개):")
        for stream in sorted(self.subscribed_streams):
            print(f"  - {stream}")

프로덕션 환경 Best Practices

1. 재연결 전략

# 완전한 재연결 핸들러 예시

class RobustReconnectionHandler:
    def __init__(self):
        self.max_consecutive_failures = 5
        self.base_delay = 1
        self.max_delay = 300
        self.failure_count = 0
        self.last_failure_time = None
        
    def get_reconnect_delay(self):
        """적응형 지수 백오프"""
        if self.failure_count >= self.max_consecutive_failures:
            print("⚠️ 연속 실패 과다 - 연결 상태 점검 필요")
            return None
            
        delay = min(
            self.base_delay * (2 ** self.failure_count),
            self.max_delay
        )
        
        # 재시도 간 무작위 지터 추가
        import random
        jitter = random.uniform(0.5, 1.5)
        
        return delay * jitter
        
    def on_failure(self):
        """연결 실패 처리"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        print(f"실패 #{self.failure_count}")
        
    def on_success(self):
        """연결 성공 처리"""
        self.failure_count = 0
        print("✅ 연결 복구

관련 리소스
관련 문서