챕터 20: 메모리 관리

파이썬의 메모리 관리는 개발자가 직접 메모리를 할당하고 해제할 필요가 없도록 자동화되어 있습니다. 하지만 효율적인 프로그램을 작성하기 위해서는 파이썬의 메모리 관리 메커니즘을 이해하고 최적화 기법을 적용할 수 있어야 합니다.

학습 목표

파이썬의 메모리 관리 메커니즘을 이해할 수 있다
메모리 누수를 식별하고 해결할 수 있다
효율적인 메모리 사용 패턴을 적용할 수 있다
가비지 컬렉션을 제어하고 최적화할 수 있다

핵심 개념(이론)

1) 메모리 관리의 역할과 경계

이 챕터의 핵심은 “무엇을 할 수 있나”가 아니라, 어떤 문제를 해결하고 어디까지 책임지는지를 분명히 하는 것입니다. 경계가 흐리면 코드는 커질수록 결합이 늘어나고 수정 비용이 커집니다.

2) 왜 이 개념이 필요한가(실무 동기)

실무에서는 예외 상황, 성능, 협업, 테스트가 항상 문제를 만듭니다. 따라서 이 주제는 기능이 아니라 **품질(신뢰성/유지보수성/보안)**을 위한 기반으로 이해해야 합니다.

3) 트레이드오프: 간단함 vs 확장성

대부분의 선택은 “더 단순하게”와 “더 확장 가능하게” 사이에서 균형을 잡는 일입니다. 초기에는 단순함을, 장기 운영/팀 협업이 커질수록 확장성을 더 우선합니다.

4) 실패 모드(Failure Modes)를 먼저 생각하라

무엇이 실패하는지(입력, I/O, 동시성, 외부 시스템)를 먼저 떠올리면 설계가 안정적으로 변합니다. 이 챕터의 예제는 실패 모드를 축소해서 보여주므로, 실제 적용 시에는 더 많은 방어가 필요합니다.

5) 학습 포인트: 외우지 말고 “판단 기준”을 남겨라

핵심은 API를 외우는 것이 아니라, “언제 무엇을 선택할지” 판단 기준을 정리하는 것입니다. 이 기준이 쌓이면 새로운 라이브러리/도구가 나와도 빠르게 적응할 수 있습니다.

선택 기준(Decision Guide)

기본은 가독성/명확성 우선(최적화는 측정 이후).
외부 의존이 늘수록 경계/추상화와 테스트를 먼저 강화.
복잡도가 증가하면 “규칙을 코드로”가 아니라 “구조로” 담는 방향을 고려.

흔한 오해/주의점

도구/문법이 곧 실력이라는 오해가 있습니다. 실력은 문제를 단순화하고 구조화하는 능력입니다.
극단적 최적화/과설계는 학습과 유지보수를 방해할 수 있습니다.

요약

메모리 관리는 기능이 아니라 구조/품질을 위한 기반이다.
트레이드오프와 실패 모드를 먼저 생각하고, 판단 기준을 남기자.

파이썬 메모리 모델

객체와 참조

파이썬에서 모든 것은 객체이며, 변수는 객체에 대한 참조입니다.

1
2
3
4
5
6
7
8
# 객체 생성과 참조
a = [1, 2, 3]  # 리스트 객체 생성
b = a          # 같은 객체를 참조
c = [1, 2, 3]  # 새로운 리스트 객체 생성

print(id(a), id(b), id(c))  # 메모리 주소 확인
print(a is b)  # True - 같은 객체
print(a is c)  # False - 다른 객체

참조 카운팅

파이썬은 각 객체의 참조 횟수를 추적합니다.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import sys

class MyClass:
    def __init__(self, name):
        self.name = name
    
    def __del__(self):
        print(f"{self.name} 객체가 삭제됩니다")

# 참조 카운팅 예제
obj = MyClass("테스트")
print(f"참조 횟수: {sys.getrefcount(obj)}")  # 2 (obj 변수 + getrefcount 인수)

obj2 = obj  # 참조 추가
print(f"참조 횟수: {sys.getrefcount(obj)}")  # 3

del obj2    # 참조 제거
print(f"참조 횟수: {sys.getrefcount(obj)}")  # 2

del obj     # 마지막 참조 제거, 객체 삭제됨

메모리 레이아웃

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
import sys

# 다양한 객체의 크기 확인
objects = [
    42,
    "hello",
    [1, 2, 3],
    {"key": "value"},
    set([1, 2, 3])
]

for obj in objects:
    print(f"{type(obj).__name__}: {sys.getsizeof(obj)} bytes")

가비지 컬렉션

순환 참조 문제

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
import gc

class Node:
    def __init__(self, value):
        self.value = value
        self.child = None
        self.parent = None

# 순환 참조 생성
parent = Node("parent")
child = Node("child")
parent.child = child
child.parent = parent

# 참조 제거 후에도 메모리에 남아있음
del parent, child

print(f"수집 전 객체 수: {len(gc.get_objects())}")
collected = gc.collect()
print(f"수집된 객체 수: {collected}")
print(f"수집 후 객체 수: {len(gc.get_objects())}")

가비지 컬렉션 제어

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import gc

# 가비지 컬렉션 정보 확인
print(f"GC 임계값: {gc.get_threshold()}")
print(f"GC 통계: {gc.get_stats()}")

# 가비지 컬렉션 비활성화/활성화
gc.disable()
print(f"GC 활성화 상태: {gc.isenabled()}")

gc.enable()
print(f"GC 활성화 상태: {gc.isenabled()}")

# 수동 가비지 컬렉션
collected = gc.collect()
print(f"수집된 객체 수: {collected}")

약한 참조 (Weak Reference)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import weakref

class ExpensiveObject:
    def __init__(self, name):
        self.name = name
    
    def __del__(self):
        print(f"{self.name} 삭제됨")

# 일반 참조
obj = ExpensiveObject("강한참조")
ref = obj

# 약한 참조
weak_ref = weakref.ref(obj)

print(f"객체 존재: {weak_ref() is not None}")

# 강한 참조 제거
del obj
print(f"객체 존재: {weak_ref() is not None}")

del ref  # 마지막 강한 참조 제거
print(f"객체 존재: {weak_ref() is not None}")

메모리 프로파일링

tracemalloc 모듈

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import tracemalloc

def memory_intensive_function():
    # 메모리를 많이 사용하는 함수
    data = []
    for i in range(100000):
        data.append(str(i) * 10)
    return data

# 메모리 추적 시작
tracemalloc.start()

# 첫 번째 스냅샷
snapshot1 = tracemalloc.take_snapshot()

# 메모리 집약적 작업 수행
result = memory_intensive_function()

# 두 번째 스냅샷
snapshot2 = tracemalloc.take_snapshot()

# 메모리 사용량 비교
top_stats = snapshot2.compare_to(snapshot1, 'lineno')

print("메모리 사용량 상위 10개:")
for stat in top_stats[:10]:
    print(stat)

memory_profiler 사용

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
# pip install memory-profiler 필요

from memory_profiler import profile

@profile
def memory_test():
    # 리스트 생성
    a = [1] * (10**6)
    
    # 딕셔너리 생성  
    b = {i: i**2 for i in range(10**5)}
    
    # 메모리 해제
    del a
    del b

if __name__ == "__main__":
    memory_test()

메모리 최적화 기법

slots 사용

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import sys

# 일반 클래스
class RegularClass:
    def __init__(self, x, y):
        self.x = x
        self.y = y

# __slots__ 사용 클래스
class SlottedClass:
    __slots__ = ['x', 'y']
    
    def __init__(self, x, y):
        self.x = x
        self.y = y

# 메모리 사용량 비교
regular = RegularClass(1, 2)
slotted = SlottedClass(1, 2)

print(f"일반 클래스: {sys.getsizeof(regular)} bytes")
print(f"슬롯 클래스: {sys.getsizeof(slotted)} bytes")

# 대량 생성 시 메모리 차이
import tracemalloc

tracemalloc.start()

# 일반 클래스 10000개 생성
regular_objects = [RegularClass(i, i) for i in range(10000)]
snapshot1 = tracemalloc.take_snapshot()

# 슬롯 클래스 10000개 생성  
slotted_objects = [SlottedClass(i, i) for i in range(10000)]
snapshot2 = tracemalloc.take_snapshot()

current, peak = tracemalloc.get_traced_memory()
print(f"현재 메모리: {current / 1024 / 1024:.2f} MB")
print(f"최대 메모리: {peak / 1024 / 1024:.2f} MB")

제너레이터를 통한 메모리 효율성

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import sys

# 리스트 vs 제너레이터 메모리 사용량 비교
def create_list(n):
    return [i**2 for i in range(n)]

def create_generator(n):
    return (i**2 for i in range(n))

n = 100000

# 리스트 방식
list_result = create_list(n)
print(f"리스트 크기: {sys.getsizeof(list_result)} bytes")

# 제너레이터 방식
gen_result = create_generator(n)
print(f"제너레이터 크기: {sys.getsizeof(gen_result)} bytes")

메모리 효율적 자료구조

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import array
import sys

# 일반 리스트 vs array 모듈
normal_list = [i for i in range(10000)]
int_array = array.array('i', [i for i in range(10000)])

print(f"일반 리스트: {sys.getsizeof(normal_list)} bytes")
print(f"int 배열: {sys.getsizeof(int_array)} bytes")

# collections.deque 활용
from collections import deque

# 큐 연산이 빈번한 경우
queue_list = []
queue_deque = deque()

# 성능과 메모리 효율성에서 deque가 유리

대용량 데이터 처리

메모리 매핑

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import mmap
import os

# 대용량 파일을 메모리에 매핑
def process_large_file(filename):
    with open(filename, 'r+b') as f:
        # 파일을 메모리에 매핑
        with mmap.mmap(f.fileno(), 0) as mm:
            # 첫 100바이트 읽기
            data = mm[:100]
            print(f"파일 크기: {len(mm)} bytes")
            print(f"첫 100바이트: {data}")
            
            # 특정 위치로 이동
            mm.seek(50)
            chunk = mm.read(20)
            print(f"50-70 바이트: {chunk}")

# 테스트 파일 생성
if __name__ == "__main__":
    test_file = "large_test_file.txt"
    with open(test_file, 'w') as f:
        f.write("A" * 1000000)  # 1MB 파일
    
    process_large_file(test_file)
    os.remove(test_file)

청크 단위 처리

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def process_large_dataset_chunked(data_generator, chunk_size=1000):
    """대용량 데이터를 청크 단위로 처리"""
    chunk = []
    for item in data_generator:
        chunk.append(item)
        
        if len(chunk) >= chunk_size:
            # 청크 처리
            process_chunk(chunk)
            chunk = []  # 메모리 해제
    
    # 마지막 청크 처리
    if chunk:
        process_chunk(chunk)

def process_chunk(chunk):
    """청크 데이터 처리"""
    result = sum(chunk)
    print(f"청크 합계: {result}")

# 사용 예제
def large_data_generator():
    for i in range(100000):
        yield i

# 청크 단위로 처리
process_large_dataset_chunked(large_data_generator())

메모리 누수 탐지 및 해결

일반적인 메모리 누수 패턴

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import gc
import sys
import tracemalloc
from collections import defaultdict
import weakref

# 1. 전역 변수에 대한 참조
global_cache = {}

def leaky_function(key, value):
    # 전역 캐시에 계속 추가만 하고 제거하지 않음
    global_cache[key] = value

# 2. 콜백 함수의 순환 참조
class EventEmitter:
    def __init__(self):
        self.callbacks = []
    
    def add_callback(self, callback):
        self.callbacks.append(callback)
    
    def remove_callback(self, callback):
        if callback in self.callbacks:
            self.callbacks.remove(callback)

class Widget:
    def __init__(self):
        self.emitter = EventEmitter()
        # 순환 참조 발생
        self.emitter.add_callback(self.handle_event)
    
    def handle_event(self):
        print("이벤트 처리")

# 3. 메모리 누수 해결 방법
class ImprovedWidget:
    def __init__(self):
        self.emitter = EventEmitter()
        # 약한 참조 사용
        self._callback_ref = weakref.WeakMethod(self.handle_event)
        self.emitter.add_callback(self._callback_ref)
    
    def handle_event(self):
        print("이벤트 처리")
    
    def cleanup(self):
        self.emitter.remove_callback(self._callback_ref)

메모리 모니터링 도구

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import psutil
import os
import time

class MemoryMonitor:
    def __init__(self):
        self.process = psutil.Process(os.getpid())
        self.initial_memory = self.get_memory_usage()
    
    def get_memory_usage(self):
        """현재 메모리 사용량 반환 (MB)"""
        memory_info = self.process.memory_info()
        return memory_info.rss / 1024 / 1024
    
    def log_memory_usage(self, label=""):
        """메모리 사용량 로깅"""
        current_memory = self.get_memory_usage()
        diff = current_memory - self.initial_memory
        print(f"{label} - 메모리 사용량: {current_memory:.2f} MB (변화: {diff:+.2f} MB)")

# 사용 예제
monitor = MemoryMonitor()
monitor.log_memory_usage("시작")

# 메모리 집약적 작업
data = [i**2 for i in range(100000)]
monitor.log_memory_usage("리스트 생성 후")

del data
monitor.log_memory_usage("리스트 삭제 후")

gc.collect()
monitor.log_memory_usage("GC 수행 후")

실습 프로젝트

프로젝트 1: 메모리 누수 탐지기

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
import gc
import sys
import tracemalloc
from collections import defaultdict
import weakref

class MemoryLeakDetector:
    def __init__(self):
        self.snapshots = []
        self.object_counts = defaultdict(int)
        
    def start_monitoring(self):
        """메모리 모니터링 시작"""
        tracemalloc.start()
        gc.collect()  # 시작 전 정리
        
    def take_snapshot(self, label=""):
        """스냅샷 생성"""
        gc.collect()
        snapshot = tracemalloc.take_snapshot()
        self.snapshots.append((label, snapshot))
        
        # 객체 유형별 카운트
        for obj in gc.get_objects():
            obj_type = type(obj).__name__
            self.object_counts[obj_type] += 1
            
        print(f"스냅샷 '{label}' 생성 완료")
        
    def analyze_memory_growth(self):
        """메모리 증가 분석"""
        if len(self.snapshots) < 2:
            print("분석을 위해 최소 2개의 스냅샷이 필요합니다.")
            return
            
        label1, snapshot1 = self.snapshots[-2]
        label2, snapshot2 = self.snapshots[-1]
        
        top_stats = snapshot2.compare_to(snapshot1, 'lineno')
        
        print(f"\n{label1} → {label2} 메모리 변화:")
        print("=" * 50)
        
        for i, stat in enumerate(top_stats[:10]):
            print(f"{i+1}. {stat}")
            
    def find_potential_leaks(self):
        """잠재적 메모리 누수 탐지"""
        print("\n잠재적 메모리 누수 객체:")
        print("=" * 30)
        
        # 순환 참조 객체 찾기
        for obj in gc.garbage:
            print(f"가비지 객체: {type(obj)} - {obj}")
            
        # 참조 추적
        referrers = gc.get_referrers
        for obj_type, count in self.object_counts.items():
            if count > 1000:  # 임계값
                print(f"많은 수의 {obj_type} 객체: {count}개")

# 사용 예제
def test_memory_leak():
    detector = MemoryLeakDetector()
    detector.start_monitoring()
    
    detector.take_snapshot("시작")
    
    # 메모리 누수 시뮬레이션
    leak_list = []
    for i in range(10000):
        leak_list.append([j for j in range(100)])
    
    detector.take_snapshot("데이터 생성 후")
    
    # 일부만 해제
    del leak_list[::2]
    
    detector.take_snapshot("일부 해제 후")
    
    detector.analyze_memory_growth()
    detector.find_potential_leaks()

if __name__ == "__main__":
    test_memory_leak()

프로젝트 2: 메모리 효율적 데이터 처리기

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import mmap
import json
import csv
from typing import Iterator, Any
import os

class EfficientDataProcessor:
    def __init__(self, chunk_size: int = 8192):
        self.chunk_size = chunk_size
        
    def process_large_text_file(self, filepath: str) -> Iterator[str]:
        """대용량 텍스트 파일을 청크 단위로 처리"""
        try:
            with open(filepath, 'r', encoding='utf-8') as file:
                while True:
                    chunk = file.read(self.chunk_size)
                    if not chunk:
                        break
                    yield chunk
        except FileNotFoundError:
            print(f"파일을 찾을 수 없습니다: {filepath}")
            
    def process_csv_stream(self, filepath: str) -> Iterator[dict]:
        """CSV 파일을 스트림으로 처리"""
        try:
            with open(filepath, 'r', encoding='utf-8') as file:
                reader = csv.DictReader(file)
                for row in reader:
                    yield row
        except FileNotFoundError:
            print(f"파일을 찾을 수 없습니다: {filepath}")
            
    def process_json_lines(self, filepath: str) -> Iterator[dict]:
        """JSON Lines 파일을 스트림으로 처리"""
        try:
            with open(filepath, 'r', encoding='utf-8') as file:
                for line in file:
                    try:
                        yield json.loads(line.strip())
                    except json.JSONDecodeError:
                        continue
        except FileNotFoundError:
            print(f"파일을 찾을 수 없습니다: {filepath}")
            
    def memory_mapped_search(self, filepath: str, search_term: bytes) -> list:
        """메모리 매핑을 사용한 파일 검색"""
        results = []
        try:
            with open(filepath, 'rb') as file:
                with mmap.mmap(file.fileno(), 0, access=mmap.ACCESS_READ) as mm:
                    start = 0
                    while True:
                        pos = mm.find(search_term, start)
                        if pos == -1:
                            break
                        
                        # 라인 시작과 끝 찾기
                        line_start = mm.rfind(b'\n', 0, pos) + 1
                        line_end = mm.find(b'\n', pos)
                        if line_end == -1:
                            line_end = len(mm)
                            
                        line = mm[line_start:line_end].decode('utf-8', errors='ignore')
                        results.append((pos, line))
                        start = pos + 1
                        
        except FileNotFoundError:
            print(f"파일을 찾을 수 없습니다: {filepath}")
            
        return results
    
    def batch_process(self, data_iterator: Iterator[Any], 
                     batch_size: int = 1000) -> None:
        """배치 단위로 데이터 처리"""
        batch = []
        for item in data_iterator:
            batch.append(item)
            
            if len(batch) >= batch_size:
                self._process_batch(batch)
                batch = []  # 메모리 해제
                
        # 마지막 배치 처리
        if batch:
            self._process_batch(batch)
            
    def _process_batch(self, batch: list) -> None:
        """배치 데이터 처리 (오버라이드 가능)"""
        print(f"배치 처리: {len(batch)}개 항목")
        # 실제 처리 로직 구현

# 사용 예제
def create_test_files():
    """테스트 파일 생성"""
    # 대용량 텍스트 파일 생성
    with open('large_text.txt', 'w') as f:
        for i in range(100000):
            f.write(f"Line {i}: This is test data for memory efficient processing.\n")
    
    # CSV 파일 생성
    with open('test_data.csv', 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['id', 'name', 'value'])
        for i in range(50000):
            writer.writerow([i, f'name_{i}', i * 10])
    
    # JSON Lines 파일 생성
    with open('test_data.jsonl', 'w') as f:
        for i in range(30000):
            json.dump({'id': i, 'data': f'test_{i}'}, f)
            f.write('\n')

def main():
    # 테스트 파일 생성
    create_test_files()
    
    processor = EfficientDataProcessor()
    
    # 텍스트 파일 청크 처리
    print("텍스트 파일 처리:")
    chunk_count = 0
    for chunk in processor.process_large_text_file('large_text.txt'):
        chunk_count += 1
        if chunk_count <= 3:  # 처음 3개 청크만 출력
            print(f"청크 {chunk_count}: {len(chunk)} 문자")
    
    # CSV 스트림 처리
    print(f"\nCSV 파일 처리:")
    row_count = 0
    for row in processor.process_csv_stream('test_data.csv'):
        row_count += 1
        if row_count <= 5:  # 처음 5행만 출력
            print(f"행 {row_count}: {row}")
    
    # 메모리 매핑 검색
    print(f"\n메모리 매핑 검색:")
    results = processor.memory_mapped_search('large_text.txt', b'Line 100:')
    for pos, line in results[:3]:  # 처음 3개 결과만 출력
        print(f"위치 {pos}: {line[:50]}...")
    
    # 청리
    os.remove('large_text.txt')
    os.remove('test_data.csv') 
    os.remove('test_data.jsonl')
    
    print("\n처리 완료!")

if __name__ == "__main__":
    main()

프로젝트 3: 실시간 메모리 모니터

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
import psutil
import time
import threading
import matplotlib.pyplot as plt
from collections import deque
import tkinter as tk
from tkinter import ttk
import queue

class RealTimeMemoryMonitor:
    def __init__(self, max_points=100):
        self.max_points = max_points
        self.memory_data = deque(maxlen=max_points)
        self.time_data = deque(maxlen=max_points)
        self.cpu_data = deque(maxlen=max_points)
        self.running = False
        self.process = psutil.Process()
        
    def get_system_info(self):
        """시스템 정보 수집"""
        memory = psutil.virtual_memory()
        cpu_percent = psutil.cpu_percent(interval=0.1)
        process_memory = self.process.memory_info().rss / 1024 / 1024  # MB
        
        return {
            'total_memory': memory.total / 1024 / 1024 / 1024,  # GB
            'used_memory': memory.used / 1024 / 1024 / 1024,    # GB
            'memory_percent': memory.percent,
            'cpu_percent': cpu_percent,
            'process_memory': process_memory
        }
    
    def start_monitoring(self, interval=1.0):
        """모니터링 시작"""
        self.running = True
        
        def monitor_loop():
            start_time = time.time()
            while self.running:
                current_time = time.time() - start_time
                system_info = self.get_system_info()
                
                self.time_data.append(current_time)
                self.memory_data.append(system_info['memory_percent'])
                self.cpu_data.append(system_info['cpu_percent'])
                
                time.sleep(interval)
        
        self.monitor_thread = threading.Thread(target=monitor_loop)
        self.monitor_thread.daemon = True
        self.monitor_thread.start()
    
    def stop_monitoring(self):
        """모니터링 중지"""
        self.running = False
        if hasattr(self, 'monitor_thread'):
            self.monitor_thread.join()
    
    def plot_data(self):
        """데이터 시각화"""
        if not self.time_data:
            return
            
        plt.figure(figsize=(12, 8))
        
        # 메모리 사용률 그래프
        plt.subplot(2, 1, 1)
        plt.plot(list(self.time_data), list(self.memory_data), 'b-', label='Memory %')
        plt.ylabel('Memory Usage (%)')
        plt.title('Real-time System Monitor')
        plt.legend()
        plt.grid(True)
        
        # CPU 사용률 그래프
        plt.subplot(2, 1, 2)
        plt.plot(list(self.time_data), list(self.cpu_data), 'r-', label='CPU %')
        plt.xlabel('Time (seconds)')
        plt.ylabel('CPU Usage (%)')
        plt.legend()
        plt.grid(True)
        
        plt.tight_layout()
        plt.show()

class MemoryMonitorGUI:
    def __init__(self):
        self.monitor = RealTimeMemoryMonitor()
        self.root = tk.Tk()
        self.root.title("실시간 메모리 모니터")
        self.root.geometry("400x300")
        
        self.create_widgets()
        self.update_data()
        
    def create_widgets(self):
        # 메인 프레임
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
        
        # 시스템 정보 표시
        self.info_frame = ttk.LabelFrame(main_frame, text="시스템 정보", padding="5")
        self.info_frame.grid(row=0, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=5)
        
        self.memory_label = ttk.Label(self.info_frame, text="메모리: ")
        self.memory_label.grid(row=0, column=0, sticky=tk.W)
        
        self.cpu_label = ttk.Label(self.info_frame, text="CPU: ")
        self.cpu_label.grid(row=1, column=0, sticky=tk.W)
        
        self.process_label = ttk.Label(self.info_frame, text="프로세스 메모리: ")
        self.process_label.grid(row=2, column=0, sticky=tk.W)
        
        # 제어 버튼
        self.start_button = ttk.Button(main_frame, text="모니터링 시작", 
                                      command=self.start_monitoring)
        self.start_button.grid(row=1, column=0, pady=10)
        
        self.stop_button = ttk.Button(main_frame, text="모니터링 중지", 
                                     command=self.stop_monitoring)
        self.stop_button.grid(row=1, column=1, pady=10)
        
        self.plot_button = ttk.Button(main_frame, text="그래프 보기", 
                                     command=self.show_plot)
        self.plot_button.grid(row=2, column=0, columnspan=2, pady=10)
        
    def start_monitoring(self):
        """모니터링 시작"""
        self.monitor.start_monitoring()
        self.start_button.config(state='disabled')
        self.stop_button.config(state='normal')
        
    def stop_monitoring(self):
        """모니터링 중지"""
        self.monitor.stop_monitoring()
        self.start_button.config(state='normal')
        self.stop_button.config(state='disabled')
        
    def show_plot(self):
        """그래프 표시"""
        self.monitor.plot_data()
        
    def update_data(self):
        """데이터 업데이트"""
        try:
            system_info = self.monitor.get_system_info()
            
            self.memory_label.config(
                text=f"메모리: {system_info['memory_percent']:.1f}% "
                     f"({system_info['used_memory']:.1f}/{system_info['total_memory']:.1f} GB)"
            )
            
            self.cpu_label.config(text=f"CPU: {system_info['cpu_percent']:.1f}%")
            
            self.process_label.config(
                text=f"프로세스 메모리: {system_info['process_memory']:.1f} MB"
            )
            
        except Exception as e:
            print(f"데이터 업데이트 오류: {e}")
        
        # 1초마다 업데이트
        self.root.after(1000, self.update_data)
        
    def run(self):
        """GUI 실행"""
        self.root.mainloop()

# 사용 예제
if __name__ == "__main__":
    # 콘솔 모드
    print("실시간 메모리 모니터 (콘솔 모드)")
    monitor = RealTimeMemoryMonitor()
    monitor.start_monitoring()
    
    try:
        time.sleep(10)  # 10초 동안 모니터링
        monitor.plot_data()
    except KeyboardInterrupt:
        pass
    finally:
        monitor.stop_monitoring()
    
    # GUI 모드 (주석 해제하여 사용)
    # gui = MemoryMonitorGUI()
    # gui.run()

요약

이 챕터에서는 파이썬의 메모리 관리 메커니즘을 깊이 있게 다뤘습니다:

메모리 모델: 객체 참조, 참조 카운팅, 메모리 레이아웃
가비지 컬렉션: 순환 참조 해결, GC 제어, 약한 참조
메모리 프로파일링: tracemalloc, memory_profiler 활용
최적화 기법: slots, 제너레이터, 효율적 자료구조
대용량 데이터: 메모리 매핑, 청크 처리, 스트리밍
누수 탐지: 일반적인 패턴과 해결 방법

이러한 기법들을 활용하면 메모리 효율적인 파이썬 프로그램을 작성할 수 있습니다.

체크리스트

파이썬 메모리 모델 이해
가비지 컬렉션 메커니즘 파악
메모리 누수 진단 능력
메모리 최적화 기법 적용
메모리 모니터링 도구 활용

다음 단계

메모리 관리를 마스터했다면, 네트워크 프로그래밍과 통신 기술을 학습합니다.