🤔 왜 블록체인 API는 느릴까?

🎯 병렬 처리의 핵심 개념

파이썬의 병렬 처리 옵션들

💻 실전 코드: 순차 처리 vs 병렬 처리

기본 설정 및 라이브러리

# 필요한 라이브러리 설치
pip install web3 requests

# 또는 requirements.txt에 추가
web3==6.11.0
requests==2.31.0

import time
import requests
from web3 import Web3
from concurrent.futures import ThreadPoolExecutor, as_completed
import asyncio
import aiohttp
from typing import List, Dict, Any

# 이더리움 노드 연결 (Infura 사용 예시)
INFURA_URL = "https://mainnet.infura.io/v3/YOUR_PROJECT_ID"
w3 = Web3(Web3.HTTPProvider(INFURA_URL))

방법 1: 순차 처리 (기본 방식)

def get_block_sequential(block_numbers: List[int]) -> List[Dict]:
    """
    순차적으로 블록 정보를 가져오는 함수
    """
    results = []
    start_time = time.time()
    
    for block_num in block_numbers:
        try:
            # 블록 정보 가져오기
            block = w3.eth.get_block(block_num)
            
            # 필요한 정보만 추출
            block_info = {
                'number': block['number'],
                'timestamp': block['timestamp'],
                'transactions': len(block['transactions']),
                'gas_used': block['gasUsed'],
                'hash': block['hash'].hex()
            }
            
            results.append(block_info)
            print(f"✅ 블록 {block_num} 완료")
            
        except Exception as e:
            print(f"❌ 블록 {block_num} 에러: {e}")
            results.append(None)
    
    elapsed_time = time.time() - start_time
    print(f"\n⏱️ 총 소요 시간: {elapsed_time:.2f}초")
    print(f"📊 평균 처리 시간: {elapsed_time/len(block_numbers):.2f}초/블록")
    
    return results

# 테스트 실행
block_numbers = list(range(18000000, 18000020))  # 20개 블록
sequential_results = get_block_sequential(block_numbers)

방법 2: ThreadPoolExecutor 사용

def fetch_single_block(block_num: int) -> Dict:
    """
    단일 블록 정보를 가져오는 헬퍼 함수
    """
    try:
        block = w3.eth.get_block(block_num)
        return {
            'number': block['number'],
            'timestamp': block['timestamp'],
            'transactions': len(block['transactions']),
            'gas_used': block['gasUsed'],
            'hash': block['hash'].hex()
        }
    except Exception as e:
        print(f"❌ 블록 {block_num} 에러: {e}")
        return None

def get_block_parallel(block_numbers: List[int], max_workers: int = 10) -> List[Dict]:
    """
    병렬로 블록 정보를 가져오는 함수
    
    Args:
        block_numbers: 가져올 블록 번호 리스트
        max_workers: 동시에 실행할 최대 스레드 수
    """
    results = []
    start_time = time.time()
    
    # ThreadPoolExecutor 생성
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 모든 작업을 제출
        future_to_block = {
            executor.submit(fetch_single_block, block_num): block_num 
            for block_num in block_numbers
        }
        
        # 완료된 작업부터 결과 수집
        for future in as_completed(future_to_block):
            block_num = future_to_block[future]
            try:
                result = future.result()
                results.append(result)
                print(f"✅ 블록 {block_num} 완료")
            except Exception as e:
                print(f"❌ 블록 {block_num} 예외 발생: {e}")
                results.append(None)
    
    elapsed_time = time.time() - start_time
    print(f"\n⏱️ 총 소요 시간: {elapsed_time:.2f}초")
    print(f"📊 평균 처리 시간: {elapsed_time/len(block_numbers):.2f}초/블록")
    print(f"🚀 성능 향상: {45.32/elapsed_time:.2f}배 빠름!")
    
    return results

# 테스트 실행
parallel_results = get_block_parallel(block_numbers, max_workers=10)

방법 3: AsyncIO로 더 효율적으로

async def fetch_block_async(session: aiohttp.ClientSession, 
                            block_num: int, 
                            url: str) -> Dict:
    """
    비동기로 단일 블록 정보를 가져오는 함수
    """
    payload = {
        "jsonrpc": "2.0",
        "method": "eth_getBlockByNumber",
        "params": [hex(block_num), False],
        "id": 1
    }
    
    try:
        async with session.post(url, json=payload) as response:
            data = await response.json()
            
            if 'result' in data and data['result']:
                block = data['result']
                return {
                    'number': int(block['number'], 16),
                    'timestamp': int(block['timestamp'], 16),
                    'transactions': len(block['transactions']),
                    'gas_used': int(block['gasUsed'], 16),
                    'hash': block['hash']
                }
    except Exception as e:
        print(f"❌ 블록 {block_num} 에러: {e}")
        return None

async def get_blocks_async(block_numbers: List[int], url: str) -> List[Dict]:
    """
    비동기로 여러 블록 정보를 가져오는 메인 함수
    """
    start_time = time.time()
    
    # 동시 연결 수 제한 (Rate Limiting 고려)
    connector = aiohttp.TCPConnector(limit=20)
    
    async with aiohttp.ClientSession(connector=connector) as session:
        # 모든 작업을 동시에 실행
        tasks = [
            fetch_block_async(session, block_num, url) 
            for block_num in block_numbers
        ]
        
        results = await asyncio.gather(*tasks)
        
        # 완료된 작업 카운트
        success_count = sum(1 for r in results if r is not None)
        print(f"✅ 성공: {success_count}/{len(block_numbers)}")
    
    elapsed_time = time.time() - start_time
    print(f"\n⏱️ 총 소요 시간: {elapsed_time:.2f}초")
    print(f"📊 평균 처리 시간: {elapsed_time/len(block_numbers):.2f}초/블록")
    
    return results

# 실행 (Jupyter나 일반 스크립트에서)
async_results = asyncio.run(
    get_blocks_async(block_numbers, INFURA_URL)
)

⚙️ 최적화 전략과 베스트 프랙티스

1. 적절한 Worker 수 설정

def find_optimal_workers(block_numbers: List[int], 
                         test_workers: List[int] = [5, 10, 20, 30, 50]) -> Dict:
    """
    최적의 worker 수를 찾는 함수
    """
    results = {}
    
    for workers in test_workers:
        print(f"\n🔍 Testing with {workers} workers...")
        start_time = time.time()
        
        get_block_parallel(block_numbers, max_workers=workers)
        
        elapsed = time.time() - start_time
        results[workers] = elapsed
        print(f"⏱️ {workers} workers: {elapsed:.2f}초")
    
    # 최적값 찾기
    optimal = min(results, key=results.get)
    print(f"\n🎯 최적 worker 수: {optimal} ({results[optimal]:.2f}초)")
    
    return results

# 테스트 실행
optimization_results = find_optimal_workers(block_numbers)

2. 에러 처리와 재시도 로직

import time
from functools import wraps

def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
    """
    실패 시 재시도하는 데코레이터
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        print(f"❌ 최종 실패 after {max_retries} attempts: {e}")
                        raise
                    
                    wait_time = delay * (2 ** attempt)  # 지수 백오프
                    print(f"⚠️ 시도 {attempt + 1} 실패, {wait_time}초 후 재시도...")
                    time.sleep(wait_time)
            
        return wrapper
    return decorator

@retry_on_failure(max_retries=3, delay=1.0)
def fetch_block_with_retry(block_num: int) -> Dict:
    """
    재시도 로직이 포함된 블록 가져오기 함수
    """
    block = w3.eth.get_block(block_num)
    return {
        'number': block['number'],
        'timestamp': block['timestamp'],
        'transactions': len(block['transactions']),
        'gas_used': block['gasUsed'],
        'hash': block['hash'].hex()
    }

# 사용 예시
def get_blocks_with_retry(block_numbers: List[int], max_workers: int = 10):
    """
    재시도 로직이 포함된 병렬 처리
    """
    results = []
    failed_blocks = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_block = {
            executor.submit(fetch_block_with_retry, block_num): block_num 
            for block_num in block_numbers
        }
        
        for future in as_completed(future_to_block):
            block_num = future_to_block[future]
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                failed_blocks.append(block_num)
                results.append(None)
    
    if failed_blocks:
        print(f"\n⚠️ 실패한 블록들: {failed_blocks}")
    
    return results, failed_blocks

3. Rate Limiting 처리

import threading
from collections import deque
from datetime import datetime, timedelta

class RateLimiter:
    """
    Rate Limiting을 구현한 클래스
    """
    def __init__(self, max_calls: int, time_window: float):
        """
        Args:
            max_calls: 시간 창 내 최대 호출 수
            time_window: 시간 창 (초)
        """
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = deque()
        self.lock = threading.Lock()
    
    def __call__(self, func):
        """
        데코레이터로 사용
        """
        @wraps(func)
        def wrapper(*args, **kwargs):
            with self.lock:
                now = datetime.now()
                
                # 시간 창 밖의 호출 기록 제거
                while self.calls and self.calls[0] < now - timedelta(seconds=self.time_window):
                    self.calls.popleft()
                
                # 제한 확인
                if len(self.calls) >= self.max_calls:
                    sleep_time = (self.calls[0] + timedelta(seconds=self.time_window) - now).total_seconds()
                    if sleep_time > 0:
                        print(f"⏸️ Rate limit 도달, {sleep_time:.2f}초 대기...")
                        time.sleep(sleep_time)
                
                # 호출 기록 추가
                self.calls.append(datetime.now())
            
            return func(*args, **kwargs)
        
        return wrapper

# Infura 무료 플랜: 초당 10개 제한
rate_limiter = RateLimiter(max_calls=10, time_window=1.0)

@rate_limiter
def fetch_block_rate_limited(block_num: int) -> Dict:
    """
    Rate limiting이 적용된 블록 가져오기
    """
    return fetch_single_block(block_num)

# 사용
def get_blocks_rate_limited(block_numbers: List[int], max_workers: int = 15):
    """
    Rate limiting을 고려한 병렬 처리
    Worker는 많지만 실제 호출은 제한됨
    """
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_block = {
            executor.submit(fetch_block_rate_limited, block_num): block_num 
            for block_num in block_numbers
        }
        
        for future in as_completed(future_to_block):
            block_num = future_to_block[future]
            try:
                result = future.result()
                results.append(result)
                print(f"✅ 블록 {block_num} 완료")
            except Exception as e:
                print(f"❌ 블록 {block_num} 에러: {e}")
                results.append(None)
    
    return results

4. 배치 처리로 효율성 높이기

def process_in_batches(block_numbers: List[int], 
                       batch_size: int = 100,
                       max_workers: int = 10) -> List[Dict]:
    """
    배치 단위로 블록을 처리하는 함수
    """
    all_results = []
    total_blocks = len(block_numbers)
    
    # 배치로 나누기
    for i in range(0, total_blocks, batch_size):
        batch = block_numbers[i:i + batch_size]
        batch_num = i // batch_size + 1
        total_batches = (total_blocks + batch_size - 1) // batch_size
        
        print(f"\n📦 배치 {batch_num}/{total_batches} 처리 중 ({len(batch)}개 블록)...")
        
        # 배치 처리
        batch_results = get_block_parallel(batch, max_workers=max_workers)
        all_results.extend(batch_results)
        
        # 배치 간 잠깐 쉬기 (선택사항)
        if i + batch_size < total_blocks:
            time.sleep(0.5)
    
    print(f"\n✅ 전체 처리 완료: {len(all_results)}개 블록")
    return all_results

# 1000개 블록을 100개씩 배치로 처리
large_block_range = list(range(18000000, 18001000))
batch_results = process_in_batches(large_block_range, batch_size=100, max_workers=10)

📊 실전 활용 사례

사례 1: 특정 주소의 트랜잭션 히스토리 수집

def get_address_transactions(address: str, 
                            start_block: int, 
                            end_block: int,
                            max_workers: int = 20) -> List[Dict]:
    """
    특정 주소의 트랜잭션을 병렬로 수집
    """
    address = Web3.to_checksum_address(address)
    transactions = []
    
    def check_block_for_address(block_num: int) -> List[Dict]:
        """
        블록에서 해당 주소의 트랜잭션 찾기
        """
        try:
            block = w3.eth.get_block(block_num, full_transactions=True)
            found_txs = []
            
            for tx in block['transactions']:
                if tx['from'] == address or tx['to'] == address:
                    found_txs.append({
                        'hash': tx['hash'].hex(),
                        'from': tx['from'],
                        'to': tx['to'],
                        'value': Web3.from_wei(tx['value'], 'ether'),
                        'block': block_num,
                        'timestamp': block['timestamp']
                    })
            
            if found_txs:
                print(f"✅ 블록 {block_num}: {len(found_txs)}개 트랜잭션 발견")
            
            return found_txs
            
        except Exception as e:
            print(f"❌ 블록 {block_num} 에러: {e}")
            return []
    
    # 병렬 처리
    block_range = range(start_block, end_block + 1)
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(check_block_for_address, block_num) 
                  for block_num in block_range]
        
        for future in as_completed(futures):
            result = future.result()
            transactions.extend(result)
    
    # 시간순 정렬
    transactions.sort(key=lambda x: x['timestamp'])
    
    print(f"\n🎉 총 {len(transactions)}개 트랜잭션 발견!")
    return transactions

# 사용 예시
vitalik_address = "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
vitalik_txs = get_address_transactions(
    vitalik_address, 
    start_block=18000000, 
    end_block=18000100,
    max_workers=20
)

사례 2: NFT 컬렉션 메타데이터 수집

import json

async def fetch_nft_metadata_async(session: aiohttp.ClientSession, 
                                   token_id: int, 
                                   base_uri: str) -> Dict:
    """
    NFT 메타데이터를 비동기로 가져오기
    """
    url = f"{base_uri}/{token_id}"
    
    try:
        async with session.get(url, timeout=10) as response:
            if response.status == 200:
                metadata = await response.json()
                print(f"✅ Token #{token_id} 메타데이터 수집 완료")
                return {
                    'token_id': token_id,
                    'name': metadata.get('name'),
                    'description': metadata.get('description'),
                    'image': metadata.get('image'),
                    'attributes': metadata.get('attributes', [])
                }
    except Exception as e:
        print(f"❌ Token #{token_id} 에러: {e}")
        return None

async def collect_nft_collection(base_uri: str, 
                                 total_supply: int,
                                 batch_size: int = 100) -> List[Dict]:
    """
    NFT 컬렉션 전체 메타데이터 수집
    """
    all_metadata = []
    
    # 배치로 나눠서 처리
    for start in range(0, total_supply, batch_size):
        end = min(start + batch_size, total_supply)
        token_ids = range(start, end)
        
        print(f"\n📦 Token #{start}~#{end-1} 수집 중...")
        
        connector = aiohttp.TCPConnector(limit=50)
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = [
                fetch_nft_metadata_async(session, token_id, base_uri)
                for token_id in token_ids
            ]
            
            batch_results = await asyncio.gather(*tasks)
            all_metadata.extend([r for r in batch_results if r is not None])
        
        # 배치 간 쉬기
        await asyncio.sleep(1)
    
    print(f"\n🎉 총 {len(all_metadata)}개 NFT 메타데이터 수집 완료!")
    return all_metadata

# 사용 예시 (BAYC 같은 유명 컬렉션)
# base_uri = "https://ipfs.io/ipfs/QmeSjSinHpPnmXmspMjwiXyN6zS4E9zccariGR3jxcaWtq"
# nft_data = asyncio.run(collect_nft_collection(base_uri, total_supply=10000))

사례 3: 가스비 분석 대시보드 데이터 수집

def analyze_gas_prices(block_count: int = 100, max_workers: int = 20) -> Dict:
    """
    최근 블록들의 가스비를 분석
    """
    latest_block = w3.eth.block_number
    start_block = latest_block - block_count
    
    def get_block_gas_info(block_num: int) -> Dict:
        """
        블록의 가스 정보 추출
        """
        try:
            block = w3.eth.get_block(block_num, full_transactions=True)
            
            # 트랜잭션별 가스 가격 수집
            gas_prices = [tx['gasPrice'] for tx in block['transactions']]
            
            if gas_prices:
                return {
                    'block': block_num,
                    'timestamp': block['timestamp'],
                    'base_fee': block.get('baseFeePerGas', 0),
                    'gas_used': block['gasUsed'],
                    'gas_limit': block['gasLimit'],
                    'tx_count': len(block['transactions']),
                    'min_gas_price': min(gas_prices),
                    'max_gas_price': max(gas_prices),
                    'avg_gas_price': sum(gas_prices) // len(gas_prices)
                }
        except Exception as e:
            print(f"❌ 블록 {block_num} 에러: {e}")
            return None
    
    # 병렬로 데이터 수집
    block_range = range(start_block, latest_block + 1)
    gas_data = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(get_block_gas_info, block_num) 
                  for block_num in block_range]
        
        for future in as_completed(futures):
            result = future.result()
            if result:
                gas_data.append(result)
    
    # 정렬
    gas_data.sort(key=lambda x: x['block'])
    
    # 통계 계산
    if gas_data:
        avg_gas_prices = [d['avg_gas_price'] for d in gas_data]
        
        analysis = {
            'period': f"블록 {start_block}~{latest_block}",
            'total_blocks': len(gas_data),
            'overall_avg_gas': sum(avg_gas_prices) // len(avg_gas_prices),
            'min_gas': min(avg_gas_prices),
            'max_gas': max(avg_gas_prices),
            'data': gas_data
        }
        
        print(f"\n📊 가스비 분석 결과:")
        print(f"평균 가스비: {Web3.from_wei(analysis['overall_avg_gas'], 'gwei'):.2f} Gwei")
        print(f"최저: {Web3.from_wei(analysis['min_gas'], 'gwei'):.2f} Gwei")
        print(f"최고: {Web3.from_wei(analysis['max_gas'], 'gwei'):.2f} Gwei")
        
        return analysis
    
    return {}

# 실행
gas_analysis = analyze_gas_prices(block_count=100, max_workers=20)

🎓 성능 측정 및 비교

import pandas as pd
import matplotlib.pyplot as plt
from typing import Callable

def benchmark_methods(block_numbers: List[int], 
                     methods: Dict[str, Callable],
                     runs: int = 3) -> pd.DataFrame:
    """
    여러 방법의 성능을 벤치마크
    
    Args:
        block_numbers: 테스트할 블록 번호들
        methods: {방법명: 함수} 딕셔너리
        runs: 각 방법당 실행 횟수
    """
    results = []
    
    for method_name, method_func in methods.items():
        print(f"\n{'='*50}")
        print(f"🔍 {method_name} 테스트 중...")
        print(f"{'='*50}")
        
        times = []
        
        for run in range(runs):
            print(f"\n실행 {run + 1}/{runs}...")
            start_time = time.time()
            
            try:
                method_func(block_numbers)
                elapsed = time.time() - start_time
                times.append(elapsed)
                print(f"⏱️ 소요 시간: {elapsed:.2f}초")
                
            except Exception as e:
                print(f"❌ 에러 발생: {e}")
                times.append(None)
        
        # 유효한 시간만 계산
        valid_times = [t for t in times if t is not None]
        
        if valid_times:
            results.append({
                '방법': method_name,
                '평균 시간': sum(valid_times) / len(valid_times),
                '최소 시간': min(valid_times),
                '최대 시간': max(valid_times),
                '성공률': len(valid_times) / runs * 100
            })
    
    # DataFrame으로 변환
    df = pd.DataFrame(results)
    df = df.sort_values('평균 시간')
    
    # 기준 대비 성능 향상 계산
    baseline = df.iloc[-1]['평균 시간']  # 가장 느린 것을 기준으로
    df['성능 향상'] = baseline / df['평균 시간']
    
    return df

# 테스트할 방법들 정의
test_methods = {
    '순차 처리': lambda blocks: get_block_sequential(blocks),
    '병렬 처리 (5 workers)': lambda blocks: get_block_parallel(blocks, max_workers=5),
    '병렬 처리 (10 workers)': lambda blocks: get_block_parallel(blocks, max_workers=10),
    '병렬 처리 (20 workers)': lambda blocks: get_block_parallel(blocks, max_workers=20),
}

# 벤치마크 실행
test_blocks = list(range(18000000, 18000050))  # 50개 블록
benchmark_df = benchmark_methods(test_blocks, test_methods, runs=3)

print("\n" + "="*70)
print("📊 벤치마크 결과")
print("="*70)
print(benchmark_df.to_string(index=False))
print("="*70)

⚠️ 주의사항과 함정 피하기

1. GIL (Global Interpreter Lock) 이해하기

2. 메모리 관리

import psutil
import os

def monitor_memory_usage(func):
    """
    메모리 사용량을 모니터링하는 데코레이터
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        process = psutil.Process(os.getpid())
        
        # 시작 전 메모리
        mem_before = process.memory_info().rss / 1024 / 1024  # MB
        print(f"📊 시작 메모리: {mem_before:.2f} MB")
        
        # 함수 실행
        result = func(*args, **kwargs)
        
        # 종료 후 메모리
        mem_after = process.memory_info().rss / 1024 / 1024  # MB
        mem_diff = mem_after - mem_before
        
        print(f"📊 종료 메모리: {mem_after:.2f} MB")
        print(f"📊 메모리 증가: {mem_diff:.2f} MB")
        
        if mem_diff > 500:  # 500MB 이상 증가
            print("⚠️ 경고: 메모리 사용량이 많습니다!")
        
        return result
    
    return wrapper

@monitor_memory_usage
def process_large_dataset(block_count: int, max_workers: int):
    """
    대량의 데이터를 처리하면서 메모리 모니터링
    """
    latest_block = w3.eth.block_number
    blocks = list(range(latest_block - block_count, latest_block))
    
    return get_block_parallel(blocks, max_workers=max_workers)

# 테스트
process_large_dataset(1000, max_workers=50)

3. 연결 풀 관리

from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import requests

def create_session_with_retries(
    pool_connections: int = 10,
    pool_maxsize: int = 20,
    max_retries: int = 3
) -> requests.Session:
    """
    재시도 로직과 연결 풀이 설정된 세션 생성
    """
    session = requests.Session()
    
    # 재시도 전략
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=1,  # 1, 2, 4초로 증가
        status_forcelist=[429, 500, 502, 503, 504],
        method_whitelist=["HEAD", "GET", "POST", "PUT", "DELETE", "OPTIONS", "TRACE"]
    )
    
    # HTTP 어댑터 설정
    adapter = HTTPAdapter(
        pool_connections=pool_connections,
        pool_maxsize=pool_maxsize,
        max_retries=retry_strategy
    )
    
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session

# 전역 세션 생성 (재사용)
global_session = create_session_with_retries(
    pool_connections=20,
    pool_maxsize=50,
    max_retries=3
)

def fetch_with_session(url: str, payload: Dict) -> Dict:
    """
    세션을 재사용하여 요청
    """
    response = global_session.post(url, json=payload, timeout=30)
    return response.json()

4. 타임아웃 설정

from concurrent.futures import TimeoutError as FuturesTimeoutError

def fetch_with_timeout(block_num: int, timeout: int = 30) -> Dict:
    """
    타임아웃이 설정된 블록 가져오기
    """
    try:
        # Web3 호출에도 타임아웃 적용 가능
        block = w3.eth.get_block(block_num)
        return {
            'number': block['number'],
            'timestamp': block['timestamp'],
            'transactions': len(block['transactions'])
        }
    except Exception as e:
        print(f"❌ 블록 {block_num} 타임아웃 또는 에러: {e}")
        return None

def get_blocks_with_timeout(block_numbers: List[int], 
                           max_workers: int = 10,
                           timeout: int = 60) -> List[Dict]:
    """
    전체 작업에 타임아웃 적용
    """
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_block = {
            executor.submit(fetch_with_timeout, block_num): block_num 
            for block_num in block_numbers
        }
        
        try:
            for future in as_completed(future_to_block, timeout=timeout):
                block_num = future_to_block[future]
                try:
                    result = future.result(timeout=10)  # 개별 타임아웃
                    results.append(result)
                except FuturesTimeoutError:
                    print(f"⏱️ 블록 {block_num} 개별 타임아웃")
                    results.append(None)
                except Exception as e:
                    print(f"❌ 블록 {block_num} 에러: {e}")
                    results.append(None)
                    
        except FuturesTimeoutError:
            print(f"⏱️ 전체 작업 타임아웃 ({timeout}초)")
    
    return results

🛠️ 프로덕션 레벨 구현

완전한 블록체인 API 클라이언트

import logging
from dataclasses import dataclass
from typing import Optional, List, Dict, Any, Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
from functools import wraps
import json

# 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class BlockchainConfig:
    """
    블록체인 API 설정
    """
    rpc_url: str
    max_workers: int = 10
    timeout: int = 30
    max_retries: int = 3
    retry_delay: float = 1.0
    rate_limit_per_second: int = 10
    batch_size: int = 100

class BlockchainAPIClient:
    """
    프로덕션 레벨 블록체인 API 클라이언트
    """
    
    def __init__(self, config: BlockchainConfig):
        self.config = config
        self.w3 = Web3(Web3.HTTPProvider(config.rpc_url))
        self.rate_limiter = RateLimiter(
            max_calls=config.rate_limit_per_second,
            time_window=1.0
        )
        
        # 연결 확인
        if not self.w3.is_connected():
            raise ConnectionError("블록체인 노드에 연결할 수 없습니다")
        
        logger.info(f"블록체인 클라이언트 초기화 완료: {config.rpc_url}")
    
    def _retry_on_failure(self, func: Callable) -> Callable:
        """
        재시도 로직 데코레이터
        """
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(self.config.max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == self.config.max_retries - 1:
                        logger.error(f"최종 실패: {e}")
                        raise
                    
                    wait_time = self.config.retry_delay * (2 ** attempt)
                    logger.warning(f"재시도 {attempt + 1}/{self.config.max_retries}, "
                                 f"{wait_time}초 후 재시도...")
                    time.sleep(wait_time)
        
        return wrapper
    
    @_retry_on_failure
    def get_block(self, block_number: int, full_transactions: bool = False) -> Dict:
        """
        단일 블록 정보 가져오기
        """
        self.rate_limiter(lambda: None)()  # Rate limiting 적용
        
        block = self.w3.eth.get_block(block_number, full_transactions=full_transactions)
        
        return {
            'number': block['number'],
            'hash': block['hash'].hex(),
            'timestamp': block['timestamp'],
            'transactions': len(block['transactions']) if not full_transactions 
                          else [tx['hash'].hex() for tx in block['transactions']],
            'gas_used': block['gasUsed'],
            'gas_limit': block['gasLimit'],
            'base_fee': block.get('baseFeePerGas', 0)
        }
    
    def get_blocks_parallel(self, 
                          block_numbers: List[int],
                          full_transactions: bool = False,
                          progress_callback: Optional[Callable] = None) -> List[Dict]:
        """
        여러 블록을 병렬로 가져오기
        
        Args:
            block_numbers: 블록 번호 리스트
            full_transactions: 전체 트랜잭션 정보 포함 여부
            progress_callback: 진행상황 콜백 함수
        """
        results = []
        total = len(block_numbers)
        completed = 0
        
        logger.info(f"{total}개 블록 병렬 처리 시작 (workers: {self.config.max_workers})")
        start_time = time.time()
        
        with ThreadPoolExecutor(max_workers=self.config.max_workers) as executor:
            future_to_block = {
                executor.submit(self.get_block, block_num, full_transactions): block_num
                for block_num in block_numbers
            }
            
            for future in as_completed(future_to_block):
                block_num = future_to_block[future]
                
                try:
                    result = future.result(timeout=self.config.timeout)
                    results.append(result)
                    completed += 1
                    
                    # 진행상황 콜백
                    if progress_callback:
                        progress_callback(completed, total)
                    
                    if completed % 10 == 0:
                        logger.info(f"진행: {completed}/{total} ({completed/total*100:.1f}%)")
                        
                except Exception as e:
                    logger.error(f"블록 {block_num} 처리 실패: {e}")
                    results.append(None)
        
        elapsed = time.time() - start_time
        success_count = sum(1 for r in results if r is not None)
        
        logger.info(f"완료: {success_count}/{total} 성공, {elapsed:.2f}초 소요")
        
        return results
    
    def get_blocks_in_batches(self,
                            start_block: int,
                            end_block: int,
                            full_transactions: bool = False) -> List[Dict]:
        """
        배치 단위로 블록 가져오기
        """
        all_results = []
        total_blocks = end_block - start_block + 1
        
        logger.info(f"배치 처리 시작: 블록 {start_block}~{end_block} ({total_blocks}개)")
        
        for batch_start in range(start_block, end_block + 1, self.config.batch_size):
            batch_end = min(batch_start + self.config.batch_size - 1, end_block)
            batch_numbers = list(range(batch_start, batch_end + 1))
            
            logger.info(f"배치 처리: {batch_start}~{batch_end}")
            
            batch_results = self.get_blocks_parallel(
                batch_numbers,
                full_transactions=full_transactions
            )
            
            all_results.extend(batch_results)
            
            # 배치 간 쉬기
            if batch_end < end_block:
                time.sleep(0.5)
        
        return all_results
    
    def get_transaction(self, tx_hash: str) -> Dict:
        """
        트랜잭션 정보 가져오기
        """
        self.rate_limiter(lambda: None)()
        
        tx = self.w3.eth.get_transaction(tx_hash)
        receipt = self.w3.eth.get_transaction_receipt(tx_hash)
        
        return {
            'hash': tx['hash'].hex(),
            'from': tx['from'],
            'to': tx['to'],
            'value': str(tx['value']),
            'gas': tx['gas'],
            'gas_price': str(tx['gasPrice']),
            'nonce': tx['nonce'],
            'block_number': tx['blockNumber'],
            'status': receipt['status'],
            'gas_used': receipt['gasUsed']
        }
    
    def get_address_balance(self, address: str, block: str = 'latest') -> Dict:
        """
        주소의 잔액 조회
        """
        self.rate_limiter(lambda: None)()
        
        checksum_address = Web3.to_checksum_address(address)
        balance_wei = self.w3.eth.get_balance(checksum_address, block)
        
        return {
            'address': checksum_address,
            'balance_wei': str(balance_wei),
            'balance_eth': str(Web3.from_wei(balance_wei, 'ether')),
            'block': block
        }
    
    def export_to_json(self, data: List[Dict], filename: str):
        """
        데이터를 JSON 파일로 저장
        """
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
        
        logger.info(f"데이터 저장 완료: {filename}")

# 사용 예시
def main():
    # 설정
    config = BlockchainConfig(
        rpc_url="https://mainnet.infura.io/v3/YOUR_PROJECT_ID",
        max_workers=20,
        timeout=30,
        max_retries=3,
        rate_limit_per_second=10,
        batch_size=100
    )
    
    # 클라이언트 생성
    client = BlockchainAPIClient(config)
    
    # 진행상황 콜백
    def progress_callback(completed: int, total: int):
        percentage = completed / total * 100
        print(f"진행: {completed}/{total} ({percentage:.1f}%)")
    
    # 최근 100개 블록 가져오기
    latest_block = client.w3.eth.block_number
    blocks = client.get_blocks_in_batches(
        start_block=latest_block - 100,
        end_block=latest_block,
        full_transactions=False
    )
    
    # JSON으로 저장
    client.export_to_json(blocks, 'blocks_data.json')
    
    print(f"✅ {len(blocks)}개 블록 데이터 수집 완료!")

if __name__ == "__main__":
    main()

📈 실제 성능 개선 사례

사례 1: NFT 마켓플레이스 데이터 수집

사례 2: DeFi 프로토콜 분석 대시보드

사례 3: 지갑 포트폴리오 트래커

🔮 고급 기법과 최적화

1. 캐싱 전략

from functools import lru_cache
import redis
import pickle
from datetime import timedelta

class BlockchainCache:
    """
    Redis를 사용한 블록체인 데이터 캐싱
    """
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis_client = redis.from_url(redis_url)
        self.default_ttl = 300  # 5분
    
    def get_block_cached(self, block_number: int, ttl: int = None) -> Optional[Dict]:
        """
        캐시에서 블록 데이터 가져오기
        """
        cache_key = f"block:{block_number}"
        
        # 캐시 확인
        cached_data = self.redis_client.get(cache_key)
        
        if cached_data:
            logger.info(f"캐시 히트: 블록 {block_number}")
            return pickle.loads(cached_data)
        
        # 캐시 미스 - 실제 데이터 가져오기
        logger.info(f"캐시 미스: 블록 {block_number}")
        block_data = self.fetch_block_from_chain(block_number)
        
        # 캐시에 저장
        if block_data:
            ttl = ttl or self.default_ttl
            self.redis_client.setex(
                cache_key,
                timedelta(seconds=ttl),
                pickle.dumps(block_data)
            )
        
        return block_data
    
    def fetch_block_from_chain(self, block_number: int) -> Dict:
        """
        실제 블록체인에서 데이터 가져오기
        """
        # 여기에 실제 API 호출 로직
        pass
    
    def invalidate_cache(self, block_number: int):
        """
        특정 블록의 캐시 무효화
        """
        cache_key = f"block:{block_number}"
        self.redis_client.delete(cache_key)
        logger.info(f"캐시 무효화: 블록 {block_number}")

# 메모리 캐싱 (간단한 경우)
@lru_cache(maxsize=1000)
def get_block_memory_cached(block_number: int) -> Dict:
    """
    메모리 캐싱을 사용한 블록 가져오기
    """
    return fetch_single_block(block_number)

2. 연결 풀링 최적화

from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
import requests

class OptimizedWeb3Provider:
    """
    최적화된 Web3 프로바이더
    """
    
    def __init__(self, rpc_url: str, pool_size: int = 50):
        # 세션 생성
        self.session = requests.Session()
        
        # 재시도 전략
        retry_strategy = Retry(
            total=5,
            backoff_factor=0.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["HEAD", "GET", "POST"]
        )
        
        # 어댑터 설정
        adapter = HTTPAdapter(
            pool_connections=pool_size,
            pool_maxsize=pool_size * 2,
            max_retries=retry_strategy,
            pool_block=False
        )
        
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        
        # Web3 인스턴스
        self.w3 = Web3(Web3.HTTPProvider(
            rpc_url,
            session=self.session
        ))
        
        logger.info(f"최적화된 프로바이더 초기화: pool_size={pool_size}")
    
    def batch_request(self, requests_data: List[Dict]) -> List[Dict]:
        """
        배치 요청으로 여러 호출을 한 번에 처리
        """
        # JSON-RPC 배치 요청
        response = self.session.post(
            self.w3.provider.endpoint_uri,
            json=requests_data,
            timeout=30
        )
        
        return response.json()

# 사용 예시
provider = OptimizedWeb3Provider(
    rpc_url="https://mainnet.infura.io/v3/YOUR_PROJECT_ID",
    pool_size=50
)

# 배치 요청으로 여러 블록 한 번에 가져오기
batch_requests = [
    {
        "jsonrpc": "2.0",
        "method": "eth_getBlockByNumber",
        "params": [hex(block_num), False],
        "id": i
    }
    for i, block_num in enumerate(range(18000000, 18000010))
]

results = provider.batch_request(batch_requests)

3. 적응형 동시성 제어

import statistics

class AdaptiveConcurrencyController:
    """
    적응형 동시성 제어기
    """
    
    def __init__(self, 
                 initial_workers: int = 10,
                 min_workers: int = 5,
                 max_workers: int = 50,
                 adjustment_interval: int = 10):
        self.current_workers = initial_workers
        self.min_workers = min_workers
        self.max_workers = max_workers
        self.adjustment_interval = adjustment_interval
        
        self.response_times = []
        self.error_count = 0
        self.request_count = 0
    
    def record_request(self, response_time: float, success: bool):
        """
        요청 결과 기록
        """
        self.request_count += 1
        
        if success:
            self.response_times.append(response_time)
        else:
            self.error_count += 1
        
        # 주기적으로 조정
        if self.request_count % self.adjustment_interval == 0:
            self.adjust_concurrency()
    
    def adjust_concurrency(self):
        """
        성능 지표를 기반으로 동시성 조정
        """
        if len(self.response_times) < 5:
            return
        
        # 최근 응답 시간 분석
        recent_times = self.response_times[-20:]
        avg_time = statistics.mean(recent_times)
        std_time = statistics.stdev(recent_times) if len(recent_times) > 1 else 0
        
        # 에러율 계산
        error_rate = self.error_count / self.request_count
        
        old_workers = self.current_workers
        
        # 조정 로직
        if error_rate > 0.1:  # 에러율 10% 이상
            # 워커 수 감소
            self.current_workers = max(
                self.min_workers,
                int(self.current_workers * 0.8)
            )
            logger.warning(f"에러율 높음 ({error_rate:.1%}), 워커 감소: {old_workers} → {self.current_workers}")
            
        elif avg_time > 5.0:  # 평균 응답 시간 5초 이상
            # 워커 수 감소
            self.current_workers = max(
                self.min_workers,
                int(self.current_workers * 0.9)
            )
            logger.warning(f"응답 시간 느림 ({avg_time:.2f}s), 워커 감소: {old_workers} → {self.current_workers}")
            
        elif avg_time < 1.0 and std_time < 0.5:  # 빠르고 안정적
            # 워커 수 증가
            self.current_workers = min(
                self.max_workers,
                int(self.current_workers * 1.2)
            )
            logger.info(f"성능 좋음, 워커 증가: {old_workers} → {self.current_workers}")
        
        # 통계 초기화
        self.response_times = self.response_times[-50:]
        self.error_count = 0
        self.request_count = 0

# 사용 예시
controller = AdaptiveConcurrencyController(
    initial_workers=10,
    min_workers=5,
    max_workers=50
)

def fetch_with_adaptive_concurrency(block_numbers: List[int]) -> List[Dict]:
    """
    적응형 동시성을 사용한 블록 가져오기
    """
    results = []
    
    for batch_start in range(0, len(block_numbers), 100):
        batch = block_numbers[batch_start:batch_start + 100]
        
        # 현재 워커 수 사용
        workers = controller.current_workers
        logger.info(f"현재 워커 수: {workers}")
        
        with ThreadPoolExecutor(max_workers=workers) as executor:
            futures = {
                executor.submit(fetch_single_block, block_num): block_num
                for block_num in batch
            }
            
            for future in as_completed(futures):
                start_time = time.time()
                
                try:
                    result = future.result()
                    elapsed = time.time() - start_time
                    
                    # 성공 기록
                    controller.record_request(elapsed, success=True)
                    results.append(result)
                    
                except Exception as e:
                    elapsed = time.time() - start_time
                    
                    # 실패 기록
                    controller.record_request(elapsed, success=False)
                    results.append(None)
    
    return results

💡 실무 팁과 체크리스트

성능 최적화 우선순위

흔한 실수와 해결책

코드 리뷰 체크포인트

# 좋은 예시 ✅
def fetch_blocks_good(block_numbers: List[int]) -> List[Dict]:
    """
    블록 정보를 안전하게 병렬로 가져오기
    
    Args:
        block_numbers: 블록 번호 리스트
        
    Returns:
        블록 정보 리스트 (실패한 경우 None 포함)
        
    Raises:
        ConnectionError: 블록체인 노드 연결 실패
        TimeoutError: 전체 작업 타임아웃
    """
    if not block_numbers:
        logger.warning("빈 블록 리스트")
        return []
    
    results = []
    max_workers = min(20, len(block_numbers))  # 적절한 워커 수
    
    logger.info(f"{len(block_numbers)}개 블록 처리 시작")
    
    try:
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_block = {
                executor.submit(fetch_single_block_with_retry, block_num): block_num
                for block_num in block_numbers
            }
            
            for future in as_completed(future_to_block, timeout=300):
                block_num = future_to_block[future]
                
                try:
                    result = future.result(timeout=30)
                    results.append(result)
                    
                except Exception as e:
                    logger.error(f"블록 {block_num} 처리 실패: {e}")
                    results.append(None)
                    
    except TimeoutError:
        logger.error("전체 작업 타임아웃")
        raise
    
    success_count = sum(1 for r in results if r is not None)
    logger.info(f"완료: {success_count}/{len(block_numbers)} 성공")
    
    return results

# 나쁜 예시 ❌
def fetch_blocks_bad(block_numbers):
    # 문서화 없음
    # 에러 처리 없음
    # 로깅 없음
    # 타임아웃 없음
    
    with ThreadPoolExecutor(max_workers=1000) as executor:  # 워커 수 과다
        futures = [executor.submit(get_block, n) for n in block_numbers]
        return [f.result() for f in futures]  # 에러 시 전체 실패

🎬 마무리하며