30년동안 열심히 돈을 모은 상근이는 드디어 아파트 하나를 구매했다. 상근이는 친구들을 최대한 많이 초대해 집들이를 하려고 한다.

집들이를 하기 전에 모든 사람이 앉을 수 있는 직사각형 식탁을 하나 사려고 한다. 식탁에 앉을 수 있는 사람의 수는 식탁의 둘레 길이와 같다. (네 변의 길이의 합)

상근이는 되도록 큰 식탁을 구매해서 되도록 많은 사람들과 같이 저녁을 먹을 수 있게 하려고 한다. 식탁은 항상 아파트의 변에 평행하게 놓아야 한다.

아파트의 레이아웃이 주어졌을 때, 상근이가 초대할 수 있는 사람의 수를 구하는 프로그램을 작성하시오.

입력 첫째 줄에 아파트의 크기를 나타내는 R과 C가 주어진다. (1 ≤ R, C ≤ 400)

다음 R개 줄에는 C개의 문자가 주어지며, 빈 칸은 '.', 막힌 칸은 'X'로 주어진다.

상근이는 오직 빈 칸에만 식탁을 놓을 수 있다. 또, 사람의 크기는 매우 작다고 생각하면 된다.

출력 첫째 줄에 상근이가 초대할 수 있는 사람의 수를 출력한다.

예제 입력 1 2 2 .. .. 예제 출력 1 7 예제 입력 2 4 4 X.XX X..X ..X. ..XX 예제 출력 2 9 예제 입력 3 3 3 X.X .X. X.X 예제 출력 3 3

풀이 1. 완전 탐색

우선 가장 빠르게 생각할 수 있는 풀이인 완전 탐색 풀이 입니다. 모든 직사각형을 확인해서 최대 사람 수를 구하는 방식입니다.

import java.io.*;
import java.util.*;

public class Main {

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringTokenizer st = new StringTokenizer(br.readLine());
        int R = Integer.parseInt(st.nextToken());
        int C = Integer.parseInt(st.nextToken());

        char[][] grid = new char[R][C];
        for (int i = 0; i < R; i++) {
            grid[i] = br.readLine().toCharArray();
        }

        int ans = 0;

        for (int r1 = 0; r1 < R; r1++) {
            for (int c1 = 0; c1 < C; c1++) {
                for (int r2 = r1; r2 < R; r2++) {
                    for (int c2 = c1; c2 < C; c2++) {
                        boolean ok = true;

                        for (int i = r1; i <= r2 && ok; i++) {
                            for (int j = c1; j <= c2; j++) {
                                if (grid[i][j] == 'X') {
                                    ok = false;
                                    break;
                                }
                            }
                        }

                        if (ok) {
                            int h = r2 - r1 + 1;
                            int w = c2 - c1 + 1;
                            int a = 2 * (h + w) - 1;
                            ans = Math.max(ans, a);
                        }
                    }
                }
            }
        }

        System.out.println(ans);
    }
}

정말 단순하게 모든 좌표를 돌아서 (4중 for문;;) 중요한 2*(h+w) - 1(자기 자신)의 최댓값을 갱신하는 풀이입니다.

시간복잡도가 O(R^2 * C^2) 로 당연히 시간 초과가 나게 됩니다.

완전 탐색을 구현하고 나니 직사각형 내부 검사를 최적화하는 방법이 필요하다 생각이 들었습니다.

풀이 2. 누적합

행 구간을 top ~ bottom 을 하나 고정하면 높이가 h = bottom - top + 1로 정해집니다. 그 후 각 열 j에 대해,

• top~bottom 구간에 X가 하나도 없으면 사용 가능
• 하나라도 있으면 사용 불가

로 판단하면 됩니다. 이를 위해 열 방향 누적합을 만들게 되는데

sum[i + 1][j] = 0행부터 i행까지 j열에 있는 X 개수

이렇게 두면 top ~ bottom 구간을 j열에 X가 있는지 여부를

sum[bottom + 1][j] - sum[top][j]

로 O(1)으로 알 수 있습니다.

이러면 행 시작점을 top : R, 행 끝점 bottom : R, 각 경우 스캔 : C 로 O(R^2 * C)의 시간복잡도가 됩니다.

import java.io.*;
import java.util.*;

public class Main {

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringTokenizer st = new StringTokenizer(br.readLine());
        int R = Integer.parseInt(st.nextToken());
        int C = Integer.parseInt(st.nextToken());

        char[][] grid = new char[R][C];
        for (int i = 0; i < R; i++) {
            grid[i] = br.readLine().toCharArray();
        }

        int[][] sum = new int[R + 1][C];
        for (int i = 0; i < R; i++) {
            for (int j = 0; j < C; j++) {
                sum[i + 1][j] = sum[i][j] + (grid[i][j] == 'X' ? 1 : 0);
            }
        }

        int ans = 0;

        for (int top = 0; top < R; top++) {
            for (int bottom = top; bottom < R; bottom++) {
                int h = bottom - top + 1;
                int w = 0;

                for (int j = 0; j < C; j++) {
                    int block = sum[bottom + 1][j] - sum[top][j];

                    if (block == 0) {
                        w++;
                        int a = 2 * (h + w) - 1;
                        ans = Math.max(ans, a);
                    } else {
                        w = 0;
                    }
                }
            }
        }

        System.out.println(ans);
    }
}

풀이 3. 모노톤 스택

또 다른 방식으로 각 행을 바닥으로 생각하고 각 열마다 위로 연속된 빈 칸의 개수를 height[j]로 관리합니다.

예를 들어 현재 행이 i일 때

• grid[i][j] == '.' 이면 height[j]++
• grid[i][j] == 'X' 이면 height[j] = 0

이렇게 판단하면 height[jh]는 현재 행 기준으로 막대 그래프 모양이 그려집니다. 이 막대 그래프에서 만들 수 있는 직사각형 중 최댓값을 찾으면 정답이 됩니다.

각 위치에서 왼쪽으로 처음 나오는 더 작은 높이와 오른쪽으로 처음 나오는 더 작은 높이를 찾아서 너비를 구하게 됩니다. 이 두 경계를 알면 j를 최소 높이로 하는 최대 너비를 바로 구할 수 있게 됩니다.

l[j] : j의 왼쪽에서 처음 만나는 더 작은 높이의 인덱스 r[j] : j의 오른쪽에서 처음 만나는 더 작은 높이의 인덱스

그러면 l[j] + 1 부터 r[j] - 1 까지는 모두 height[j] 이상이므로, 이 구간 전체를 너비로 사용할 수 있습니다.

w = r[j] - l[j] - 1;

그리고 초대 가능한 사람 수는

2 * (height[j] + w) - 1

로 구하게 됩니다.

이때 l, r을 매번 선형 탐색으로 구하면 또 C만큼 걸리기 때문에 높이가 증가하는 형태를 유지하는 모노톤 스택을 활용할 수 있습니다.

import java.io.*;
import java.util.*;

public class Main {

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringTokenizer st = new StringTokenizer(br.readLine());
        int R = Integer.parseInt(st.nextToken());
        int C = Integer.parseInt(st.nextToken());
        char[][] grid = new char[R][C];
        for(int i = 0; i < R; i++) {
            grid[i] = br.readLine().toCharArray();
        }

        int[] height = new int[C];
        int ans = 0;

        for(int i = 0; i < R; i++) {
            for(int j = 0; j < C; j++) {
                if(grid[i][j] == '.') height[j]++;
                else height[j] = 0;
            }

            int[] l = new int[C];
            int[] r = new int[C];
            Deque<Integer> s = new ArrayDeque<>();

            for(int j = 0; j < C; j++) {
                while(!s.isEmpty() && height[s.peek()] >= height[j]) {
                    s.pop();
                }
                l[j] = s.isEmpty() ? -1 : s.peek();
                s.push(j);
            }

            s.clear();

            for(int j = C - 1; j >= 0; j--) {
                while(!s.isEmpty() && height[s.peek()] >= height[j]) {
                    s.pop();
                }
                r[j] = s.isEmpty() ? C : s.peek();
                s.push(j);
            }

            for(int j = 0; j < C; j++) {
                if(height[j] == 0) continue;
                int w = r[j] - l[j] - 1;
                int a = 2 * (height[j] + w) - 1;
                ans = Math.max(ans, a);
            }
        }

        System.out.println(ans);
    }
}

이러면 아마...O(R*C)라 생각했는데.....

메모리만 더 쓰고 시간이 똑같더라구요!

왜 시간이 그대로지?!!?!?!?!?? ...혹시 아신다면 댓글로 지혜를 나눠주십쇼.

마무리

요즘 알고리즘 풀면서 최대한 다양한 방식으로 풀어보려 하는데 단순 구현에서 점점 최적화 하는 방식을 떠올리는 것이 뭔가 더 어렵게 느껴집니다. 이전엔 사고의 흐름이 이거 맞겠지~? 하는 느낌이라면 요즘엔 진짜 검산하고 들어가는 느낌이라 정확도는 높은데 시간이 더 걸리는 느낌. 장단점이 있는 것 같습니다.

DAsP에서 다루는 데이터 요건 분석은 데이터를 수집하고, 조사하고, 분석하고, 검증함으로써 데이터의 일관성과 추적 가능성을 확보하는 것을 목표로 한다. 이 단계를 제대로 거쳐야 이후의 개념 데이터 모델, 논리 데이터 모델, 물리 데이터 모델 역시 흔들리지 않는다. 반대로 이 단계가 부실하면 설계와 개발, 테스트, 운영 전반에서 끝없는 수정이 반복된다.

1장. 정보 요구사항 개요

1.1 정보 요구사항?

정보 요구사항은 사용자가 업무를 수행하는 과정에서 필요로 하는 정보와 기능, 그리고 이를 시스템에 반영하기 위해 요청하는 내용을 의미한다. 향후 시스템이 어떤 데이터를 저장하고, 어떤 규칙으로 처리하며, 어떤 방식으로 제공해야 하는지를 결정하는 과정이다.

예를 들어 영업 부서가 “고객별 월간 매출 현황을 보고 싶다”라고 요청했다고 하자. 이 문장은 겉보기에는 단순한 보고서 요구처럼 보이지만, 데이터 관점에서는 이미 여러 요소가 내포되어 있다.

• 고객이라는 관리 대상이 존재
• 매출이라는 사실 데이터를 구조적으로 관리
• 월 단위 집계를 위한 시간 기준 요구
• 확실한 고객과 매출 사이의 관계
• 조회 권한과 성능 요건 고려

즉 하나의 요구사항 안에는 이미 업무 개념, 데이터 구조, 처리 규칙, 통제 기준이 함께 들어 있다. 따라서 데이터 요건 분석은 그 안에 숨어 있는 의미를 데이터 관점에서 해석하는 과정이다.

1.2 정보 요구사항의 역할

정보 요구사항은 프로젝트와 시스템 개선 과정에서 다음과 같은 역할을 수행한다.

1. 개발 범위 정의
   무엇을 구현해야 하고, 무엇은 범위 밖인지 구분하는 기준이 된다.

2. 데이터 설계의 근거
   어떤 엔터티가 필요한지, 어떤 속성이 필요한지, 데이터 간 관계는 어떻게 정의해야 하는지를 도출할 수 있다.

3. 사용자와 개발자 사이의 기준점
   사용자는 자신이 원하는 것이 반영되었는지 판단하고, 개발자는 무엇을 구현해야 하는지 판단한다.

4. 검증과 테스트의 기준
   요구사항이 명확해야 테스트 케이스도 명확해진다.

5. 변경 관리의 기준
   향후 추가 요구나 변경 요청이 발생했을 때 어떤 항목이 바뀌었는지 추적할 수 있다.

1.3 정보 요구사항의 유형

정보 요구사항은 여러 기준으로 분류할 수 있지만, 데이터 요건 분석 관점에서는 보통 다음과 같이 구분된다.

1. 기능적 요구사항 가장 일반적인 요구사항으로 시스템이 무엇을 해야 하는가에 대한 요구 ex) 고객 등록, 주문 생성, 매출 조회, 정산 처리 등

2. 비기능적 요구사항 시스템이 어느 수준으로 동작해야 하는가에 대한 요구 ex) 성능, 보안, 가용성, 확장성, 백업, 접근통제 등

3. 데이터 요구사항 어떤 데이터를 수집하고 저장하고 활용해야 하는가에 대한 요구 ex) 고객 식별값, 거래일시, 상태코드, 이력 보관 여부, 보존 기간 등

4. 인터페이스 요구사항 다른 시스템과 어떤 방식으로 연계해야 하는가에 대한 요구 ex) 외부 기관 연계, 배치 파일 교환, API 호출, 메시지 기반 송수신 등

실제로는 이 유형들이 깔끔하게 분리되어 나타나지 않고 섞여 있는 경우가 대다수다. 예를 들어 “외부 시스템에서 회원 정보를 받아 실시간 반영하고, 오류 시 재처리하며, 개인정보는 암호화해야 한다”는 문장에는 기능, 인터페이스, 데이터, 보안 요구가 모두 함께 들어 있다. 따라서 요구사항을 수집한 후에는 이를 적절히 분해하고 정리하는 작업을 거쳐야한다.

1.4 정보 요구사항 생명주기

정보 요구사항은 한 번 수집하고 끝나는 정적인 대상이 아니다. 일반적으로 다음과 같은 생명주기를 가진다.

• 요구사항 수집
• 요구사항 정리 및 분석
• 요구사항 상세화
• 요구사항 검증
• 요구사항 확정 및 변경 관리

이 흐름은 소프트웨어 생명주기와도 밀접하게 연결된다. 요구사항이 애매하면 설계가 흔들리고, 설계가 흔들리면 개발이 흔들리며, 개발이 흔들리면 테스트 비용과 운영 리스크가 크게 증가한다. 그래서 요구사항 단계는 뒤 단계보다 비용은 적게 들 수 있지만, 프로젝트 전체에 미치는 영향력은 오히려 훨씬 크다.

2장. 정보 요구사항 조사

2.1 정보 요구사항 조사의 목적

정보 요구사항 조사 단계는 실제 현업과 시스템 환경 속에서 필요한 요구를 수집하고 드러내는 과정이다. 개요 단계가 “무엇을 어떤 관점에서 볼 것인가”를 정하는 준비였다면, 조사 단계는 “실제로 어떤 요구가 존재하는가”를 찾는 단계다.

이 단계가 중요한 이유는, 사용자의 요구가 처음부터 명확하게 정리되어 있지 않은 경우가 대부분이기 때문이다. 사용자는 자신이 불편한 점은 말할 수 있어도, 그것을 데이터 구조나 시스템 기능 수준으로 체계화해서 표현하지는 못하는 경우가 많다. 따라서 요구 사항에 암묵적으로 포함된 내용을 분석하는 과정이 필수다.

2.2 조사 대상

정보 요구사항 조사의 대상은 단순히 사용자 인터뷰만을 의미하지 않는다. 실제로는 다양한 출처가 함께 고려되어야 한다.

• 현업 사용자 인터뷰
• 중간 관리자 및 의사결정자 인터뷰
• 기존 시스템 화면과 메뉴 구조
• 업무 매뉴얼, 보고서, 전표, 입력 양식
• 데이터 사전, 테이블 정의서, ERD
• 외부 연계 문서와 인터페이스 명세
• 장애 이력, 개선 요청 이력
• 현행 프로그램 및 배치 흐름

현실에선 사용자가 직접 말해 주는 요구사항보다 문서와 시스템 분석을 통해 더 중요한 단서를 발견하는 경우가 많다. 왜냐하면 사용자는 현재 시스템에 익숙해져 있어서 불편함을 당연하게 받아들이는 경우가 있기 때문이다. 반면 화면 구조, 입력 양식, 보고서 체계, 코드 정의 등을 살펴보면 중복 입력이나 용어 불일치, 책임 주체 모호성 같은 문제가 더 잘 드러난다.

2.3 정보 요구사항 수집 기법

2.3.1 관련 문서 조사

가장 기본적이면서도 중요한 방법이다.
업무 매뉴얼, 보고서, 규정집, 시스템 설계서, 인터페이스 문서, 화면 정의서, 코드 체계 등을 검토하여 현재 업무와 데이터 구조를 파악한다.

문서 조사의 장점은 공식 기준과 실제 관행을 동시에 확인할 수 있다는 점이다. 예를 들어 시스템에서는 “회원상태”라는 용어를 쓰고 있는데 보고서에서는 “고객등급”이라는 용어를 쓴다면, 두 용어가 같은 개념인지 다른 개념인지 확인이 필요하다. 이러한 차이는 나중에 데이터 표준화와 모델링에서 매우 중요한 이슈가 된다.

또한 문서 조사 단계에서는 수집한 자료를 무작정 쌓아두는 것이 아니라, 분석 목적에 맞게 구조화해야 한다. 예를 들어 업무 문서, 시스템 문서, 데이터 문서, 외부 연계 문서를 분리해 관리하면 이후 분석 효율이 높아진다.

2.3.2 사용자 면담

사용자 면담은 현업 실무자와 직접 대화하면서 요구를 끌어내는 방식이다. 이때 중요한 것은 “무엇이 필요하십니까?”라고 단순히 묻는 것이 아니라, 현재 어떤 업무를 어떤 순서로 처리하고 있으며, 어떤 정보를 기준으로 판단하고, 어떤 지점에서 불편이 발생하는지를 파악하는 것이다.

예를 들어 다음과 같은 질문이 효과적이다.

• 현재 업무는 어떤 절차로 진행됩니까?
• 이 과정에서 직접 입력하거나 확인하는 정보는 무엇입니까?
• 현재 가장 자주 조회하는 화면이나 보고서는 무엇입니까?
• 지금 시스템에서 가장 불편한 점은 무엇입니까?
• 누락되면 가장 문제가 되는 데이터는 무엇입니까?
• 다른 부서와 데이터를 주고받을 때 자주 충돌하는 항목은 무엇입니까?

좋은 면담은 기능 요청을 받아 적는 자리가 아니라, 업무 흐름과 데이터 사용 행태를 이해하는 자리다. 특히 면담 과정에서는 사용자의 표현을 그대로 사실로 받아들이기보다, 그 뒤에 있는 업무 목적과 데이터 구조를 파악해야 한다.

2.3.3 워크숍

워크숍은 여러 부서가 동시에 참여해 공통 요구사항을 조정하고 합의점을 찾는 방식이다. 부서 간 이해관계가 얽혀 있거나, 동일한 데이터를 서로 다르게 해석하는 경우에 매우 유용하다.

예를 들어 영업 부서, 정산 부서, 재무 부서가 모두 “주문 상태”를 다른 기준으로 사용하고 있다면, 개별 면담만으로는 일관된 정의를 만들기 어렵다. 이럴 때 워크숍을 통해 상태 정의, 변경 시점, 책임 부서를 함께 조정해야 한다.

워크숍의 장점은 여러 관점을 한 자리에서 비교할 수 있다는 점이다. 반면 준비 없이 진행하면 목소리가 큰 부서의 의견만 반영될 위험이 있으므로, 사전 안건 정리와 진행자 역할이 매우 중요하다.

2.3.4 현행 업무 조사서

전체 부서를 대상으로 동일한 형식의 조사서를 배포하여 업무와 정보 요구를 수집하는 방식이다. 대규모 조직에서 광범위한 요구를 빠르게 파악할 때 유용하다.

이 방식의 핵심은 양식이 단순하고 명확해야 한다는 점이다. 작성이 어렵거나 기준이 불분명하면 회수율도 떨어지고 내용의 품질도 낮아진다. 따라서 조사서에는 작성 예시와 용어 설명을 포함해 혼선을 줄여야 한다.

2.3.5 현행 시스템 및 데이터 구조 조사

현재 운영 중인 프로그램, 테이블, 인터페이스 구조를 직접 분석하는 방식이다. 문서가 오래되었거나 실제 시스템과 일치하지 않는 경우에는 이 방법이 특히 중요하다. 경우에 따라서는 테이블 스키마, 소스 코드, 배치 로그 등을 직접 확인하는 역공학적 접근이 필요할 수도 있다.

이 과정은 단순히 현재 구조를 확인하는 데 그치지 않는다. 기존 데이터 구조의 문제점, 중복 저장, 정합성 오류, 비표준 용어 사용 등을 식별하는 데에도 큰 도움이 된다.

2.4 정보 요구사항 통합과 분할

조사 단계에서 수집된 정보 요구사항은 그대로 분석에 사용하기 어렵다. 왜냐하면 부서와 사용자별로 표현 방식이 다르고, 중복되거나 충돌하는 내용이 섞여 있기 때문이다. 따라서 조사 후에는 반드시 요구사항을 통합하고 분할하는 작업이 필요하다.

통합이 필요한 경우

• 표현만 다를 뿐 본질적으로 같은 요구인 경우
• 여러 부서가 동일한 데이터 구조 변경을 요구하는 경우
• 유사 보고서나 유사 화면 요청을 하나의 공통 기능으로 묶을 수 있는 경우

분할이 필요한 경우

• 하나의 요구 안에 여러 기능이 섞여 있는 경우
• 기능 요구와 데이터 요구, 성능 요구가 함께 섞여 있는 경우
• 책임 부서나 처리 시점이 다른 요구가 하나로 묶여 있는 경우

예를 들어 “주문 조회 화면을 개선하고, 주문 취소 이력을 남기고, 외부 정산 시스템으로 전송해 주세요”라는 요구는 하나처럼 보이지만 실제로는 다음과 같이 분리할 수 있다.

• 주문 조회 기능 개선
• 취소 이력 데이터 관리
• 외부 시스템 인터페이스 처리

이처럼 통합과 분할은 단순 편집 작업이 아니라, 이후 설계와 추적성을 확보하기 위한 구조화 작업이다.

2.5 우선순위 분석

현실에서는 모든 요구를 동시에 반영할 수 없다. 예산, 일정, 인력은 항상 제한되어 있기 때문이다. 따라서 요구사항 조사 이후에는 우선순위를 판단해야 한다.

우선순위 판단 기준으로는 보통 다음 요소가 사용된다.

• 업무 기여도
• 긴급성
• 전사적 파급효과
• 데이터 품질 영향도
• 구현 난이도
• 법규 및 컴플라이언스 대응 필요성
• 비용 대비 효과

데이터 아키텍처 관점에서는 단순히 특정 부서가 강하게 요청했다는 이유만으로 우선순위를 정하면 안 된다. 더 중요한 것은 해당 요구가 전사 데이터의 일관성, 핵심 업무 흐름, 마스터 데이터 관리, 인터페이스 안정성에 어떤 영향을 미치는가이다.

3장. 정보 요구사항 분석

3.1 조사와 분석의 차이

많은 경우 조사와 분석이 혼동되지만, 두 단계는 분명히 다르다.
조사는 요구를 수집하는 단계이고, 분석은 수집된 요구를 해석하고 구조화하여 데이터 설계가 가능한 수준으로 정리하는 단계이다.

즉 조사 단계에서는 “무슨 말이 나왔는가”가 중요하고, 분석 단계에서는 “그 말을 데이터와 프로세스 관점에서 어떻게 이해해야 하는가”가 중요하다.

3.2 분석의 목적

정보 요구사항 분석의 목적은 단순히 문장을 정리하는 데 있지 않다. 핵심 목적은 다음과 같다.

• 요구사항의 의미를 명확히 한다.
• 데이터 관점에서 필요한 관리 대상을 식별한다.
• 업무 흐름과 데이터 흐름의 관계를 정리한다.
• 중복과 충돌을 해소한다.
• 데이터 모델링이 가능한 수준으로 요구를 상세화한다.

예를 들어 “회원의 최근 활동을 보고 싶다”라는 요구는 분석 과정을 거치며 훨씬 더 구체적인 질문으로 바뀌게 된다.

• 회원의 정의는 무엇인가?
• 최근 활동의 범위는 어디까지인가?
• 로그인, 구매, 문의, 리뷰 작성 중 무엇을 포함하는가?
• 최근의 기준은 7일인가, 30일인가?
• 활동은 이력 데이터로 관리하는가, 요약 데이터로 관리하는가?
• 개인정보와 연결될 때 보존 정책은 어떻게 되는가?

이처럼 분석은 막연한 요구를 정의, 범위, 기준, 관계, 제약조건 수준으로 구체화하는 과정이다.

3.3 분석 대상 시스템 및 자료 확인

분석에 들어가기 전에는 현재 시스템과 관련 자료의 신뢰성을 먼저 검토해야 한다. 일반적으로 다음 네 가지 관점이 중요하다.

• 유용성 해당 문서나 자료가 현재 요구사항 분석에 실제로 도움이 되는가

• 완전성 필요한 정보가 빠짐없이 포함되어 있는가

• 정확성 문서의 내용과 실제 운영 중인 시스템이 일치하는가

• 유효성 자료가 최신 상태를 반영하고 있는가

실무에서는 설계서보다 실제 데이터베이스 구조나 운영 중인 화면이 더 정확한 경우가 적지 않다. 따라서 문서만 믿고 분석을 진행하는 것은 위험하다. 필요하다면 운영 DB 스키마, 로그, 인터페이스 메시지까지 직접 확인해야 한다.

3.4 업무 분석과 데이터 식별

정보 요구사항 분석의 핵심 결과 중 하나는 관리해야 할 데이터 대상을 식별하는 것이다. 즉 어떤 엔터티가 필요한지, 어떤 속성이 필요한지, 데이터 간 어떤 관계가 존재하는지를 도출하게 된다.

예를 들어 “주문 취소 사유를 관리하고 싶다”라는 요구를 분석하면 다음과 같은 데이터 요소가 도출될 수 있다.

• 주문
• 주문 상태
• 주문 취소 이력
• 취소 사유 코드
• 취소 요청자
• 취소 일시
• 환불 처리 여부

이때 중요한 것은 단순히 항목을 많이 뽑는 것이 아니라, 무엇이 독립적인 관리 대상인지를 판단하는 것이다. 예를 들어 취소 사유가 단순 문자열 입력인지, 아니면 전사 표준코드 관리가 필요한 값인지에 따라 데이터 구조는 달라진다. 또한 주문 상태를 주문 엔터티의 단순 속성으로 둘 것인지, 상태 이력 엔터티를 별도로 둘 것인지 역시 업무 요구에 따라 달라진다.

3.5 프로세스 분석과 데이터 흐름

데이터는 언제나 업무 프로세스와 연결되어 있다. 따라서 요구사항 분석에서는 데이터를 고립된 대상으로 보지 않고, 어떤 프로세스에서 생성되고 조회되고 수정되고 삭제되는지를 함께 파악해야 한다.

예를 들어 주문 관리 업무를 살펴보면 다음과 같이 세분화할 수 있다.

• 주문 접수
• 결제 확인
• 주문 확정
• 배송 처리
• 주문 취소
• 환불 처리
• 정산 반영

이런 흐름을 따라가면 “주문”이라는 데이터가 단순히 한 번 저장되고 끝나는 것이 아니라, 여러 단계에서 상태를 바꾸며 활용된다는 사실을 알 수 있다. 따라서 요구사항 분석 단계에서는 단순히 주문 테이블 하나만 정의하는 것이 아니라, 상태 전이, 이력 관리, 제약조건, 참조 관계까지 함께 고려해야 한다.

3.6 SWOT과 RAEW 관점의 해석

DAsP에서는 요구사항 분석 과정에서 경영 환경과 조직 구조를 이해하기 위한 도구로 SWOT과 RAEW 같은 프레임워크를 함께 활용하기도 한다.

1. SWOT 분석 SWOT은 조직이나 시스템의 내외부 환경을 강점, 약점, 기회, 위협으로 나누어 보는 방식이다. 데이터 관점에서 보면 다음과 같이 활용할 수 있다.

• 강점: 이미 잘 관리되고 있는 핵심 데이터 자산
• 약점: 중복 데이터, 품질 저하, 표준 미흡
• 기회: 데이터 통합을 통한 신규 서비스 가능성
• 위협: 규제 강화, 보안 리스크, 외부 연계 복잡성 증가 즉 SWOT은 단순한 경영 분석 도구가 아니라, 데이터 요구가 어떤 배경에서 등장했는지 이해하는 데 도움이 된다.

2. RAEW 분석 RAEW는 Responsibility, Authority, Expertise, Work의 관점에서 역할 구조를 분석하는 방식이다. 데이터 요건 분석에서는 특히 누가 데이터를 생성하고, 누가 수정하고, 누가 책임지는지를 이해하는 데 유용하다. 예를 들어 어떤 고객 마스터 데이터에 대해 영업 부서가 책임을 갖고 있지만 실제 입력은 타 부서가 하고 있다면, 데이터 품질 문제가 발생할 가능성이 높다.

3.7 정보 요구사항 명세서 작성

분석을 통해 정리된 내용은 결국 명세서 형태로 구체화되어야 한다. 정보 요구사항 명세서는 사용자, 분석가, 개발자, 테스터 모두가 참조하는 공식 기준 문서다. 작성이 잘된 명세서는 다음과 같은 특징을 가진다.

• 모호한 표현 X
• 누락 X
• 용어의 일관성
• 구체적인 처리 기준
• 명확한 데이터 항목과 제약조건
• 테스트 가능한 수준

예제)

• “빠르게 조회되어야 한다”는 표현은 좋은 명세가 아니다.
  “최근 1개월 주문 내역 조회는 3초 이내에 완료되어야 한다”처럼 정량적인 표현이 필요하다.

• “개인정보를 안전하게 관리한다”도 모호하다.
  “주민등록번호는 저장 시 암호화하며, 조회 화면에서는 마스킹 처리한다”처럼 구체적으로 명시해야 한다.

4장. 정보 요구 검증

4.1 정보 요구 검증의 필요성

아무리 요구사항을 잘 수집하고 분석하고 명세서까지 작성했다고 해도, 그 결과가 항상 완전하고 정확하다고 볼 수는 없다. 요구사항에는 사용자 해석의 차이, 부서 간 용어 차이, 분석 누락, 기존 시스템과의 충돌 가능성 등이 있을 수 있기 때문이다. 검증 단계는 작성된 요구사항 정의서와 명세서가 사용자의 의도와 전사 데이터 아키텍처 원칙에 부합하는지 최종 확인하는 단계이다.

4.2 검증 기준

정보 요구 검증에서는 일반적으로 다음 네 가지 기준이 있다

• 완전성 필요한 요구가 빠짐없이 반영되었는가를 확인하는 기준이다.
  예를 들어 조회 기능은 정의했지만 생성, 수정, 삭제, 이력 관리, 예외 처리 기준이 빠져 있다면 완전성이 부족하다고 볼 수 있다.

• 정확성 요구가 실제 업무 의도와 맞는가를 확인하는 기준이다.
  사용자는 “거래 취소”를 원했는데 문서에는 “삭제”로 적혀 있다면, 이는 단순한 용어 차이가 아니라 데이터 보존 정책과 감사 추적에 큰 차이를 만드는 심각한 문제다.

• 일관성 문서 전체에서 같은 개념이 동일한 용어와 동일한 규칙으로 표현되고 있는지를 확인하는 기준이다.
  한 곳에서는 고객, 다른 곳에서는 회원, 또 다른 곳에서는 사용자라고 표현하고 있다면 같은 대상을 뜻하더라도 혼란을 야기할 수 있다.

• 안정성 새로운 요구를 반영했을 때 기존 시스템이나 기존 데이터 구조에 과도한 부작용을 일으키지 않는가를 확인하는 기준이다.
  예를 들어 신규 상태값 추가가 기존 인터페이스, 배치, 정산 로직 전반에 영향을 준다면 안정성 검토가 필요하다.

4.3 검증 절차

정보 요구 검증은 보통 다음과 같은 흐름으로 진행된다.

• 검증 범위와 대상 정의
• 검토 회의 계획 수립
• 요구사항 정의서 및 명세서 검토
• 용어와 표준 검토
• 상관분석 수행
• 수정 및 보완 반영
• 베이스라인 확정

여기서 중요한 것은 베이스라인 설정이다. 검증을 마친 요구사항은 공식 기준으로 고정되어야 이후 변경 관리가 가능하다. 그렇지 않으면 개발 중간마다 구두 요청이 계속 섞이면서 통제가 어려워진다.

4.4 요구사항 명세서 검토

명세서 검토는 문서의 문장을 단순히 읽는 것이 아니라, 논리적 흐름과 데이터 구조의 정합성을 함께 확인하는 작업이다. 이 단계에서는 다음과 같은 질문이 중요하다.

• 업무 흐름이 현실과 일치하는가
• 입력과 출력이 논리적으로 연결되는가
• 데이터 정의가 모호하지 않은가
• 누락된 예외 상황은 없는가
• 테스트 가능한 수준으로 작성되었는가

특히 요구사항 명세서는 사용자 입장, 개발자 입장, 테스터 입장을 모두 만족해야 한다. 사용자는 이해할 수 있어야 하고, 개발자는 구현할 수 있어야 하며, 테스터는 검증할 수 있어야 한다.

4.5 요구사항 용어 검증

데이터 아키텍처에서 용어의 일관성은 매우 중요하다. 같은 개념을 부서마다 다른 용어로 부르면 데이터 통합과 표준화가 어려워지고, 결국 품질 문제와 해석 차이로 이어진다.

따라서 검증 단계에서는 다음을 확인해야 한다.

• 여러 문서 내의 용어의 의미 통일성
• 데이터 표준과 도메인 정의 준수
• 코드값과 상태값의 정의가 일관성
• 약어와 축약어 혼동 가능성

4.6 CRUD 매트릭스를 활용한 검증

정보 요구 검증의 대표적인 방법 중 하나가 CRUD 매트릭스이다. CRUD는 Create, Read, Update, Delete를 의미하며, 데이터 엔터티와 기본 프로세스 간의 관계를 행렬 형태로 정리하여 요구사항의 누락과 충돌을 점검한다.

예를 들어 다음과 같은 구조를 생각해 볼 수 있다.

• 어떤 데이터가 어떤 프로세스에서 사용되는가
• 생성은 되는데 종료나 삭제 처리가 없는 데이터는 없는가
• 중요한 데이터인데 어떤 프로세스와도 연결되지 않은 것은 없는가
• 하나의 데이터가 지나치게 많은 프로세스와 연결되어 있지는 않은가

[CRUD 매트릭스 검증 시 특히 주의할 점]

1. 모든 행과 열은 의미를 가져야 한다 어떤 엔터티가 어떤 프로세스와도 연결되지 않으면, 도출은 했지만 실제로 쓰이지 않는 데이터일 가능성이 있다. 반대로 어떤 프로세스가 어떤 엔터티와도 연결되지 않으면, 처리 대상 데이터가 빠졌을 가능성이 있다.

2. 생성된 데이터의 수명주기 고려 데이터는 생성만 되고 끝나지 않는다. 조회, 변경, 종료 또는 삭제까지 포함한 전체 생명주기 관점에서 검토해야 한다.

3. 과도한 연결은 복잡도의 신호 하나의 데이터가 지나치게 많은 프로세스에서 생성·수정된다면, 이는 결합도가 높아지고 무결성 관리가 어려워지는 신호일 수 있다. 이 경우 프로세스 분리나 데이터 구조 재검토가 필요하다.

Intro

프로젝트를 진행하면서 채팅 기록을 저장하기 위해 mongo DB를 많이 사용했습니다. 문서형 NoSQL의 주축을 담당하고 있는 mongo DB는 제가 사용했던 채팅이나 대용량 로그, 실시간 피드 등 다양한 도메인에서 사용되고 있습니다. 이번 글에서는 Mongo DB 공식 홈페이지 강의를 들으며 공부한 내용을 정리해보겠습니다.

NoSQL

이전에 작성한 글에 더해 좀 더 자세히 알아보겠습니다. Mongo DB는 NoSQL의 한 종류로 문서형 데이터베이스로 일컬어집니다. 관계형 데이터베이스 (RDBMS)의 한계를 극복하기 위해 등장한 데이터베이스로 기존의 정형화된 테이블 구조 대신, 유연한 스키마와 확장성 중심 설계를 통해 대용량 데이터, 비정형 데이터에 유리합니다

• 스키마가 자유로움 : 구조가 고정되어 있지 않아 변경이 쉽다
• 수평적 확장이 쉬움 : 분산 저장이 기본 전제
• 고속 쓰기 처리에 강점
• CAP 이론에 따라 일관성 대신 가용성을 중시하는 경우가 많다

1. RDBMS ➡️ Document Model

RDB는 정규화, MongoDB는 사용 패턴 (Workloads)이 설계의 중심

이번 프로젝트를 하며 느낀점은 Mongo DB는 RDB와는 데이터 모델링 자체에 대한 철학이 다른 것을 인지해야합니다.

워크로드 기반 설계

Mongo DB에서는 데이터를 어떻게 쓰고, 읽고, 업데이트 할 것인지 실제 사용 패턴을 중심으로 데이터 모델을 설계해야합니다.

• 어떤 데이터를 자주 함께 조회하는가?
• 읽기와 쓰기 중 어떤 작업이 더 많은가?
• 특정 문서의 크기가 계속 커질 가능성이 있는지?

이러한 설계의 차이는 RDB는 데이터 정합성과 무결성이 중요하고 Mongo DB는 가장 큰 특징인 데이터 중복 허용과 쿼리 효율성 우선시 하기 때문입니다.

모델링 시 고려사항

Mongo DB에서 문서 설계 시 다음과 같은 기준을 고려합니다.

• 한 번의 요청으로 필요한 데이터를 모두 가져올 수 있는지?
• 문서 크기가 16MB (Mongo DB의 단일 문서 크기)를 초과할 위험이 있는지?

이 기준을 통해 문서 간 관계를 Embedding으로 할 지 Referencing으로 할 지 결정됩니다.

{
  "userId": "abc123",
  "name": "홍길동",
  "posts": [
    {"title": "첫 글", "createdAt": "..."},
    {"title": "두 번째 글", "createdAt": "..."}
  ]
}

MongoDB로 사용자 게시글을 조회한 예시입니다. 자주 같이 조회된다면 예시와 같이 임베딩이 적절하지만 게시글 수가 수천 개 이상으로 커진다면 분리한 후 레퍼런싱 하는 것이 더 적절합니다. 예시를 통해 더 알아봅시다.

• 유저 프로필 + 설정

  {
  "userId": "123",
  "name": "철수",
  "settings": {
    "theme": "dark",
    "notifications": true
  }
  }

  name과 settings는 유저 정보를 볼 때 항상 함께 조회됩니다. 따라서 지금처럼 임베딩이 적절합니다.

• 유저 + 게시글

  {
  "userId": "123",
  "name": "철수"
  }

{
  "postId": "999",
  "userId": "123",
  "title": "MongoDB 잘 쓰는 법",
  "content": "..."
}

사용자에 대한 정보는 자주 변경되지 않습니다. 하지만 게시글은 개수가 많고 조회, 추가, 삭제가 잦습니다 따라서 유저 데이터를 조회할 때 게시글이 항상 가져올 필요가 없기 때문에 레퍼런싱하는 것이 적절합니다.

2. 스키마 설계 패턴과 안티 패턴

좋은 스키마 설계는 데이터를 저장하는 구조를 넘어 읽기, 쓰기 효율성 + 유지보수성을 좌우합니다.

Mongo DB는 자유로운 구조 덕분에 처음엔 빠르게 개발할 수 있지만, 데이터가 많아지면 조회 성능 저하, 문서 크기 한계, 쿼리 복잡도 증가 등의 문제가 발생합니다. 그래서 데이터의 특성에 맞게 구조를 설계하는 패턴이 필요합니다.

주요 스키마 패턴

1️⃣ Extended Reference Pattern

관계형 DB처럼 모든 것을 분리하면 매번 $lookup (JOIN느낌?) 해야 하므로 비효율적입니다. 따라서 자주 쓰는 필드는 함께 저장해 읽기 성능을 높입니다.

{
  "postId": "abc",
  "title": "몽고디비 패턴 정리",
  "author": {
    "userId": "u123",
    "name": "철수"
  }
}

• 게시글 조회 시 name은 자주 보여주기 때문에 포함
• 이메일, 주소 등 자주 사용하지 않는 필드는 참조 ($lookup 사용)

2️⃣ Outlier Pattern

대다수가 작은 용량을 갖지만 소수의 문서가 너무 커서 성능에 문제가 발생할 때 사용하는 패턴입니다. 예외 (Outlier)를 분리해 저장하는 패턴입니다.

[Outlier 적용 ❌]

// 모든 리뷰를 한 문서에 저장한 경우
{
  "productId": "P123",
  "reviews": [
    {"user": "철수", "comment": "좋아요"},
    {"user": "영희", "comment": "별로예요"},
    ...
    // 한 제품에 수천 개의 리뷰
  ]
}

• 대부분 제품의 리뷰수는 소수
• 특정 제품이 유명세를 타서 리뷰수가 1만개 이상이 된다면? 특정 한 문서만 크기가 커짐
• Mongo DB는 단일 문서 크기가 16MB 제한이기 때문에 오류 발생 가능
• 조회/ 수정 시 rewrite cost 증가 : 타 문서의 성능에도 영향

[Outlier 적용 ⭕]

// 메인 리뷰 문서
{
  "productId": "p123",
  "reviews": [
    {"comment": "좋아요!", "user": "철수"},
    {"comment": "추천합니다", "user": "영희"}
  ]
}

// 리뷰가 수천 개인 경우 별도 컬렉션
{
  "productId": "p123",
  "overflowReviews": [
    {"comment": "너무 많음", "user": "길동"}
  ]
}

• 기본 문서는 빠르게 조회 가능해짐
• Outlier 문서가 존재하면 필요할 때만 추가 조회 (2단계 쿼리 필요)

이 패턴은 사용 시 분기 처리 및 N 단계의 쿼리가 필요할 수 있기 때문에 적용에 고려가 필요합니다.

3️⃣ Bucket Pattern 시간 기반 로그 데이터는 쌓이면 매우 많이지기 때문에 Bucket이라는 묶음 개념을 사용합니다. 이를 통해 인덱스 수를 줄일 수 있습니다.

{
  "sensorId": "abc",
  "bucketStart": "2024-01-01T00:00:00Z",
  "readings": [
    {"ts": "01:00", "value": 23},
    {"ts": "01:10", "value": 25},
    ...
  ]
}

• 로그를 하루 혹은 한 시간 등의 단위로 묶어 저장
• 읽을 때는 $unwind 사용

안티 패턴

Mongo DB를 설계 없이 사용하면 발생할 수 있는 패턴입니다.

1️⃣ Unbounded Arrays

말 그대로 끝도 없이 배열이 늘어지는 경우입니다.

{
  "userId": "123",
  "likedPosts": ["a", "b", "c", ..., "z", "aa", "ab", "ac", ...]
}

• 배열이 무제한으로 커지기 때문에 검색 성능 및 문서 크기 문제 발생 야기
• 일정 개수 이상이 되면 분리하거나 Bucket Pattern 활용으로 해결 가능

2️⃣ Growing Document

이 또한 말 그대로 문서가 계속 커지는 경우입니다. 주로 임베딩 구조에서 나타나며 필요 시 설계를 다시한 후 데이터베이스를 수정해야합니다.

• 문서가 16MB 크기 제한을 넘음
• 수정 시 rewrite cost 증가
• 임베딩을 레퍼런싱으로 변경하여 별도 컬렉션으로 저장하여 해결 가능

3. MongoDB 인덱스 설계 전략

MongoDB를 사용하면서 체감했던 점은 스키마 설계만큼이나 인덱스 설계가 성능에 큰 영향을 준다는 것입니다.

앞에서 살펴본 것처럼 MongoDB는 워크로드 기반 설계가 핵심이며, 인덱스 역시 “어떤 쿼리가 가장 많이 실행되는가”를 기준으로 설계해야 합니다.

MongoDB 인덱스의 기본 개념

MongoDB의 인덱스는 RDBMS와 마찬가지로 조회 성능을 향상시키기 위한 자료구조이지만, 문서형 데이터베이스 특성에 맞는 인덱스들이 존재합니다.

• 단일 필드 인덱스
• 복합 인덱스 (Compound Index)
• Multikey Index (배열 필드 인덱싱)
• TTL Index (자동 만료 인덱스)
• Text Index (문자열 검색)

단일 필드 인덱스

하나의 필드에 대해 인덱스를 생성하는 가장 기본적인 형태입니다.

db.users.createIndex({ userId: 1 })

• 특정 ID 기반 조회에 적합
• 단건 조회 성능이 매우 빠름
• 기본적으로 _id 필드는 자동 인덱싱됨

복합 인덱스

MongoDB에서는 여러 조건을 함께 사용하는 쿼리가 매우 빈번하기 때문에, 복합 인덱스 설계가 중요합니다.

db.messages.createIndex({ roomId: 1, createdAt: -1 })

이 인덱스는 다음과 같은 쿼리를 효율적으로 처리합니다.

db.messages
    .find({ roomId: "room123" })
    .sort({ createdAt: -1 })
    .limit(50)

• 특정 채팅방의 메시지 조회
• 최신 메시지부터 N개 조회
• 채팅, 로그, 피드 도메인에서 매우 자주 등장하는 패턴

인덱스 필드 순서의 중요성

복합 인덱스에서는 필드의 순서가 매우 중요합니다

{ roomId: 1, createdAt: -1 }

MongoDB는 인덱스를 왼쪽(prefix)부터 순차적으로 사용합니다 따라서 아래와 같은 쿼리는 인덱스를 제대로 활용하지 못할 수 있습니다

db.messages.find({
  createdAt: { $gte: ISODate("2024-01-01") }
})

👉 인덱스는 반드시 실제 쿼리 패턴의 필터 → 정렬 순서를 기준으로 설계해야 합니다.

Multikey Index (배열 인덱스)

MongoDB는 배열 필드에 대해서도 자동으로 인덱스를 생성할 수 있습니다.

{
  "userId": "123",
  "tags": ["mongodb", "nosql", "backend"]
}

db.posts.createIndex({ tags: 1 })

• 배열 요소 각각이 인덱스로 관리됨
• $in, $all 조건 검색에 유리
• 태그 기반 검색, 관심사 필터링 등에 활용 가능

단, 배열의 크기가 계속 증가하는 경우 인덱스 크기 증가 및 성능 저하로 이어질 수 있으므로
앞에서 다룬 Unbounded Array 안티 패턴과 함께 고려해야 합니다.

TTL Index – 로그와 세션 데이터 관리

TTL(Time To Live) 인덱스는 특정 시간이 지나면 문서를 자동으로 삭제하는 인덱스입니다.

db.sessions.createIndex(
  { createdAt: 1 },
  { expireAfterSeconds: 60 * 60 * 24 }
)

• 세션 데이터
• 임시 인증 정보
• 오래된 로그 데이터

TTL 인덱스를 활용하면 별도의 배치 작업 없이도 데이터 수명 주기가 자동으로 관리됩니다.
로그나 채팅 시스템처럼 데이터가 빠르게 쌓이는 구조에서 유용합니다.

Outro

채팅 기능을 구현하게 된다면 한번쯤 고려해보게 되는 몽고 DB에 대해 간략하게 소개해봤습니다. NoSQL을 다룬다면 반드시 알아야 하는 CAP 이론으로 다시 찾아뵙겠습니다.

n개의 정점을 갖는 이진 트리의 정점에 1부터 n까지의 번호가 중복 없이 매겨져 있다. 이와 같은 이진 트리의 인오더와 포스트오더가 주어졌을 때, 프리오더를 구하는 프로그램을 작성하시오.

입력 첫째 줄에 n(1 ≤ n ≤ 100,000)이 주어진다. 다음 줄에는 인오더를 나타내는 n개의 자연수가 주어지고, 그 다음 줄에는 같은 식으로 포스트오더가 주어진다.

출력 첫째 줄에 프리오더를 출력한다.

예제 입력 3 1 2 3 1 3 2

예제 출력 2 1 3

해설

출처

• 전위 순회 Preorder : Root - Left - Right

• 중위 순회 Inorder : Left - Root - Right

• 후위 순회 Postorder : Left - Right - Root

설계

• 후위 순회와 중위 순회가 주어졌을 때 전위 순회를 구해야 하는 상황

1. 후위 순회의 가장 마지막 원소는 현재 트리의 루트
2. 중위 순회에서는 후위 순위를 찾은 루트를 기준으로 왼쪽 서브트리 / 오른쪽 서브트리로 나뉨
3. 첫번째 루트 ➡️ 가장 큰 분기점으로 왼쪽 서브트리 / 오른쪽 서브트리 나뉨
4. 두번째 루트 ➡️ 두번째로 큰 분기점으로 첫번째 루트로 나뉜 서브트리들에서 1~2번 반복
5. N번째 루트 ➡️ N번째로 큰 분기점으로 이전에 나뉜 서브트리들에서 모두 1~2번 반복 ➡️ 재귀

파이썬 재귀 사용 주의점

import sys
input = sys.stdin.readline

def build(il, ir, pl, pr):
    if il > ir:
        return

    root = postorder[pr]
    ans.append(root)

    idx = inorder.index(root, il, ir + 1)

    ls = idx - il

    build(il, idx - 1, pl, pl + ls - 1)
    build(idx + 1, ir, pl + ls, pr - 1)

N = int(input())
inorder = list(map(int, input().split()))
postorder = list(map(int, input().split()))

ans = []
build(0, N - 1, 0, N - 1)
print(" ".join(map(str, ans)))

• 파이썬은 기본 재귀 횟수가 1000으로 제한
• sys.setrecursionlimit(N) 을 통해 횟수 제한을 늘려줘야함
• 대부분 10^6 정도 사용하는 느낌...

시간 초과 해결

import sys
input = sys.stdin.readline
sys.setrecursionlimit(10**6)
def build(il, ir, pl, pr):
    if il > ir:
        return

    root = postorder[pr]
    ans.append(root)

    idx = inorder.index(root, il, ir + 1)

    ls = idx - il

    build(il, idx - 1, pl, pl + ls - 1)
    build(idx + 1, ir, pl + ls, pr - 1)

N = int(input())
inorder = list(map(int, input().split()))
postorder = list(map(int, input().split()))

ans = []
build(0, N - 1, 0, N - 1)
print(" ".join(map(str, ans)))

• 재귀 해결하고 이번엔 시간 초과 발생
• root를 찾기 위해 사용한 파이썬의 arr.index()가 문제라 생각
• 직접 돌리면 빠르려나???

import sys
input = sys.stdin.readline
sys.setrecursionlimit(10**6)
def build(il, ir, pl, pr):
    if il > ir:
        return

    root = postorder[pr]
    ans.append(root)

    idx = -1
    for k in range(il, ir + 1):
        if inorder[k] == root:
            idx = k
            break
    ls = idx - il

    build(il, idx - 1, pl, pl + ls - 1)
    build(idx + 1, ir, pl + ls, pr - 1)

N = int(input())
inorder = list(map(int, input().split()))
postorder = list(map(int, input().split()))

ans = []
build(0, N - 1, 0, N - 1)
print(" ".join(map(str, ans)))

• 똑 같 다 !
• .indx()를 사용한 이유 : 트리의 순회 문제에서는 값이 모두 다르기 때문에 가장 빠른 배열값의 인덱스를 리턴해도 괜찮다고 생각함

출처

• index()가 선형 탐색으로 O(n)
• O(1)로 해결해야 시간 초과 안난다 생각

정답 코드

import sys
input = sys.stdin.readline
sys.setrecursionlimit(10**6)
def build(il, ir, pl, pr):
    if il > ir:
        return

    root = postorder[pr]
    ans.append(root)

    idx = pos[root]
    ls = idx - il

    build(il, idx - 1, pl, pl + ls - 1)
    build(idx + 1, ir, pl + ls, pr - 1)

N = int(input())
inorder = list(map(int, input().split()))
postorder = list(map(int, input().split()))
pos = {v: i for i, v in enumerate(inorder)}

ans = []
build(0, N - 1, 0, N - 1)
print(" ".join(map(str, ans)))

• pos = {v: i for i, v in enumerate(inorder)}

  • inorder 리스트를 처음부터 끝까지 보면서 각 값 v가 중위 순회에서 몇 번째 인덱스에 있는지를 저장
  • ex) inorder = [4, 2, 5, 1, 6, 3, 7] ➡️pos = {4: 0, 2: 1, 5: 2, 1: 3, 6: 4, 3: 5, 7:6}

• 배열 한번 돌고 인덱스 찾을 때는 O(1)

새로운 프로젝트를 진행하며 배포를 진행하다보니 AWS의 비용 부담이,, 크게 느껴졌습니다. 프리 티어를 사용하면 스프링 + 젠킨스 띄우는데도 압박이 있기에 다른 선택지를 알아봤습니다. 몇 년 전 부터 오라클 클라우드가 비교적 좋은 스펙의 VM을 무료로 제공해준다 들었습니다. 다만, 그동안은 레퍼런스도 많고 가장 대중적인 AWS를 활용해보는 것이 좋다 생각했는데 이번엔 비용 최소화를 통해 실제로 운영이 가능한 인프라가 필요해서 오라클을 선택하게 되었습니다. 오라클 클라우드를 사용해보며 AWS와 어떤 차이가 있는지 어떤 상황에서 선택할 만한지를 정리해보겠습니다.

AWS vs OCI

몇 개월 전에 AWS는 프리 티어 정책을 개편하면서 신규 사용자에게 최대 200USD 상당의 크레딧을 제공하는 방식으로 전환했습니다. 가입 시 기본 100 USD 크레딧이 주어지고 EC2, RDS, Lambda 등 주요 서비스를 사용해보면 추가 크레딧이 주어집니다. 이 정책은 AWS 서비스를 처음 접하는 입장에서 장점이 분명합니다. 단기간 다양한 서비스를 체험해볼 수 있고 비용 부담 없이 실습을 진행할 수 있습니다. 하지만 이번 프로젝트에서는 운영이 핵심이라 단기간에 소모되는 AWS의 무료 플랜은 아쉬웠고 크레딧이 있어도 워낙 ^^ 비싸서 ^^ 합리적인 선택으로 느껴지지 않았습니다.

인프라 시장을 잡아먹고 있는 AWS에 대항하기 위해 다양한 업체가 등장하고 있는데 오라클도 그 중 하나입니다. 후발주자답게 공격적인 프로모션을 다양하게 진행하고 있습니다. AWS 처럼 일정 기간 동안 크레딧을 제공하는 방식이 아닌 기간 제한 없이 무료로 사용할 수 있는 Always Free 정책이 가장 큰 차이점입니다. 이 외에도 VM, 스토리지, DB 포함해서 실제 서비스를 위한 선택폭이 굉장히 넓습니다.

AWS vs OCI 비교

OCI로 배포하기

VM 인스턴스 생성

Oracle Cloud 계정을 생성하면 Always Free 대상인 VM 인스턴스를 생성할 수 있습니다. 그 중 VM.Stantadr.A1.Flex가 현재 오라클이 가장 크게 프로모션 중인 옵션입니다.

AWS 기준으로 아마..? r7g.xlarge 버전이 가장 비슷할텐데 오사카 리전 온디맨드 기준으로 1시간에 $0.2584입니다. 24/7 운영을 하게 되면 한달에 약 $190로 꽤나 괜찮은 성능의 VM을 제공해주는 것을 알 수 있습니다.

현재 일본과 한국은 VM 수가 부족해 무료 계정은 해당 VM 생성이 막혀있습니다. 하지만 유료 계정 전환 후 해당 VM을 생성해도 무료로 이용이 가능합니다. 즉, 비용이 발생하는 서비스만 사용하지 않는다면 유료 계정으로 전환하더라도 서비스를 무료로 즐길 수 있게됩니다.

• ARM 기반 VM (Ampere A1)
  • 최대 4 OCPU
  • 24GB 메모리
  • 인스턴스 1대 or 분할 사용 가능
  • 기간 제한 없음

스토리지 구성

Orcale은 Block Volume을 통해 200GB의 무료 디스크를 제공합니다. AWS의 EBS는 기본 30GB를 12개월간 사용가능한 것과 비교하면 매우 합리적인 선택입니다. 또한 Object Storage에서도 Oracle은 20GB를 제공해주지만 AWS S3는 5GB를 12개월 제공해주고 있습니다. 물론 개인 프로젝트를 진행하며 이 정도 용량이 필요하지 않으실 수도 있지만 모든 하드웨어는 다다익선 거거익선 아니겠습니까. 특히 이번 저희 프로젝트는 "게임"이라 리소스가 생각보다 많아 용량이 크다는 것이 큰 장점으로 이어집니다.

데이터베이스

OCI DB는 MySQL HeatWave의 한정적인 리소스를 활용하면 Always Free 옵션으로 사용이 가능합니다. 대규모 HeatWave 가속 노드 확장이 불가하지만 사이드 프로젝트에서는 충분한 성능을 낼 수 있습니다. AWS RDS가 20GB에 가격이 조금 사악한 것을 고려하면 이 또한 오라클이 무료 버전에서 압도하고 있습니다.

이렇게 Name 옆에 Always Free가 명시된다면 무료로 사용 중이라는 의미입니다.

단점

Always Free가 주는 안정감이 있지만 동아시아 쪽은 VM 인스턴스가 부족해서 유료 계정 전환이 필수입니다. 그래서 Billing Alert/Butget 설정이 필수입니다. 또한 네트워크 설정이 AWS 보다 헷갈리기 쉽습니다. AWS는 보안 그룹 하나로 끝나는 경우가 많은데, OCI는 VCN Security List, 인스턴스 자체 방화벽 등 좀 더 신경쓸 부분이 있었습니다. 가장 큰 단점 중 하나인데, 요즘엔 AI가 있어 크게 체감은 안될수도 있지만 막상 진짜 막혀서 검색을 하면 레퍼런스가 부족합니다. AWS는 구글링하면 비슷한 상황을 겪었고 해결책이 바로 나오는데 반해 OCI는 주로 공식 레퍼런스나 커뮤니티를 돌아가며 찾아봐야하는 것이 크게 체감됐습니다. AWS를 사용해본 경험이 있으시다면 크게 문제가 되진 않겠지만 좀 더 복잡한 환경을 구성하신다면 큰 단점이 될 것 같습니다.

Outro

AWS와 OCI를 비교하며 느낀 점은 두 클라우드의 장단점이 확연하다는 것입니다. AWS는 전통의 강자답게 고급 기능 활용에 좀 더 특화되어 있고 OCI는 무료로 사용하기에 압도적인 퍼포먼스를 내고 있습니다. 물론 이번엔 비용이 최우선이었고 상시 운영을 목표로 하고 있기에 OCI를 선택하게 되었지만 좀 더 복잡한 환경 + 포트폴리오를 위함이라면 AWS가 더 좋은 선택지라 생각합니다. 다음엔 실제 운영 환경에서 발생한 문제점을 정리해보겠습니다,,

바늘 구멍 같은 취업 시장 속 계속 코딩 테스트를 보다보니 구현하는 문제는 자바로 풀이하기 조금 힘들다는 생각이 들고 있습니다! 최근 기업 코딩 테스트 출제 경향이 주로 구현, BFS/DFS, 시뮬레이션이 강세이고 DP나 이분탐색과 같은 것이 곁들여진다 느껴지는데 이런 상황에서 자바보다 파이썬이 훨씬 구현이 빠르고 쉽다는 생각을 지울 수가 없었습니다. 물론 자바만 가능한 곳이 있으니 참고 사항으로 "파이썬을 활용하면 이런 부분이 편하다~" 정도로 생각해주시면 좋을 것 같네요.

자료구조

알고리즘을 풀다보면 주로 사용하는 자료구조로는 배열, 해시맵, 큐/스택, 인접 리스트 등이 있습니다. 자바와 파이썬이 이런 자료구조를 어떻게 다루는지 차이점을 알아보겠습니다.

Java

Queue<Integer> q = new ArrayDeque<>();

Map<Integer, Integer> cnt = new HashMap<>();

Map<Integer, List<Integer>> graph = new HashMap<>();

graph.putIfAbsent(u, new ArrayList<>());
graph.get(u).add(v);

아마 자바로 풀이하시는 분들은 작성에 어려움이 없을 것입니다. 하지만 실제 programmers에서 풀이를 하다보면 작성에 시간이 오래걸리고 그만큼 엣지를 생각할 시간이 줄어들게 됩니다. 특히 각 자료구조의 사용법이 미묘하게 비슷하면서 다르기 때문에 IDE에 익숙하신 분들은 풀이 작성에 더 오랜 시간이 걸리게 됩니다.

Python

from collections import deque, Counter, defaultdict

q = deque()
cnt = Counter(arr)
graph = defaultdict(list)

파이썬의 collections 라이브러리는 구현에 필요한 왠만한 기능이 들어있다 볼 수 있습니다. BFS를 예시로 들면 Queue 생성에 deque()를 호출하면 끝이고 빈도수 계산은 Counter()를 통해 해결할 수 있습니다. 그래프 또한 defaultdict()를 사용하면 쉽게 사용이 가능합니다.

예를 들어 Map의 원소 개수를 세야한다면

 Map<Integer, Integer> cnt = new HashMap<>();
for (int n : numbers) {
    cnt.put(n, cnt.getOrDefault(n, 0) + 1);
}

자바는 getOrDefault()를 통해 개수를 구하게 되고

 cnt = Counter(numbers)

파이썬은 Counter()로 딸깍이 가능합니다.

defaultdict()가 편리한 이유는 없으면 만들어 준다가 기본 동작이기 때문입니다.

 if (!graph.containsKey(u)) {
    graph.put(u, new ArrayList<>());
}
graph.get(u).add(v);

자바에서는 containsKey()를 통해 확인을 거쳐야 하는 반면에

graph = defaultdict(list)
graph[u].append(v)

파이썬은 확인 없이 사용이 가능하다는 장점이 있습니다. 기본적으로 python의 dict는 해시맵 기반으로 자바의 HashMap과 동일선상에 있습니다. defaultdict는 HashMap에 putIfAbsent를 내부적으로 구현해준 것으로 키가 없으면 자동 초기화 기능, 그래프/그룹핑/카운팅 최적 등의 장점이 있습니다.

[Java ↔️ Python 자료구조 대응]

• HashMap ➡️ dict or defaltdict
• TreeMap ➡️ 안씀. 정렬이 필요하면 sort 사용
• PriorityQueue ➡️ heapq
• Graph..? ➡️ dict + list

정렬

알고리즘 풀이하며 정렬하는 경우가 굉장히 많습니다. 물론 자바도 이 정렬이 어렵지 않지만 가끔 직접 Comparator를 구현해야하는 경우가 있습니다.

• 오름차순으로 정렬하되 값이 같으면 다른 기준으로 정렬
• 우선순위가 높은 것부터 처리 (주로 PQ 문제)

Java

import java.util.*;

// 1. 일반 배열 정렬
Arrays.sort(arr);

// 2. 리스트 정렬
Collections.sort(list);

// 3. 커스텀 정렬 (예: 2차원 배열에서 첫 번째 원소 오름차순, 같으면 두 번째 원소 내림차순)
Arrays.sort(arr, (o1, o2) -> {
    if (o1[0] == o2[0]) {
        return o2[1] - o1[1];
    }
    return o1[0] - o2[0];
});

자바 8부터 람다시글 지원해서 간결해졌지만, 여전히 조건이 복잡해지면 Comparator 내부 로직을 작성해야하며 부등호 방향을 헷갈리거나 Integer.compare() 등을 활용해야하는 순간이 많이 있습니다.

Python

# 1. 오름차순
arr.sort()

# 2. 내림차순
arr.sort(reverse=True)

# 3. 커스텀 정렬 (lambda 활용)
arr.sort(key=lambda x: (x[0], -x[1]))

그에 반해 파이썬은 key 파라미터와 lambda의 조합을 활용합니다. 튜플 형태로 우선순위를 지정할 수 있고 -를 붙임으로써 내림차순 정렬이 가능합니다. 자바에서 if-else로 분기를 두며 작성하던 로직이 파이썬에선 단 한 줄로 정리가 됩니다. 시간 단축이 중요한 코딩 테스트에서 간결한 코드 작성이 가능해집니다.

문자열

가장 큰 분기점인 문자열입니다. 이 글을 작성하게 된 계기와 마찬가지인데 구현 문제에서 은근히 괴로운 것이 문자열 파싱과 조작입니다. 자바는 String이 불변(Immutable) 객체라 수정 시 새로운 객체가 생성되어 성능을 위해 StringBuilder를 사용해야 하고 문법도 꽤나 장황한 편입니다.

Java

String s = "Hello World";

// 문자열 뒤집기
StringBuilder sb = new StringBuilder(s);
String reversed = sb.reverse().toString();

// 문자열 자르기
String sub = s.substring(0, 5);

// 문자열 분리 및 결합
String[] parts = s.split(" ");
String joined = String.join(", ", parts);

자바는 문자열을 인덱스로 접근하기 위해 charAt(i)를 사용해야 하고, 부분 문자열을 얻기 위해서는 substring을 사용해야합니다.

Python

s = "Hello World"

# 문자열 뒤집기
reversed_s = s[::-1]

# 문자열 자르기 (슬라이싱)
sub = s[:5]

# 문자열 분리 및 결합
parts = s.split()
joined = ", ".join(parts)

파이썬은 슬라이싱([ : : ]) 을 통해 문자열 처리를 쉽게 다룰 수 있습니다. s[::-1] 하나로 문자열 뒤집기를 할 수 있고 인덱스 접근도 배열처럼 s[i]로 가능해 직관적입니다.

순열과 조합

완전 탐색 혹은 백트래킹 문제를 접하면 순열 (nPr)이나 조합 (nCr)을 구현해야 할 때가 많습니다. 실제로 제가 SSAFY에서 가장 먼저 배운 알고리즘 구현 중 하나가 순조부 였습니다.

Java

자바는 라이브러리 차원에서 순열과 조합을 지원하지 않습니다. 재귀나 스왑, visited (방문 처리), next permutation 등을 활용하여 구현하게 됩니다.

// 조합(Combination) 예시 - 직접 구현 필요
static void combination(int[] arr, boolean[] visited, int start, int n, int r) {
    if (r == 0) {
        // 로직 수행
        return;
    }
    for (int i = start; i < n; i++) {
        visited[i] = true;
        combination(arr, visited, i + 1, n, r - 1);
        visited[i] = false;
    }
}

가장 간단하게 구현하면 위 코드와 같이 구현이 되게 되는데 매번 문제를 풀 때마다 작성해야하고 문제 조건에 맞게 인덱스 조작을 해야하기 때문에 신경을 많이 써야하는 풀이 중 하나라 생각합니다.

Python

가~~끔 쓰면 안되는 기업이 있긴 하지만 Python은 itertools 라는 라이브러리에서 순열과 조합을 지원합니다.

from itertools import permutations, combinations

arr = [1, 2, 3, 4]

# 순열
for p in permutations(arr, 2):
    print(p)

# 조합
for c in combinations(arr, 2):
    print(c)

itertools를 통해 구현 실수가 없고 속도 또한 C로 최적화되어 있어 무지렁이인 제가 작성한 코드보다는 훨씬 빠르게 동작하게 됩니다.

[문제] 2021 kakao 순위 검색

순위 검색 문제는 효율성 테스트가 포함되어 있어 단순히 find나 contains로 접근하면 시간 초과가 나는 문제입니다. 이 문제를 통해 왜 파이썬을 고려하게 되었는지 설명해보겠습니다.

Java

자바에서는 지원자의 정보 (언어, 직군, 경력, 소울푸드)에 대해 - 조건을 포함한 모든 조합을 만들기 위해 재귀 함수(DFS)나 4중 for문을 활용해 구현해야 합니다.

[재귀 함수]

static void makeSentence(String[] p, String str, int cnt) {
    if (cnt == 4) {
        if (!map.containsKey(str)) {
            List<Integer> list = new ArrayList<Integer>();
            map.put(str, list);
        }
        map.get(str).add(Integer.parseInt(p[4]));
        return;
    }
    // 해당 조건을 포함하는 경우
    makeSentence(p, str + p[cnt], cnt + 1);
    // 해당 조건을 '-'로 처리하는 경우
    makeSentence(p, str + "-", cnt + 1);
}

[4중 for문]

for (String la : langs) {
    for (String jo : jobs) {
        for (String ca : careers) {
            for (String fo : foods) {
                String key = la + jo + ca + fo;
                map.computeIfAbsent(key, k -> new ArrayList<>()).add(score);
            }
        }
    }
}

또한 저장된 점수 리스트에서 특정 점수 이상인 사람의 수를 세기 위해 Collections.binarySearch를 사용하거나 직접 lowerBound를 구현해야 했스빈다. 자바의 binarySearch는 값이 없으면 음수를 반환하기 때문에 'X점 이상인 사람의 수'를 구하려면 인덱스 보정 로직을 추가로 작성해야 합니다.

[lowerBound 구현]

private int bs(List<Integer> list, int target) {
    int l = 0;
    int r = list.size();

    while (l < r) {
        int mid = (l + r) / 2;
        if (list.get(mid) < target) {
            l = mid + 1;
        } else {
             r = mid;
        }
    }
    return l;
}

그래서 전체 코드 작성을 보면 심상치 않습니다. 물론 제가 하드 코딩해서 더 더러운 것도 있습니다.

import java.util.*;

class Solution {

    static Map<String, List<Integer>> map = new HashMap<>();

    public int[] solution(String[] info, String[] query) {
        for (String s : info) {
            String[] ss = s.split(" ");
            String lang = ss[0];
            String job = ss[1];
            String career = ss[2];
            String food = ss[3];
            int score = Integer.parseInt(ss[4]);
            String[] langs = {lang, "-"};
            String[] jobs = {job, "-"};
            String[] careers = {career, "-"};
            String[] foods = {food, "-"};
            for (String la : langs) {
                for (String jo : jobs) {
                    for (String ca : careers) {
                        for (String fo : foods) {
                            String key = la + jo + ca + fo;
                            map.computeIfAbsent(key, k -> new ArrayList<>()).add(score);
                        }
                    }
                }
            }
        }

        for (List<Integer> list : map.values()) Collections.sort(list);

        int[] answer = new int[query.length];

        for (int i = 0; i < query.length; i++) {
            String q = query[i];
            String replaced = q.replace(" and ", " ");
            String[] arr = replaced.split(" ");
            String key = arr[0] + arr[1] + arr[2] + arr[3];
            int score = Integer.parseInt(arr[4]);
            List<Integer> list = map.getOrDefault(key, Collections.emptyList());
            int idx = bs(list, score);
            answer[i] = list.size() - idx;
        }
        return answer;
    }

    private int bs(List<Integer> list, int target) {
        int l = 0;
        int r = list.size();

        while (l < r) {
            int mid = (l + r) / 2;
            if (list.get(mid) < target) {
                l = mid + 1;
            } else {
                r = mid;
            }
        }
        return l;
    }
}

Python

파이썬은 이 문제에서 압도적인 퍼포먼스를 보여줍니다.

1. itertools.combinations : 재귀 함수 없이 모든 조합 생성
2. bisect.bisect_left : 직접 구현 없이 이분 탐색 수행

from itertools import combinations
from collections import defaultdict
from bisect import bisect_left

def solution(info, query):
    answer = []
    info_dict = defaultdict(list)

    # 1. 모든 경우의 수 생성
    for i in info:
        temp = i.split()
        conditions = temp[:-1]
        score = int(temp[-1])

        # 4개의 조건 중 0~4개를 선택하여 '-'가 들어갈 조합 생성 (Java의 4중 for문을 combinations로 대체)
        for k in range(5):
            for comb in combinations(range(4), k):
                key_temp = conditions[:]
                for idx in comb:
                    key_temp[idx] = "-" # 선택된 인덱스를 '-'로 변경
                key = "".join(key_temp)
                info_dict[key].append(score)

    # 2. 이분 탐색을 위한 정렬
    for key in info_dict:
        info_dict[key].sort()

    # 3. 쿼리 처리
    for q in query:
        q = q.replace("and ", "").split()
        target_key = "".join(q[:-1])
        target_score = int(q[-1])

        # Lower Bound를 통해 개수 구하기
        count = len(info_dict[target_key]) - bisect_left(info_dict[target_key], target_score)
        answer.append(count)

    return answer

어 아름답다. 아름다와. 구현 코드 거의 없이 라이브러리로 딸깍 해버릴 수 있습니다.

Outro

지금까지 자바와 파이썬의 알고리즘 풀이 스타일을 비교해 보았습니다. 물론 자바에 대한 분노로 작성한 글이라 파이썬 편향적인 글이지만 파이썬이 실제로 더 간결하게 작성이 가능하다는 점에서 고려해볼만 하다 생각합니다. 자바 원툴로 살기 이전엔 파이썬을 좋아했던 사람으로써 자바 공화국인 대한민국에서 코딩 테스트에 자바만 지원한다 => 자바 파이썬과 자바를 동시 지원한다 => 파이썬 이런 식으로 활용하려 합니다. 파이썬은 짱이니까.

Intro

프로젝트를 하다보면 데이터베이스 성능 개선을 위해 최적화를 시도할 때가 종종 있습니다. 인덱싱을 통해 성능 개선을 이루는 경우가 많은데 드라마틱한 변화가 없을 때도 많습니다. 삽입, 삭제, 수정 등 연산이 잦을 경우 오히려 쿼리 성능이 더 나빠지는 경우도 있는데 이런 경우 어떻게 성능을 개선할 수 있을까요? Oracle을 사용하면 좋겠지만 돈이 없는... 이유로 MySQL로 실행 계획을 통한 성능 최적화 방법에 대해 알아보겠습니다.

SQL이 실행되는 과정

MySQL에서 SQL 성능을 좌우하는 것은 옵티마이저입니다. 실행 계획은 옵티마이저가 여러 실행 전략 중 하나를 선택한 최종 판단 결과라고 볼 수 있습니다. 만약 SQL이 MySQL에 전달되었을 때의 흐름을 간단히 보면 다음과 같습니다.

• Client/Server Protocol

  • 클라이언트 (어플리케이션, CLI, JDBC 등)이 MySQL 서버에 접속해서 SQL 전송
  • 네트워크 + 인증 + 세션 영역 -> 전처리기에서 한번 더 이 SQL이 해당 객체를 건드려도 되는지 확인
  • 커넥션을 매번 만드는 것이 비싸기 때문이 커넥션 풀을 사용

• Query Cache

  • 같은 SQL 문자열에 대해 과거 계산 결과를 저장한 후 같은 SQL 요청 시 바로 반환
  • 테이블 변경 시 문제 발생 ➡️ MySQL 5.7 까지는 기본 = off, 8.0부터는 제거..

• Parser

  • 테이블, 컬럼, 함수 이름 등을 토큰 (MySQL이 인식하는 최소 단위)으로 분해
  • 분해 과정에서 SQL 문법이 올바른지 검사
  • Parse Tree 생성

• Preprocessor

  • 최적화가 가능하도록 정보 취합
  • 객체 해석 (컬럼, 테이블의 실존), Alias, 권한 체크, 타입/상수 처리
  • 주로 권한 체크, 컬럼 유무 체크

• Optimizer

  • Parse tree를 보고 어떤 방식으로 실행할지 결정 후 실행 계획 생성
    • 어떤 인덱스를 사용할지
    • 조인 순서, 방식 (Nested Loop Join 기반으로 어떤 테이블을 먼저, 다음을 어떻게 처리할지)
    • WHERE 조건을 언제, 어떻게 적용할지 (조건 푸시다운, 범위 스캔 등)
    • GROUP BY, ORDER BY를 인덱스로 해결할지, 정렬/임시테이블 쓸지
    • LIMIT 처리 방식
  • 참고하는 것 = 테이블 통계 (카디널리티, 인덱스 분포 등), 비용 모델
  • EXPLAIN 사용 시 보이는 내용이 주로 이 결과물 (실행 계획)

• Query Execution Engine

  • 실행 계획을 실제로 수행하는 단계

• Storage Engines (InnoDB, MyISAM, etc...)

  • MySQL = 서버 계층 (파서, 옵티마이저, 실행 엔진) + 스토리지 엔진 계층 (실제 데이터 읽기/쓰기)
  • 실제 물리적으로 디스크/버퍼풀에서 데이터를 읽고, 잠금/트랜잭션 처리

예시

SELECT u.id, u.name
FROM users u
WHERE u.email = 'cupwan@velog.io';

• parser : 문법 체크 및 parse tree 생성
• preprocessor : users, email 존재 확인, 권한 및 타입 확인
• optimizer : email 인덱스가 있으면 인덱스 레인지 or 포인트 룩업 선택
• execution : 인덱스로 row를 찾고 결과 컬럼만 반환
• storage engine : 인덱스 페이지/데이터 페이지를 버퍼풀에서 찾거나 디스크에서 찾아옴

이렇듯 MySQL의 쿼리 처리 과정을 살펴보면 쿼리 성능 개선은 Optimizer의 실행 계획을 먼저 분석해야함을 알 수 있습니다. 인덱스 선택 방식부터 조인 순서, 접근 방식, 추가 연산 여부까지 모두 실행 계획을 통해 알 수 있습니다.

EXPLAIN 기본

이제 MySQL에서 공식으로 제공하는 Employees 데이터베이스를 활용해서 성능 최적화를 위한 EXPLAIN에 대해 알아보겠습니다.

Employees는 test_db 깃허브에서 다운로드 할 수 있습니다.

Schema Inspection을 통해 Employees의 데이터 크기가 꽤나 큰 것을 확인할 수 있습니다. 데이터가 잘 불러오는지 + EXPLAIN을 위한 예시로 가장 큰 용량을 가진 salaries를 전부 조회해보겠습니다.

EXPLAIN SELECT * FROM salaries;

EXPLAIN는 여러 컬럼으로 구성되어 있습니다.

• select_type : SELECT 성격

  • simple, primary, subquery, derived, union 등
  • 서브쿼리나 UNION 등 사용하면 select의 성격 명시

• partitions : 파티션 사용 정보

  • 파티션 테이블일 때, 어떤 파티션만 읽었는지

• type : 테이블 접근 방식

  • 해당 테이블을 어떤 방식으로 접근하는지
  • 성능 분석에 가장 중요
  • 일반적으로 ALL < index < range < ref< eq_ref < const 순으로 좋아짐
  • 현재 type = ALL로 Full Table Scan을 하고 있다는 의미입니다. 당연히 구리다는 뜻입니다.

• key : 실제 사용된 인덱스

  • Key 컬럼은 옵티마이저가 실제로 선택한 인덱스를 의미
  • 현재 NULL값으로 인덱스를 사용하지 않습니다. 인덱스 사용 시 해당 인덱스의 이름이 표시됩니다.
  • possible_keys 는 인덱스 후보, ref 는 인덱스 비교 대상
  • ➕ 인덱스가 있음에도 key가 NULL 이라면 옵티마이저는 비용 계산 결과 인덱스를 사용하는 것이 더 비효율적이라고 판단한 것입니다.

• rows : 예상 조회 건수

  • 실제 결과 건수가 아닌 옵티마이저의 추정치 입니다.
  • 불필요하게 많은 데이터 조회 시, 조인 순서의 효율성, 인덱스 범위 등을 판단하는 용으로 사용됩니다.
  • 주로 성능 튜닝의 목표는 rows의 값을 줄이는 방향으로 진행됩니다.

• Extra : 추가 연산 여부

  • Using where, Using temporary, Using filesort 등이 값으로 나옵니다.
  • 정렬, 그룹 연산으로 인해 추가 비용이 발생하고 있음을 알려줍니다.

성능 최적화

Employees는 이렇게 이루어져 있습니다. 이제 일부러 거지같은 조건을 통해 옵티마이저가 이상한 선택을 하도록 쿼리를 만들어서 성능을 개선해보겠습니다.

-- 실행할 쿼리, 약 6초 소요
SELECT
  d.dept_name,
  COUNT(*) AS headcount,
  AVG(s.salary) AS avg_salary
FROM departments d
JOIN dept_emp de
  ON de.dept_no = d.dept_no
JOIN salaries s
  ON s.emp_no = de.emp_no
WHERE de.to_date = '9999-01-01'
  AND s.to_date  = '9999-01-01'
GROUP BY d.dept_name
ORDER BY avg_salary DESC;

[EXPLAIN]

[EXPLAIN ANALYZE (MySQL 8.0+)]

'-> Sort: avg_salary DESC  (actual time=4673..4673 rows=9 loops=1)
-> Stream results  (cost=91044 rows=9) (actual time=340..4673 rows=9 loops=1)
-> Group aggregate: count(0), avg(s.salary)  (cost=91044 rows=9) (actual time=340..4673 rows=9 loops=1)
-> Nested loop inner join  (cost=87460 rows=35834) (actual time=14.3..4465 rows=240124 loops=1)
-> Nested loop inner join  (cost=42230 rows=37254) (actual time=14.2..1367 rows=240124 loops=1)
-> Covering index scan on d using dept_name  (cost=1.9 rows=9) (actual time=0.239..0.266 rows=9 loops=1)
-> Filter: (de.to_date = DATE\'9999-01-01\')  (cost=599 rows=4139) (actual time=13.9..149 rows=26680 loops=9)
-> Index lookup on de using dept_no (dept_no=d.dept_no)  (cost=599 rows=41393) (actual time=13.9..139 rows=36845 loops=9)
-> Filter: (s.to_date = DATE\'9999-01-01\')  (cost=0.252 rows=0.962) (actual time=0.0111..0.0125 rows=1 loops=240124)
-> Index lookup on s using PRIMARY (emp_no=de.emp_no)  (cost=0.252 rows=9.62) (actual time=0.00676..0.0108 rows=10.5 loops=240124)'

• Nested loop inner join ... rows=240124 가 문제입니다.
• salaries를 24만 번 조회하는 부분이 전체 시간의 대부분을 잡아먹고 있습니다.
• 순서
  • department d : 부서 9개 읽기
  • dept_emp de : 부서별 현재 재직자 뽑기
    • dept_no 인덱스로 부서 전체 이력을 가져오고
    • 그 중 to_date='9999-01-01'만 남김
  • salaries s : 직원별 급여 이력에서 현재 급여 1개 찾기
    • 직원 1명당 급여 이력 평균 10.5건 읽고
    • 그 중 현재 1건만 남김
    • 이 과정을 24만번 반복 중;;

그럼 이제 salaries를 효율적으로 개선해보겠습니다.

• PK로만 붙어서 직원 이력 전체 조회
• 우리가 원하는 것 = to_date='9999-01-01' 단 하나의 행

인덱스를 추가하기 위해 salaries의 인덱스 상황을 확인합니다. ** 혹시라도 따라 하지 마세요!!! 이 인덱스 설정은 잘못됐습니다!! **

ALTER TABLE salaries ADD INDEX idx_salaries_to_date_emp (to_date, emp_no);

'-> Sort: avg_salary DESC  (actual time=10246..10246 rows=9 loops=1)\n    -> Stream results  (cost=77534 rows=9) (actual time=686..10246 rows=9 loops=1)\n        -> Group aggregate: count(0), avg(s.salary)  (cost=77534 rows=9) (actual time=686..10246 rows=9 loops=1)\n            -> Nested loop inner join  (cost=73808 rows=37254) (actual time=9.54..10110 rows=240124 loops=1)\n                -> Nested loop inner join  (cost=48661 rows=37254) (actual time=9.5..1657 rows=240124 loops=1)\n                    -> Covering index scan on d using dept_name  (cost=1.9 rows=9) (actual time=1.09..2.15 rows=9 loops=1)\n                    -> Filter: (de.to_date = DATE\'9999-01-01\')  (cost=1313 rows=4139) (actual time=7.52..182 rows=26680 loops=9)\n                        -> Index lookup on de using dept_no (dept_no=d.dept_no)  (cost=1313 rows=41393) (actual time=7.51..175 rows=36845 loops=9)\n                -> Index lookup on s using idx_salaries_to_date_emp (to_date=DATE\'9999-01-01\', emp_no=de.emp_no)  (cost=0.575 rows=1) (actual time=0.0344..0.0349 rows=1 loops=240124)\n'

어이쒸 왜 10초로 늘었지;; ➡️ 이 과정을 반복하는 것이 최적화의 길이겠지요... 그래도 분석 결과를 보니 Secondary Index로 해결이 안되고 (당연함. PK로 한번 더 들어감.) Join 구조를 변경해야할 것 같습니다. 즉, 인덱스를 추가해서 해결될 것 같았지만? 조인 구조가 Nested Loop이고 반복 횟수가 매우 커서 1회 탐색 비용이 더 큰 인덱스를 타게 되기 때문에 오히려 늦어짐을 알 수 있었습니다.

제.. 생각엔 현재 급여만 먼저 한 번 뽑아서 조인해보는 방법을 시도해봤습니다.

EXPLAIN ANALYZE
SELECT
  d.dept_name,
  COUNT(*) AS headcount,
  AVG(cs.salary) AS avg_salary
FROM departments d
JOIN dept_emp de
  ON de.dept_no = d.dept_no
 AND de.to_date = '9999-01-01'
JOIN (
  SELECT emp_no, salary
  FROM salaries
  WHERE to_date = '9999-01-01'
) cs
  ON cs.emp_no = de.emp_no
GROUP BY d.dept_name
ORDER BY avg_salary DESC;

'-> Sort: avg_salary DESC  (actual time=9519..9519 rows=9 loops=1)
-> Stream results  (cost=68140 rows=9) (actual time=925..9519 rows=9 loops=1)
-> Group aggregate: count(0), avg(salaries.salary)  (cost=68140 rows=9) (actual time=925..9519 rows=9 loops=1)
-> Nested loop inner join  (cost=64415 rows=37254) (actual time=6.91..9352 rows=240124 loops=1)
-> Nested loop inner join  (cost=44986 rows=37254) (actual time=6.88..1928 rows=240124 loops=1)
-> Covering index scan on d using dept_name  (cost=1.9 rows=9) (actual time=0.026..0.46 rows=9 loops=1)
-> Filter: (de.to_date = DATE\'9999-01-01\')  (cost=905 rows=4139) (actual time=7.35..212 rows=26680 loops=9)
-> Index lookup on de using dept_no (dept_no=d.dept_no)  (cost=905 rows=41393) (actual time=7.35..204 rows=36845 loops=9)
-> Index lookup on salaries using idx_salaries_to_date_emp (to_date=DATE\'9999-01-01\', emp_no=de.emp_no)  (cost=0.422 rows=1) (actual time=0.03..0.0306 rows=1 loops=240124)'

더 느려졌습니다! 10초로 늘었는데 이유를 보아하니 옵티마이저가 파생 테이블을 그냥 원래 쿼리로 병합해서 비슷한 흐름으로 실행을 하게 되네요. 그래서 24만번 lookup을 다시..하고 있습니다. 진짜 모르겠다. 이거 어떻게 최적화 하냐... 해서 GPT한테 물어봤습니다. 죄송합니다.

기존 인덱스를 삭제 후 새로운 인덱스를 생성합니다. 9999-01-01인 데이터들이 디스크 상 한 블록에 모일 수 있도록; 살짝 강제로 최적화를 진행합니다.

ALTER TABLE salaries ADD INDEX idx_salaries_final (to_date, emp_no, salary);

그 후 쿼리도 CTE를 활용해서 현재 연봉 테이블과 현재 부서 테이블을 미리 전처리한 후 합치는 구조로 변경합니다.

WITH current_salaries AS (
    SELECT emp_no, salary 
    FROM salaries 
    WHERE to_date = '9999-01-01'
),
current_dept_emp AS (
    SELECT emp_no, dept_no
    FROM dept_emp
    WHERE to_date = '9999-01-01'
)
SELECT 
    d.dept_name,
    COUNT(*) AS headcount,
    AVG(cs.salary) AS avg_salary
FROM current_dept_emp de
JOIN current_salaries cs 
    ON cs.emp_no = de.emp_no
JOIN departments d 
    ON d.dept_no = de.dept_no
GROUP BY d.dept_name
ORDER BY avg_salary DESC;

드디어 기존 6초 -> 1.3초로 최적화가 이루어졌습니다. 제가 한 건 아니지만ㅜ

-> Sort: avg_salary DESC  (actual time=1362..1362 rows=9 loops=1)
-> Stream results  (cost=86935 rows=9) (actual time=149..1362 rows=9 loops=1)
-> Group aggregate: count(0), avg(salaries.salary)  (cost=86935 rows=9) (actual time=149..1362 rows=9 loops=1)
-> Nested loop inner join  (cost=83209 rows=37254) (actual time=14.4..1278 rows=240124 loops=1)
-> Nested loop inner join  (cost=42230 rows=37254) (actual time=14.4..674 rows=240124 loops=1)\
-> Covering index scan on d using dept_name  (cost=1.9 rows=9) (actual time=0.0437..0.0605 rows=9 loops=1)
-> Filter: (dept_emp.to_date = DATE\'9999-01-01\')  (cost=599 rows=4139) (actual time=9.24..73.3 rows=26680 loops=9)
-> Index lookup on dept_emp using dept_no (dept_no=d.dept_no)  (cost=599 rows=41393) (actual time=9.23..68.8 rows=36845 loops=9)
-> Covering index lookup on salaries using idx_salaries_final (to_date=DATE\'9999-01-01\', emp_no=dept_emp.emp_no)  (cost=1 rows=1) (actual time=0.00182..0.00233 rows=1 loops=240124)\n'

• salaries 조회는 그대로 24만번;;
• covering index를 사용해 필요한 컬럼이 다 들어있으면 테이블로 다시 안가고 인덱스에서 바로 값을 꺼내옴
• 반복 시간이 이전에 비해 감소 (0.03ms -> 0.0017ms 수준)

Outro

사실 Oracle 에서 /* index hint */를 통해 최적화 하는 방법을 보고 꽂혀서 MySQL에서도 인덱싱 강제해서 최적화 하는 그런 쿼리를 가져오고 싶었으나 실패했습니다. 그리고 Explain을 통해 쿼리 성능 최적화에 대해 알아보고 싶었으나 어찌 마지막은 생성형 AI가 최적화도 잘하는 것을 알게 되었네요ㅠ 하지만! 실무에서는 복잡한 상황을 인간만이 적용할 수 있기에 이렇게 간단한 최적화가 아니라면 Explain, Explain Analyze 등을 사용해 볼 기회가 있을 것입니다. 다음엔 더 알차게 돌아오겠읍니다. 감사합니다.

문제

업무용 소프트웨어를 개발하는 니니즈웍스의 인턴인 앙몬드는 명령어 기반으로 표의 행을 선택, 삭제, 복구하는 프로그램을 작성하는 과제를 맡았습니다. 세부 요구 사항은 다음과 같습니다

위 그림에서 파란색으로 칠해진 칸은 현재 선택된 행을 나타냅니다. 단, 한 번에 한 행만 선택할 수 있으며, 표의 범위(0행 ~ 마지막 행)를 벗어날 수 없습니다. 이때, 다음과 같은 명령어를 이용하여 표를 편집합니다.

"U X": 현재 선택된 행에서 X칸 위에 있는 행을 선택합니다. "D X": 현재 선택된 행에서 X칸 아래에 있는 행을 선택합니다. "C" : 현재 선택된 행을 삭제한 후, 바로 아래 행을 선택합니다. 단, 삭제된 행이 가장 마지막 행인 경우 바로 윗 행을 선택합니다. "Z" : 가장 최근에 삭제된 행을 원래대로 복구합니다. 단, 현재 선택된 행은 바뀌지 않습니다. 예를 들어 위 표에서 "D 2"를 수행할 경우 아래 그림의 왼쪽처럼 4행이 선택되며, "C"를 수행하면 선택된 행을 삭제하고, 바로 아래 행이었던 "네오"가 적힌 행을 선택합니다(4행이 삭제되면서 아래 있던 행들이 하나씩 밀려 올라오고, 수정된 표에서 다시 4행을 선택하는 것과 동일합니다).

다음으로 "U 3"을 수행한 다음 "C"를 수행한 후의 표 상태는 아래 그림과 같습니다.

다음으로 "D 4"를 수행한 다음 "C"를 수행한 후의 표 상태는 아래 그림과 같습니다. 5행이 표의 마지막 행 이므로, 이 경우 바로 윗 행을 선택하는 점에 주의합니다.

다음으로 "U 2"를 수행하면 현재 선택된 행은 2행이 됩니다.

위 상태에서 "Z"를 수행할 경우 가장 최근에 제거된 "라이언"이 적힌 행이 원래대로 복구됩니다.

다시한번 "Z"를 수행하면 그 다음으로 최근에 제거된 "콘"이 적힌 행이 원래대로 복구됩니다. 이때, 현재 선택된 행은 바뀌지 않는 점에 주의하세요.

이때, 최종 표의 상태와 처음 주어진 표의 상태를 비교하여 삭제되지 않은 행은 "O", 삭제된 행은 "X"로 표시하면 다음과 같습니다.

처음 표의 행 개수를 나타내는 정수 n, 처음에 선택된 행의 위치를 나타내는 정수 k, 수행한 명령어들이 담긴 문자열 배열 cmd가 매개변수로 주어질 때, 모든 명령어를 수행한 후 표의 상태와 처음 주어진 표의 상태를 비교하여 삭제되지 않은 행은 O, 삭제된 행은 X로 표시하여 문자열 형태로 return 하도록 solution 함수를 완성해주세요.

제한사항

• 5 ≤ n ≤ 1,000,000
• 0 ≤ k < n
• 1 ≤ cmd의 원소 개수 ≤ 200,000
  • cmd의 각 원소는 "U X", "D X", "C", "Z" 중 하나입니다.
  • X는 1 이상 300,000 이하인 자연수이며 0으로 시작하지 않습니다.
  • X가 나타내는 자연수에 ',' 는 주어지지 않습니다. 예를 들어 123,456의 경우 123456으로 주어집니다.
  • cmd에 등장하는 모든 X들의 값을 합친 결과가 1,000,000 이하인 경우만 입력으로 주어집니다.
  • 표의 모든 행을 제거하여, 행이 하나도 남지 않는 경우는 입력으로 주어지지 않습니다.
  • 본문에서 각 행이 제거되고 복구되는 과정을 보다 자연스럽게 보이기 위해 "이름" 열을 사용하였으나, "이름"열의 내용이 실제 문제를 푸는 과정에 필요하지는 않습니다. "이름"열에는 서로 다른 이름들이 중복없이 채워져 있다고 가정하고 문제를 해결해 주세요.
• 표의 범위를 벗어나는 이동은 입력으로 주어지지 않습니다.
• 원래대로 복구할 행이 없을 때(즉, 삭제된 행이 없을 때) "Z"가 명령어로 주어지는 경우는 없습니다.
• 정답은 표의 0행부터 n - 1행까지에 해당되는 O, X를 순서대로 이어붙인 문자열 형태로 return 해주세요.

정확성 테스트 케이스 제한 사항

• 5 ≤ n ≤ 1,000
• 1 ≤ cmd의 원소 개수 ≤ 1,000

효율성 테스트 케이스 제한 사항

• 주어진 조건 외 추가 제한사항 없습니다.

아이디어 + 풀이

1. 스택

그림이 너무 스택을 쓰고싶게 나와서 스택을 사용할까 했습니다. 구현이 굉장히 쉽고 n과 cmd가 1000 이하이기 때문에 시간복잡도도 O(n*cmd) 라 괜찮습니다. 라고 생각했습니다.

import java.util.*;

class Solution {
    public String solution(int n, int k, String[] cmd) {
        boolean[] alive = new boolean[n];
        Arrays.fill(alive, true);
        Deque<Integer> stack = new ArrayDeque<>();
        int cur = k;

        for (String s : cmd) {
            char op = s.charAt(0);
            if (op == 'U' || op == 'D') {
                int x = Integer.parseInt(s.substring(2));
                if (op == 'U') {
                    while (x-- > 0) {
                        int p = cur - 1;
                        while (p >= 0 && !alive[p]) p--;
                        if (p >= 0) cur = p;
                    }
                } else {
                    while (x-- > 0) {
                        int q = cur + 1;
                        while (q < n && !alive[q]) q++;
                        if (q < n) cur = q;
                    }
                }
            } else if (op == 'C') {
                stack.push(cur);
                alive[cur] = false;

                int q = cur + 1;
                while (q < n && !alive[q]) q++;
                if (q < n) {
                    cur = q;
                } else {
                    int p = cur - 1;
                    while (p >= 0 && !alive[p]) p--;
                    cur = p;
                }
            } else { // 'Z'
                int rec = stack.pop();
                alive[rec] = true;
            }
        }

        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < n; i++) sb.append(alive[i] ? 'O' : 'X');
        return sb.toString();
    }
}

• JDK 공식 문서에서 LIFO 자료구조로 Stack 클래스 보다 Deque 사용 권장
• ArrayDeque가 Stack보다 빠를 가능성이 있다.

효율성 테스트에서 박살이 났는데 왜 그런걸까요?

• U x / D x를 처리할 때마다 alive[p]가 false인 구간을 while로 뛰어넘는 선형 스캔 ➡️ 최악 O(n*이동횟수)
• 이동횟수 = 죽은 구간 길이 X 그 구간을 건너뛴 횟수 : 각 이동마다 죽은 칸을 건너뛰는 비용이 추가됨

2. 펜윅 트리 (BIT)

최근..?에 작성한 펜윅 트리가 생각나서 펜윅트리로 풀어봤습니다. 핵심 아이디어는 다음과 같습니다.

• 살아있는 행을 0/1로 표현
  • alive[i] = 1(살) / 0(죽)을 펜윅 트리에 저장
• sum(i) = 1...i 구간에 살아있는 행의 개수 = i번째 인덱스의 순위
  • 커서가 보고 있는 인덱스 cur의 현재 순위는 r = sum(cur)
• D x -> 순위를 r+x로, U x -> 순위를 r-x로
• 삭제/복구는 해당 위치의 값을 +1 or -1 업데이트로 반영
• kth(k)에서 k가 1~tot 범위로 들어오는지 확인 (삭제 이후 tot=0이 되지 않는다는 보장)

import java.util.*;

class Solution {
    static class BIT {

        int n; 
        int[] bit;

        BIT(int n){
            this.n=n;
            bit = new int[n+1];
        }

        void add(int i,int d){
            for(; i<=n; i+=i&-i) bit[i]+=d;
        }

        int sum(int i){
            int s=0;
            for(; i>0; i-=i&-i) s+=bit[i];
            return s;
        }

        int kth(int k){
            int idx=0,mask=1;
            while((mask<<1)<=n) mask<<=1;
            for(int d=mask; d!=0; d>>=1){
                int nxt=idx+d;
                if(nxt<=n && bit[nxt]<k){
                    k-=bit[nxt];
                    idx=nxt;
                }
            }
            return idx+1;
        }
    }

    public String solution(int n, int k, String[] cmd) {
        BIT ft = new BIT(n);
        boolean[] alive = new boolean[n];
        Arrays.fill(alive,true);
        for(int i = 1; i <= n; i++) ft.add(i,1);
        Deque<Integer> st=new ArrayDeque<>();
        int cur = k+1;

        for(String s:cmd){
            char op=s.charAt(0);
            if(op == 'U' || op == 'D'){
                int x = Integer.parseInt(s.substring(2));
                int r = ft.sum(cur) + (op=='D'?x:-x);
                cur = ft.kth(r);
            }else if(op == 'C'){
                int r = ft.sum(cur);
                st.push(cur);
                ft.add(cur,-1);
                alive[cur-1]=false;
                int tot = ft.sum(n);
                cur=(r<=tot)?ft.kth(r):ft.kth(tot);
            }else{
                int rec = st.pop();
                ft.add(rec,1);
                alive[rec-1]=true;
            }
        }

        StringBuilder sb=new StringBuilder();
        for(int i=0;i<n;i++) sb.append(alive[i]?'O':'X');
        return sb.toString();
    }
}

• 장점
  • 멋있다! 펜윅 트리 실전 사용 성공이다!
  • 살아있는 개수(순위) 기준이라 스택의 문제점이었던 k번째 살아있는 원소의 실제 위치를 kth(k)로 한번에 접근 가능해졌다!
• 단점
  • 구현이 어렵다
  • 이중연결리스트보다 느림
    • U x, D x에서 x 값이 크면 BIT가 유리 (연결 리스트 : O(x), 펜윅트리 : O(log n)

3. 연결 리스트

2번 방법이 멋있지만 구현이 솔직히 너무 어렵고 다른 문제에서 구현하라면 자신이 없습니다. 결국 중간 다리 건너뛰는 것이 중요한 문제라 생각해서 연결 리스트가 생각났습니다.

• prev[i], next[i] : i의 이웃 포인터, 삭제 시 이웃끼리 포인터 연결
• stack(ArrayDeque) : {row, prev, next} push ➡️ 복구 시 포인터 원상 복구
• 이동 : 포인터만 x번 이동 (O(x), 죽은 구간 스캔 사라짐)

import java.util.*;

class Solution {
    public String solution(int n, int k, String[] cmd) {
        int[] prev = new int[n];
        int[] next = new int[n];
        boolean[] alive = new boolean[n];
        Arrays.fill(alive, true);

        // prev, next 값이 -1인 인덱스가 양 끝
        for (int i = 0; i < n; i++) {
            prev[i] = i - 1;
            next[i] = i + 1;
        }
        next[n - 1] = -1;

        // 삭제된 거 복구용
        Deque<int[]> stack = new ArrayDeque<>();
        int cur = k;

        for (String c : cmd) {
            char op = c.charAt(0);
            if (op == 'U' || op == 'D') { // 위, 아래
                int x = Integer.parseInt(c.substring(2));
                if (op == 'U') while (x-- > 0) cur = prev[cur];
                else while (x-- > 0) cur = next[cur];
            } else if (op == 'C') { // 삭제
                stack.push(new int[]{cur, prev[cur], next[cur]});
                alive[cur] = false;
                if (prev[cur] != -1) next[prev[cur]] = next[cur];
                if (next[cur] != -1) prev[next[cur]] = prev[cur];
                cur = (next[cur] != -1) ? next[cur] : prev[cur];
            } else { // 복구
                int[] rec = stack.pop();
                int row = rec[0], p = rec[1], q = rec[2];
                alive[row] = true;
                if (p != -1) next[p] = row;
                if (q != -1) prev[q] = row;
                prev[row] = p;
                next[row] = q;
            }
        }

        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < n; i++) sb.append(alive[i] ? 'O' : 'X');
        return sb.toString();
    }
}

• 장점
  • 정석 풀이? 같습니다
  • 구현이 매우 쉬움
  • 가장 빠름
    • 삭제/복구 O(1), 이동 O(x)
• 단점
  • 펜윅트리 대비 메모리가 아~~주 근소하게 더 필요

제미나이가 만들어준 썸네일 후보로 마무리하겠습니다. Easy, right? 개열받음

AI를 바라보는 느낌이 딱 이 움짤 같습니다. 취업이 힘들어지긴 했는데 이 자식 없이 어떻게 살았을까 하는 생각이 싸악 올라옵니다. 강력한 선배 개발자들의 피땀노력이 전부 인터넷에 오픈된 덕분에 지식을 흡수한 AI만 활용해도 어느정도 그럴듯한 개발이 가능해졌습니다. 물론 내부가 어떻게 돌아가는지 확인하고 성능 이슈를 피하기 위해서는 여전히 개발자가 필요하지만 언젠간 이것도...AI가 다 하는 세상이 오지 않을까 싶습니다. 서론이 길었지만 이미 맞이한 AI 시대, 개발자는 그럼 어떻게 살아가야 하는지 Claude CLI를 활용하는 바이브 코딩에 대해 알아볼까 합니다.

1. 사용 이유 & 후기

저의 후기는 아니고 다른 분들의 후기입니다. 각자 느끼는 장단점이 거의 비슷합니다. 긍/부정이 갈린 것이지 장단점은 다 똑같다는 뜻입니다.

왜?

근본적인 이야기라 생각한다. 모든 공학도들은 "만들기"를 "효율적"으로 하는 것을 좋아한다. 원래 토목을 전공했으니 비유를 해보자면 콘크리트는 미친 튼튼함을 미친 효율로 만드는 건설 자재라 사용한다. 별 조합을 해보면서 가장 효율적인 방법을 찾았고 그것이 토목 공학이다. 컴퓨터공학도 마찬가지라 생각한다.

즉 효율성이 핵심입니다. 개발자들은 반복 구현, 스타일 맞추기 (컨벤션) 등에 시간을 많이 소요했었는데 이제 AI가 이 과정을 자동화 해주기 때문에 사용합니다.

AI는 코드를 대신 써주는 보조 도구로 사용했었지만 이젠 기업 또는 개발자 개인이 가진 취향, 원칙, 제약도 코드에 반영해줍니다. EX)

• Service 계층은 비즈니스 로직만 구현
• Entity에 Setter 금지
• Log에 개인정보 금지

이런 규칙은 문서로만 기록되거나 코드 리뷰에서 계속 언급되지만 실제 작성 단계에서는 AI가 자동으로 반영하지 못했습니다. 하지만 바이브 코딩의 등장으로 이 보이지 않는 원칙을 AI 프롬프트와 명령으로 고정할 수 있게 되었습니다.

장단점

장점

• 개발 접근성 향상 및 민주화
  • 구글 프로젝트할 때도 '데이터 민주화' 라는 단어를 사용했는데 AI를 통해 모두가 쉽게 개발에 접근할 수 있게됨을 극단적으로 표현하기엔 이만한 단어가 없는 것 같습니다.
  • 프로토타이핑이 압도적입니다. 프론트 프로토타입을 몇 분안에 구현할 수 있게 됐습니다.
• 효율성 + 생산성
  • 자연어로 작업 지시, 개발자가 부족한 스타트업 같은 환경에서 압도적인 효율을 보입니다.
  • 대규모 코드 생성 Cursor가 하루에 거의 10억줄의 코드를 생성한다고합니다.
  • 개발자가 프로그래밍보다는 "설계자" 역할에 가까워졌습니다.

단점

• 보안 취약성 + 유지보수
  • 기술 부채가 쌓이면 결국 개발자가 모든 코드를 살펴봐야합니다.
  • 검증되지 않은 코드를 그대로 사용 시 보안상 문제가 발생할 수 있습니다.
• 기반 역량 저하
  • 쉽게 접근할 수 있게된 만큼 필연적으로 기초 지식이 부족해졌습니다.
  • 코드 작동 흐름을 제대로 파악하는 엔트리가 눈에 띄게 줄었습니다.
• 프로덕션 환경 부적합
  • 개인용 프로젝트엔 적합하나 기업 프로젝트에 적용하기엔 무리가 있습니다.
  • LLM의 비예측성 특성으로 인해 구조적 이해가 부족하면 디버깅이나 확장 시 제어가 어렵습니다.

꽤나 명확한 한계에도 불구하고 여러 기업에서 개발자 목을 치고 사용하는 것엔 너무 압도적인 효율에 있다 생각합니다. 이젠 바이브 코딩에 가장 많이 사용하고 있는 Claude Code (CLI)에 대해 알아보겠습니다.

2. Claude Code

현황

SWE-bench에 따르면 현재 Claude 4 Opus의 성능이 LLM Model들 중 1위라고 합니다. 실제로 현업에 계신 분들에게도 물어보니 주로 개발을 하면서는 Claude, Cursor, Gemini CLI를 더 자주 사용하신다고 합니다.

사용 방법

Anthropic 공식 홈페이지에 잘 설명되어 있습니다. Node.js 18 혹은 그 이상 버전이 설치되어 있다면 아래 명령어로 바로 사용 가능합니다.

# Install Claude Code
npm install -g @anthropic-ai/claude-code

# Navigate to your project
cd your-awesome-project

# Start coding with Claude
claude
# You'll be prompted to log in on first use

# 참고
# 정상 설치 확인
claude doctor

# 업데이트
claude update

이후 터미널에 claude 입력 시

이렇게 정상 실행되시고 설정 진행해주시면 끝입니다.

세션 기능

Claude CLI는 한번 실행하고 끝나는 것이 아니라 세션을 이어갈 수 있는 기능을 제공합니다. EX) 프로젝트의 맥락을 학습시킨 뒤 대화 이어나가기

# 새로운 세션 시작
claude "이 프로젝트의 빌드 구조를 요약해줘."

# 이어서 같은 세션 사용
claude -r "<세션ID>" "Service 계층의 테스트 전략도 알려줘."

이런 식으로 사용하시면 매번 같은 프롬프트를 입력하지 않아도 됩니다.

3. 예시

목표와 범위

• 사용자가 제약(채널, 시간, 등급, 장르, 특집)을 입력하면 LLM이 일간 편성표를 생성해줌
• 기능 범위
  • 편성 생성 (Form 입력 -> LLM 호출 -> JSON 편성표 반환)
  • 결과 조회, 다운로드 (표 형태로 렌더링, JsON/CSV 저장)
• 비기능
  • 완전 빠른 프로토타이핑
  • API키 = 환경변수로
  • 로깅 및 프롬프트 추적 (간단하게)

아키텍쳐 개요

• Frontend : Static HTML + Fetch API (단일 페이지)
• Backend : SpringBoot 3 이상, Java 17, REST API
  • POST /api/schedule/generate : 제약을 받아 LLM 호출 → 편성표 JSON 응답
  • GET /api/health : 헬스체크
• LLM Provider: Open AI
  • ENV : LLM_API_KEY, LLM_BASE_URL, LLM_MODEL

프로젝트용 프롬프트

“우리는 Java 17 + Spring Boot 3.3로 REST API를 만들 거야. 목표는 ‘LLM 기반 방송 편성표 생성 API’.
요구사항:

POST /api/schedule/generate: 입력(날짜, 채널 수, 시간대 범위, 장르 분포, 연령등급 제한 등)을 받아 LLM에게 프롬프트→ JSON 편성표 반환

OpenAI 호환 API 호출(환경변수: LLM_API_KEY, LLM_BASE_URL, LLM_MODEL)

DTO, Controller, Service, LLM 클라이언트 계층 분리

프롬프트 템플릿과 시스템 프롬프트를 별도 클래스/리소스로 분리

간단한 HTML 파일로 결과를 호출/표 렌더

Build: Gradle, 패키지명: com.example.scheduler
위 스펙으로 패키지/파일 구조와 최소 동작 코드를 생성해줘.”

이어서: “컨벤션: Entity에 Setter 금지, Service는 비즈니스 로직만, 로그에 개인정보 금지, Response는 불변 DTO, Javadoc 달아줘.”

Outro

어떠셨나요? AI가 지배하는 세상을 어떻게 헤쳐나갈지 감이 좀 오시나요? 저는 아직 오지 않아서 다음 부터는 바이브 코딩으로 뚝딱뚝딱 무엇인가를 만들어보는 글을 작성해보려합니다. 마음 속 짐인 빅분기 필기가 끝나있을 것이기 때문입니다.

-

요즘 채용 시장에 코테가 하나 둘씩 사라지고 있습니다. 조심스럽지만 이젠 정말 코딩 실력보단 설계 능력이나 AI를 기깔나게 패서 올바른 방향성을 제시하는 능력이 더 중요하다 보는 것이 아닌가 싶습니다. 아아~ 열심히? 공부했는데 이미 AI가 다 알고있고 나보다 더 자세히 알고 실수도 나보다 적은 이 세상 어쩌면 좋지.

-

주절주절했지만 결국 활용은 인간이 하는 것 아니겠습니까? 시대가 변했으면 적응하면 될 일. 다음엔 뚝딱 만들어서 그 과정을 잘 정리해보겠습니다.

최근에 유튜브를 보다가 정말 재밌는 영상을 찾았습니다. 진격의 거인으로 AWS 네트워킹을 설명한 영상이었는데, 거대한 벽과 거인을 비유로 삼아 네트워킹 개념을 풀어내는 방식이 너무 신선했습니다. 보자마자 “이거 꼭 글로 풀어보고 싶다”는 생각이 들었습니다.

요즘 AI가 워낙 시장을 주도하다 보니 CS의 중요성이 떨어진 것 아니냐는 시각도 종종 보입니다. 하지만 실제로 개발을 하다 보면 막히는 문제의 상당수가 CS 기초 부족에서 비롯되곤 합니다. 특히 네트워크는 눈에 보이지 않는 추상적인 영역이라 더더욱 개념을 제대로 잡기 어려운 것 같습니다.

그래서 이번 글에서는 영상의 비유 속에 숨어 있는 네트워크 CS 개념들을 함께 짚어보며 놓치기 쉬운 네트워크 기초를 되새겨보겠습니다.

1. 네트워크 격리와 LAN

거인을 막는 벽, VPC

네트워크 격리는 신뢰 경계를 만드는 행위입니다. 주소 공간, 전송 경로, 정책을 분리해 멀티 테넌시, 보안을 적용합니다. AWS의 VPC는 이 격리를 소프트웨어 정의 네트워킹 (SDN)으로 구현한 결과물이고, CS 관점에서는 LAN/VLAN/VRF/오버레이 개념이 응축되어 있습니다.

아마도 같은 이미지

• 네트워크 격리?

  • 벽이 없으면 거인이 사람 먹방 진행 ➡️ 서버를 모두에게 공개된 상태로 열어두게 되면 다양한 공격이 들어와서 서버 먹방 진행
  • VPC는 거인을 막기 위한 벽.

• VPC의 정체

  • VPC는 가상 사설망으로 L3 네트워크 계층 격리 도메인
  • 즉, 하나의 독립된 라우팅 영역 (사설 주소 공간 + 라우팅 테이블)을 의미
  • LAN (Local Area Network) : VPC 그 자체가 하나의 거대한 가상 LAN. 물리적인 케이블 연결 없이도 같은 네트워크 영역에 있는 것처럼 통신할 수 있는 환경 제공
  • VLAN (Virtual Local Area Network) : VPC 내에서 서브넷을 나누는 행위가 VLAN과 유사. 하나의 큰 네트워크 (VPC)를 논리적으로 작은 도메인 (서브넷)으로 분할하여 네트워크 트래픽을 효율적으로 관리하고 보안을 강화. (L2 분리)
  • VRF (Virtual Routing and Forwarding) : VPC마다 독립적인 라우팅 테이블을 갖는 것이 VRF의 핵심 개념. 각 VPC는 자신만의 가상 라우터를 가지고 있어 다른 VPC의 라우팅 정보와 완전히 격리.따라서 VPC에서 10.0.0.0/16 같은 동일한 사설 IP 대역을 충돌 없이 사용 가능.

• 필수 구성

  • 주소 공간 : 이미지 속 10.0.0.0/16 같은 사설 IP 대역 필요
  • 세분화 : 이 대역을 서브넷으로 나눔 (이미지 속 10.0.1.0/24 & 10.0.2.0/24)
  • 라우팅 : 서브넷 간 통신 경로와 게이트웨이
  • 정책 : 누가 어디로 드나들 수 있는지 방화벽 규칙 부여
  • 👽 진격의 거인 세상에서 벽 안에 마을에서 사용할 주소 (사설 IP) + 용도에 맞는 땅 분할 (서브넷) + 구역 간 이동하는 길 (라우팅), 도시 밖으로 나가는 문 (게이트웨이) + 법 (정책)

• EX) 로컬 환경에서 서브넷 2개 만들고 라우터 통해서 통신

  # 네임스페이스(가상의 네트워크 공간) 3개: router, host1, host2
  sudo ip netns add router
  sudo ip netns add host1
  sudo ip netns add host2
  

가상 케이블(veth)로 연결: router-host1, router-host2

sudo ip link add veth-h1 type veth peer name veth-r1 sudo ip link add veth-h2 type veth peer name veth-r2 sudo ip link set veth-h1 netns host1 sudo ip link set veth-h2 netns host2 sudo ip link set veth-r1 netns router sudo ip link set veth-r2 netns router

IP를 다른 대역으로 부여(서브넷 분리)

sudo ip netns exec host1 ip addr add 10.0.1.10/24 dev veth-h1 sudo ip netns exec router ip addr add 10.0.1.1/24 dev veth-r1 sudo ip netns exec host2 ip addr add 10.0.2.10/24 dev veth-h2 sudo ip netns exec router ip addr add 10.0.2.1/24 dev veth-r2

인터페이스 up + 기본 게이트웨이(router)

for ns in host1 host2 router; do sudo ip netns exec $ns ip link set lo up; done sudo ip netns exec host1 ip link set veth-h1 up sudo ip netns exec host2 ip link set veth-h2 up sudo ip netns exec router ip link set veth-r1 up sudo ip netns exec router ip link set veth-r2 up sudo ip netns exec router sysctl -w net.ipv4.ip_forward=1 >/dev/null

sudo ip netns exec host1 ip route add default via 10.0.1.1 sudo ip netns exec host2 ip route add default via 10.0.2.1

host1 -> host2 통신: 라우터를 경유해야만 가능

sudo ip netns exec host1 ping -c 2 10.0.2.10

# 2. IP 주소와 서브네팅

![](https://velog.velcdn.com/images/cup-wan/post/1fe00241-71a0-482d-a2e5-5fa7525dc9c0/image.png)


> __성벽 안의 효율적인 도시 계획__

벽 안의 도시를 건설해야 합니다. 서브네팅은 벽 내부의 땅을 용도에 맞게 __구역을 나누는 계획__입니다.

- IP 주소 : 모든 건물의 고유 번호
  - 도시의 모든 건물이 'OO로 OOO-OO'처럼 주소를 갖듯, 네트워크의 모든 장비는 IP 주소를 가짐.
  - 우리가 흔히 보는 10.0.12.34는 사실 컴퓨터가 이해하는 32자리 2진수(00001010.00000000...)를 사람이 읽기 쉽게 바꾼 것
  - __도시 계획 규칙, CIDR__: 10.0.12.34/20 같은 표기법에서 /20이 바로 도시 계획의 핵심 규칙. 
  "앞 20자리는 이 건물이 속한 구역(네트워크) 주소이고, 나머지 12자리는 구역 내 **상세 주소(호스트)**야"라고 알려주는 약속.
  - 네트워크 주소: 구역의 시작 주소 (10.0.0.0)로, '강남구'처럼 특정 구역 전체를 지칭
  - 브로드캐스트 주소: 구역의 마지막 주소 (10.0.15.255)로, '강남구 재난 알림'처럼 구역 내 모든 주민에게 소식을 전파할 때 사용되는 특별 주소

- 사용 가능 주소: 한 구역(/20 = 12비트 호스트)에 지을 수 있는 건물의 수는  2^12−2 (구역 대표 주소, 방송 주소 제외) = 4,094개

- Public IP / Private IP + NAT
  - Private IP : 도시 안에서만 사용하는 **내부 주소 체계**
    - 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16
    - 인터넷 상에서 라우팅되지 않는 사설 공간 ➡️ 대규모 내부망 설계 시
  - Public IP : 전 세계에서 유일한 **공식 주소**, 인터넷에서 라우팅되는 주소
  - NAT : Private ↔️ Public 주소 변환. (후에 자세히 기술)
  - AWS에서 Public/Private 서브넷을 가르는 기준은 주소 X, 라우팅 O
    - 퍼블릭: 0.0.0.0/0 경로의 다음 홉이 IGW (인터넷 Gateway)
    - 프라이빗: 0.0.0.0/0 경로의 다음 홉이 NAT 게이트웨이/인스턴스

- VLSM (Variable Length Subnet Mask) : 효율적인 도시 계획 기술
  - 대형 마트, 개인 집 모두 10평씩 나눠 가지면 비효율적 ➡️ 각자 다른 크기로 설정해주자 ➡️ VLSM
  - 같은 상위 대역 안에서 서브넷마다 다른 크기로 나눌 수 있게 해줌
  - 10.0.0.0/16을 용도별로 자르기
    - /20 (웹 트래픽 많음) x 4개
    - /22 (중간 규모) x 4개
    - /24 (소규모) x 여러 개
  - 주소 낭비 최소화 + 집계 용이해서 사용

- EX) 하나의 /16을 가용영역(AZ) 으로 쪼개기
웹 / 앱 / DB / 관리망을 2개의 가용영역(AZ)에 대칭해서 배치할 때
```text
1. 상위 대역 확보: 10.0.0.0/16

2. AZ별 블록 나누기: AZ-a는 10.0.0.0/17, AZ-b는 10.0.128.0/17

3. 역할별 서브넷(각 AZ에 4개씩): /20(4094호스트) 기준

AZ-a

  Web-a: 10.0.0.0/20 (퍼블릭, 0.0.0.0/0 → IGW)

  App-a: 10.0.16.0/20 (프라이빗, 0.0.0.0/0 → NAT-a)

  DB-a : 10.0.32.0/20 (프라이빗, 외부로 매우 제한)

  Mgmt-a: 10.0.48.0/20 (프라이빗, 점프/모니터링만)

AZ-b

  Web-b: 10.0.128.0/20

  App-b: 10.0.144.0/20

  DB-b : 10.0.160.0/20

  Mgmt-b: 10.0.176.0/20

AWS 주의: 각 서브넷에서는 일부 IP가 플랫폼 예약(게이트웨이·DNS 등).

Router 설계

  Web 서브넷: 0.0.0.0/0 → IGW

  App/DB/Mgmt: 0.0.0.0/0 → NAT (AZ별)

  내부 통신: VPC 내부 라우팅으로 상호 통과(필요 시 SG/NACL로 통제)

3. 방화벽과 접근 제어

벽도 세웠고 도시도 세웠으니 이제 문지기를 세워봅시다. 아무나 특정 구역에 들어가거나, 핵심 건물(서버)에 접근한다면 도시가 불에 탑니다. 이제 도시의 질서를 유지하기 위해 곳곳에 문지기 (방화벽)을 배치해야 합니다. AWS VPC의 문지기는 Security Group과 Network ACL (NACL)이 있습니다.

• Security Group (SG) : 건물을 지키는 경비원

  • SG는 각 서버 (EC2 인스턴스)의 문 앞으을 지키는 경비원
  • 서버에 들어오고 나가는 모든 트래픽은 SG를 거쳐야함
  • 특징
    • Stateful : 안으로 들어온 손님..? (Inbound)을 기억함. 해당 손님이 나갔다 올 때 (Outbound) 별도의 검사 없이 들어오기 가능 즉, 나가는 트래픽에 대한 규칙을 따로 설정할 필요가 없음
    • Allow 규칙만 사용 : Allow List만 잇고 Deny List가 없음
  • 적용 단위 : 서버(인스턴스)단위로 적용, 웹 서버 경비원, DB 서버 경비원처럼 각 특성에 맞게 개별 배치 가능
  • EX) 웹 서버 인스턴스는 전 세계 누구나 웹 서핑 (HTTP/HTTPS 포트)을 위해 들어올 수 있도록 허가

• Network Access Control List (NACL) : 구역을 지키는 검문소

  • NACL은 SG보다 더 넓은 범위 담당 ➡️ 서브넷의 입구와 출구를 지킴
  • 특징
    • Stateless : 각 패킷을 그때그때 새로 판단 ➡️ 클라이언트가 밖으로 나가면 돌아오는 응답도 인바운드 규칙에 허용이 있어야 통과
    • Allow/Deny 모두 있음
    • 규칙 번호 순서로 평가 (번호 작을수록 우선)
    • 적용 단위 : 서브넷 단위 (서브넷 내 모든 ENI에 적용)

• 언제 어떻게 사용??

  • 일반적으로 SG 중심으로 세밀한 제어, NACL은 완화/거부 몇 개만 운영해 복잡도/운영 비용을 낮춤
  • 명확한 커트라인 (외부 스캔 대역 차단, 특정 포트 차단 등)이 필요할 때 NACL의 Deny를 보조로 사용

4. 주소 변환과 NAT

벽 안에서만 살다보니 거인놈들이 뭐하는지 궁금합니다. 다른 도시에서 새로 개발한 무기 (SW 업데이트)도 있다고 하는데 어떻게 통신할 수 있을까요? 이를 해결해주는 것이 NAT 입니다. 내부 주소(사설 IP)는 도시 안에서만 통용되는 주소입니다. 이 주소로 외부와 통신하기 위해서는 Pulbic IP로 변경해줘야 합니다.

NAT는 사설망의 주소를 밖과 대화할 수 있게 바꿔주는 중간역 (게이트웨이). IPv4 고갈과 보안 경계 유지를 해결

• NAT가 필요한 이유

  • IPv4 고갈 IPv4는 32비트 주소 체계로 43억 개의 고유 주소를 표현할 수 있습니다. 하지만 인터넷 세상은 호락호락하지 않아서 43억개로도 부족합니다.
  • 경계 보안 내부 호스트가 아웃바운드로만 인터넷을 사용하고 인바운드로 임의 유입되는 걸 막아야 합니다.
  • NAT가 내부(Private IP)와 외부(Public IP) 사이에서 주소를 변환해 통신을 가능하게 합니다.

• NAT의 종류

  • SNAT (Source NAT) : 출발지 주소를 변경 내부 호스트가 외부로 나갈 때 출발지를 Public IP로 치환
  • DNAT (Destination NAT) : 목적지 주소를 변경 외부에서 들어오는 트래픽을 내부 서비스로 본래 때 사용
  • NPAT / PAT (Port Address Translation) 여러 내부 호스트가 하나의 공인 IP를 포트로 구분해 공유

• AWS에서의 NAT vs IGW(Internet GateWay)

  • IGW
    • 퍼블릭 IP를 가진 ENI/인스턴스가 직접 인터넷과 양방향 통신
    • 퍼블릭 서브넷 : 0.0.0.0/0 ➡️ IGW 라우트 (인바운드는 SG/NACL 허용시만)
  • NAT Gateway (관리형)
    • 프라이빗 서브넷이 아웃바운드로 인터넷 리소스 접근 시 사용
    • 퍼블릭 서브넷에 배치 + EIP 할당 + 해당 가용공간의 프라이빗 서브넷 라우트
    • 인바운드 신규 연결은 불가능. 관리, 확장, 가용성은 AWS가 맡음
  • NAT 인스턴스 (Self Managed)
    • EC2로 NAT 구성. 유연하고 비용이 적지만 HA, 스케일, 패치 등을 직접 책임짐
    • Source/Desk Check 해제 필요, 헬스 체크/오토스케일/멀티AZ 등 이중화 설계 필수
    • IPv6에서는 NAT가 아닌 Egress-Only IGW
      • IPv6는 퍼블릭 가정이라 NAT 대신 Egress-Only Internet Gateway로 아웃바운드만 허용

• EX) Linux로 NAT 만들어보기 내부망 (10.0.0.0/16)이 et1, 외부망(인터넷)이 eth0인 리눅스 라우터에서 SNAT(PAT)로 외부 접근 허용해보기

  # 1. 커널 포워딩 켜기
  sudo sysctl -w net.ipv4.ip_forward=1
  

2. SNAT (MASQUERADE = 동적 퍼블릭 IP일 때 편함)

sudo iptables -t nat -A POSTROUTING -s 10.0.0.0/16 -o eth0 -j MASQUERADE

3. 포워딩 정책(기본 거부 후 필요한 것만)

sudo iptables -A FORWARD -i eth1 -o eth0 -m state --state NEW,ESTABLISHED,RELATED -j ACCEPT sudo iptables -A FORWARD -i eth0 -o eth1 -m state --state ESTABLISHED,RELATED -j ACCEPT

- DNAT: 목적지를 내부 서버로

sudo iptables -t nat -A PREROUTING -d 203.0.113.5 -p tcp --dport 443 -j DNAT --to-destination 10.0.1.20:443

- SNAT: 응답이 다시 NAT를 경유하도록 출발지도 NAT의 내부 IP로

sudo iptables -t nat -A POSTROUTING -s 10.0.0.0/16 -d 10.0.1.20 -p tcp --dport 443 -j SNAT --to-source 10.0.1.1

>(➕) IPv4의 주소 고갈은 IPv6로 해결된 것이 아닌가?
실제로 IPv6 표준 철학이 __NAT가 필요 없는 End-to-End 통신 복원__입니다. 
하지만 IPv6의 보급 지연, 경계 보안 기능 (NAT 자체가 보안을 위한 것은 아니지만 기능이 외부의 임의 접근이 막히는 효과), 주소 은닉 등으로 인해 여전히 NAT가 필요합니다.


# 5. 트래픽 관리와 고가용성

![](https://velog.velcdn.com/images/cup-wan/post/f4aee3f5-15c9-4c22-83d9-48b97c5aa8a8/image.png)

도시가 잘 되니 방문객도 많아지고 상인도 많아졌습니다. 이렇게 몰리면 결국 제 기능을 못하고 도시가 또 멸망합니다. 이를 해결하기 위해서는 __새로운 도시를 세워서 방문객을 분산__해야합니다. (영상에서는 몸빵으로 비유)

> __로드 밸런싱은 연결을 어디로 보낼지 결정__
L4 전송 계층은 IP/포트/프로토콜만 보고 분산 (연결 종단 X)
L7 애플리케이션 계층은 HTTP/gRPC 같은 애플리케이션 레빌을 이해해 헤더/경로 기반으로 분산 (연결 종단 O)

- 단일 서버의 처리량/가용성에는 한계가 존재
  - 장애 발생 시 전체 중단 위험
  - (물론 분산 서버 되어있겠지만) 티켓팅, 수강 신청 등 다양한 이벤트로 서버가 죽음..
  - 서버를 늘려서 수평 확장 (Scale-out) + 헬스 체크로 불량 서버 제외
- L4 vs L7

| 구분    | L4 (TCP/UDP)                    | L7  (HTTP/HTTPS/gRPC)                                   |
| -------- | ------------------------------------- | ------------------------------------------------------------ |
| 기준 | src/dst IP, src/dst Port, 프로토콜 | URL 경로/메서드/헤더/쿠키/호스트명                                        |
| 연결 처리 | **패스스루**(비종단),  빠름                  | **종단(Proxy)**, 라우팅 유연, 기능 ⬆️                                 |
| 기능    | 소스 IP 보존 용이, 낮은 지연                    | 경로/호스트 기반 라우팅, 헤더 재작성, 쿠키 기반 **세션 스티키**, WAF 연계, TLS 종료/재암호화 |
| 사용처   | 게임 서버, 메시지 큐, DB 프록시 등      | 웹/모바일 API, gRPC, 다중 도메인/경로 라우팅 등     |

- 분산 알고리즘
  - 라운드 로빈 Round Robin
  - 가중치 라운드 로빈 Weighted RR
  - Least Connections
  - Hash
  - Consistent Hash

- 헬스 체크
  - Active : 로드밸런서가 주기적으로 /health (HTTP) 또는 TCP 핸드셰이크 체크
  - Passive : 실제 트래픽 에러, 타임아웃 관찰 후 타깃 제외
  - Slow-Start/Outlier Detection : 새로 합류한 서버는 천천히 트래픽 유입, 오류 많은 서버는 자동 격리
  - 멀티 AZ/리전 : 헬스 체크는 가까운 곳에서 빠르게, 실패 임계/간격은 서비스 RTO/RPO와 맞추기

- 세션 관리 : 스티키 vs 무상태
  - 스티키 세션 : L7 로드밸런서가 쿠키/소스 IP 기반으로 같은 서버에 붙여주는 것
    - 장점 : 서버가 메모리에 세션을 저장해도 무방
    - 단점 : 서버 증/감, 장애 발생 시 세션 유실/재분산 이슈 발생
  - 무상태 권장 : 세션 외부화 (Redis) 또는 토큰 기반 (JWT)

- TLS 처리 패턴 (종단 지점 설계)
  - LB에서 종료(Termination): LB가 인증서 보유, 백엔드에 HTTP(또는 re-encrypt).
    - 장점: 인증서/암복호화 중앙화, WAF/L7 기능 사용 가능.
    - 단점: 백엔드까지 평문이라면 경로 보호 필요(프라이빗 전용망, SG/NACL).
  - 패스스루(Pass-Through): L4로 그대로 전달(서버가 인증서 보유).
    - 장점: 소스 IP 보존/단순
    - 단점: 도메인별 라우팅/헤더 조작 불가.
  - Re-encryption(종단→재암호화): LB에서 종료 후 백엔드로 다시 TLS.
     - 장점: 경로 암호화 유지 + L7 기능 사용.
     - mTLS, SNI, ALPN(HTTP/2, gRPC)과의 궁합을 고려.

# Outro

![](https://media1.giphy.com/media/v1.Y2lkPTc5MGI3NjExNzZzN2VudDI3NWkxd2V2dXQ3Nmx4cWswbjB6cXF0MHZjOGVwZzJtYiZlcD12MV9pbnRlcm5hbF9naWZfYnlfaWQmY3Q9Zw/kgo9KtvX2dCVudr645/giphy.gif)

쓰다보니 진격거는 사라지고 어째 네트워크만 잔뜩 설명된 느낌인데 한번에 쭉 정리해보니 정말......어렵네요. 세부적인 내용을 전부 빼고 정말 AWS 한정으로 봤는데도 방대합니다. 학교에서 배웠을 때는 종이컵 전화기 수준부터 배웠던 것 생각하면 어떻게 다 배웠나 싶습니다. 이제 클라우드는 개발자에게 필수 도구로 자리잡은 만큼 관련된 내용도 많이 올려보겠습니다.

Redis 로고가...바뀌었네요???????? 공식은 왼쪽을 밀고 있지만 저는 여전히 오른쪽이 공식 아이콘이라 생각합니다. 글 쓰려고 공식 사이트 좀 보니까 생성형 AI 시대에 맞춰서 Vector DB 로써 타사 제품과 비교한 글도 있네요. 실시간성이 중요한 데이터 플랫폼이 필요한 경우 하나의 선택지가 될 수 있을 것 같습니다. +) Redis의 벡터 검색 지원은 7.2 버전부터 입니다.

1편에서 Redis의 Cache-Aside, 분산 락, Rate Limiting 에 대해 살펴봤습니다. Redis를 활용해서 Read 성능, 안정성, API 보호까지 다양한 활용 방법에 대해 알아봤는데 더욱 기발한 활용법이 많습니다. 그 중 예약 이체를 구현하는 예제를 통해 토큰, 작업 큐 & 지연 큐, 실시간 Pub/Sub (vs Stream), 순위/리더보드에 대해 알아보겠습니다.

1. 토큰

인증 후 발급되는 세션 ID나 JWT Refresh Token을 빠르고 안전하게 저장하는 용도로 Redis가 자주 사용됩니다.

• TTL로 자동 만료를 쉽게 다룰 수 있어 RDBMS 보다 효휼이 좋습니다.
• 로그아웃 / 강제 만료 / 블랙리스트가 필요한 보안에 민감한 서비스에 적합합니다.

키 설계

구현

1️⃣ Refresh Token 저장, 검증, 삭제

• 저장 시 해싱 후 저장

• 검증 로직 추가 가능

• 삭제 시 특정 디바이스에서만 로그아웃, 모든 디바이스 로그아웃 등 구현 가능

  @Service
  @RequiredArgsConstructor
  public class TokenService {
    private final StringRedisTemplate redis;
    private static final String RT_PREFIX = "auth:rt:"; // auth:rt:{userId}:{deviceId}
  
    private String key(long userId, String deviceId) {
        return RT_PREFIX + userId + ":" + deviceId;
    }
  
    // 저장 : HMAC 등으로 해시해 보관
    public void saveRefreshToken(long userId, String deviceId, String hashedRt, Duration ttl) {
        redis.opsForValue().set(key(userId, deviceId), hashedRt, ttl);
        // 기기 목록 관리
        redis.opsForSet().add("auth:devices:" + userId, deviceId);
    }
  
    // 검증
    public boolean validateRefreshToken(long userId, String deviceId, String hashedRt) {
        String saved = redis.opsForValue().get(key(userId, deviceId));
        return saved != null && saved.equals(hashedRt);
    }
  
    // 삭제 (특정 기기 로그아웃)
    public void deleteRefreshToken(long userId, String deviceId) {
        redis.delete(key(userId, deviceId));
        redis.opsForSet().remove("auth:devices:" + userId, deviceId);
    }
  }

2️⃣ Access Token 블랙리스트

• 로그아웃/탈취 의심 등으로 이미 발급된 Access Token을 즉시 무효화 하기 위해 사용
• 토큰의 jti(고유 ID)를 키로 사용하고 TTL을 남은 만료 시간과 동일하게 설정

@Service
@RequiredArgsConstructor
public class AccessTokenBlacklist {
    private final StringRedisTemplate redis;

    // Blacklist = 즉시 차단
    public void blacklist(String atJti, Duration remainingTtl) {
        redis.opsForValue().set("auth:bl:at:" + atJti, "1", remainingTtl);
    }

    // 필터나 게이트웨이에서 사용되는 검사 메서드
    public boolean isBlacklisted(String atJti) {
        return Boolean.TRUE.equals(redis.hasKey("auth:bl:at:" + atJti));
    }
}

+) JWT 필터에서의 사용 예시

// 요청마다 AT의 jti 추출 후 블랙리스트 조회
String jti = claims.getId(); // jti
if (accessTokenBlacklist.isBlacklisted(jti)) {
    throw new Unauthorized("Token revoked");
}

3️⃣ Refresh Token Rotate + 재사용 감지

• 사용자가 Access Token 재발급 요청 후 기존 Refresh Token으로 인증되면 새 Refresh Token을 발급하고 이전 Refresh Token은 즉시 폐기
• 원자성 보장을 위해 사용
  • 이전 Refresh Token 삭제 -> 새로운 Refresh Token 발급이 동시에 이루어져야 경쟁 조건 X
• 검증 -> 교체 과정을 Lua 스크립트로 한 번에 처리하면 경쟁 조건 방지 가능

@Component
@RequiredArgsConstructor
public class RefreshTokenRotator {
    private final StringRedisTemplate redis;

    private static final String ROTATE_LUA =
        // KEYS[1]=oldKey, KEYS[2]=newKey, ARGV[1]=oldHash, ARGV[2]=newHash, ARGV[3]=ttlSeconds
        "if redis.call('get', KEYS[1]) == ARGV[1] then " +
        "  redis.call('del', KEYS[1]); " +
        "  redis.call('set', KEYS[2], ARGV[2], 'EX', ARGV[3]); " +
        "  return 1 " +
        "else return 0 end";

    public boolean rotate(long userId, String oldDeviceId, String newDeviceId,
                          String oldHashedRt, String newHashedRt, Duration ttl) {
        String oldKey = "auth:rt:" + userId + ":" + oldDeviceId;
        String newKey = "auth:rt:" + userId + ":" + newDeviceId;
        Long ok = redis.execute((RedisCallback<Long>) conn ->
            conn.scriptingCommands().eval(
                ROTATE_LUA.getBytes(StandardCharsets.UTF_8),
                ReturnType.INTEGER, 2,
                oldKey.getBytes(StandardCharsets.UTF_8),
                newKey.getBytes(StandardCharsets.UTF_8),
                oldHashedRt.getBytes(StandardCharsets.UTF_8),
                newHashedRt.getBytes(StandardCharsets.UTF_8),
                String.valueOf(ttl.toSeconds()).getBytes(StandardCharsets.UTF_8)
            )
        );
        return ok != null && ok == 1L;
    }
}

고려 사항

• 세션 관리를 위해서는 Spring Session Redis 사용 시 spring-session-data-redis를 쓰면 세션 TTL, 직렬화, 정리 작업의 표준화가 자동으로 이루어집니다.
• 클러스터/복제 : 인증은 서비스 핵심이므로 Redis는 클러스터 + 복제 + Sentinel/Cloud HA 구성을 권장합니다.
• 백업 전략 : 장기 세션/토큰은 가급적 짧게 운용하고 (TTL 설정 적당히), 필요 시 감사용 이벤트 로그를 별도 저장소에 저장합니다.

2. 작업 큐 & 지연 큐

목적

예약 이체를 위해 사용자가 T 시각에 A 원 이체를 등록하면,

• 현재 : 대기 상태
• T 시간 : 정확히 한 번 실행
• 실패 시 : 재시도/만기 처리(DLQ) 의 흐름이 됩니다.

이러한 작업 큐를 Redis로 구현할 수 있습니다.

키 설계

• ZSET?
  • Redis의 정렬된 집합 자료 구조형
  • score에 대한 member의 매핑 값 저장 가능
  • member + score로 지금 처리해야 할 작업을 효율적으로 pop 가능

구현

1️⃣ 예약 이체 등록 (Producer)

@Service
@RequiredArgsConstructor
public class PaymentScheduler {
    private final StringRedisTemplate redis;
    private static final String DELAY_KEY = "jobs:delay:payments";

    // 예약 생성 Produce
    // 작업 페이로드를 Hash에 저장하고, ZSET에 실행시각을 점수로 등록
    public void schedulePayment(String jobId, long executeAtMillis,
                                String userId, long amount, String toAccount) {
        // 1) 저장 (Hash) (필요 시 TTL 부여)
        String payloadKey = "job:payload:" + jobId;
        Map<String,String> payload = Map.of(
            "userId", userId,
            "amount", String.valueOf(amount),
            "to", toAccount,
            "attempts", "0"
        );
        redis.opsForHash().putAll(payloadKey, payload);

        // 2) ZSET에 스케줄링
        redis.opsForZSet().add(DELAY_KEY, "job:" + jobId, executeAtMillis);
    }
}

• ZSET에 추가할 때 executeAtMillis를 통해 score 부여
• 페이로드는 Hash에 저장
  • 만기 후 데이터 보존 기간이 있다면 payload TTL 설정
  • redis.expire(payloadKey, Duration.ofDays(N)

2️⃣ 만기 작업 꺼내기 (Consumer)

• ZSET 활용
  • ZRANGEBYSCORE : 범위 내에 해당하는 데이터 반환
  • ZREM : 해당 데이터 삭제
  • 두번의 호출로 경쟁 상태 가능
• 경쟁..상태? = Lua로 해결 가능

@Component
@RequiredArgsConstructor
public class DueJobPopper {
    private final StringRedisTemplate redis;
    private static final String DELAY_KEY = "jobs:delay:payments";

    // nowMillis 이하 작업을 최대 N개까지 pop (원자성 보장)
    private static final String POP_DUE_LUA =
        // KEYS[1]=ZSET, ARGV[1]=nowMillis, ARGV[2]=limit
        "local r = {} " +
        "local vals = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1], 'LIMIT', 0, ARGV[2]) " +
        "for i,v in ipairs(vals) do " +
        "  redis.call('ZREM', KEYS[1], v); table.insert(r, v) " +
        "end; return r";

    // 작업 꺼내기 Consume
    public List<String> popDue(long nowMillis, int limit) {
        List<String> res = redis.execute((RedisConnection conn) -> {
            @SuppressWarnings("unchecked")
            List<byte[]> raw = (List<byte[]>) conn.scriptingCommands().eval(
                POP_DUE_LUA.getBytes(StandardCharsets.UTF_8),
                ReturnType.MULTI, 1,
                DELAY_KEY.getBytes(StandardCharsets.UTF_8),
                String.valueOf(nowMillis).getBytes(StandardCharsets.UTF_8),
                String.valueOf(limit).getBytes(StandardCharsets.UTF_8)
            );
            if (raw == null) return List.of();
            return raw.stream().map(b -> new String(b, StandardCharsets.UTF_8)).toList();
        });
        return res == null ? List.of() : res;
    }
}

3️⃣ ⭐⭐⭐ 실행

@Service
@RequiredArgsConstructor
public class PaymentWorker {
    private final StringRedisTemplate redis;
    private final DueJobPopper popper;
    private final PaymentExecutor executor; // 외부 Executor 호출, 결과 반환
    private static final int BATCH = 100;

    @Scheduled(fixedDelay = 300) // 300ms마다 폴링
    public void tick() {
        long now = System.currentTimeMillis();
        for (String member : popper.popDue(now, BATCH)) {
            String jobId = member.substring("job:".length());
            process(jobId);
        }
    }

    private void process(String jobId) {
        // 1) 멱등성 키: 이미 처리 중 or 완료면 스킵
        boolean first = Boolean.TRUE.equals(
            redis.opsForValue().setIfAbsent("idem:payment:" + jobId, "1", Duration.ofMinutes(10))
        );
        if (!first) return;

        String payloadKey = "job:payload:" + jobId;
        Map<Object,Object> p = redis.opsForHash().entries(payloadKey);
        String userId = (String) p.get("userId");
        long amount = Long.parseLong((String)p.get("amount"));
        String to = (String) p.get("to");
        int attempts = Integer.parseInt((String)p.get("attempts"));

        try {
            PaymentResult r = executor.execute(jobId, userId, amount, to); // 외부 시스템 (PaymentExecutor)
            appendLedgerEvent(r); // → 3단계에서 스트림으로 영속/분석
            publishUserNotify(r); // → Pub/Sub 즉시 알림
        } catch (TransientException te) {
            // 일시 오류 → 재시도 예약 (지수 백오프)
            int next = attempts + 1;
            long delayMs = backoffMs(next); // 예: min(2^next*1000, 5분)
            redis.opsForHash().put(payloadKey, "attempts", String.valueOf(next));
            long requeueAt = System.currentTimeMillis() + delayMs;
            redis.opsForZSet().add("jobs:delay:payments", "job:" + jobId, requeueAt);
        } catch (Throwable fatal) {
            // 영구 실패 → DLQ로 이동
            Map<String,String> dlq = Map.of(
              "jobId", jobId, "userId", userId, "amount", String.valueOf(amount),
              "to", to, "error", fatal.getClass().getSimpleName()
            );
            redis.opsForStream().add("jobs:dlq:payments", dlq);
        } finally {
        }
    }

    private long backoffMs(int attempt) {
        long base = (long)Math.min(Math.pow(2, attempt) * 1000, 5 * 60 * 1000);
        // Jitter로 몰림 완화
        double jitter = 0.8 + Math.random() * 0.4;
        return (long)(base * jitter);
    }

    private void appendLedgerEvent(PaymentResult r) {
        Map<String,String> body = Map.of(
          "txnId", r.txnId(), "userId", r.userId(),
          "amount", String.valueOf(r.amount()), "to", r.toAccount(),
          "status", r.status().name(), "ts", String.valueOf(System.currentTimeMillis())
        );
        redis.opsForStream().add("txn:stream:ledger", body);
    }

    private void publishUserNotify(PaymentResult r) {
        redis.convertAndSend("notify:pub:" + r.userId(),
            "{\"type\":\"PAYMENT\",\"status\":\""+r.status()+"\",\"amount\":"+r.amount()+"}");
    }
}

• 멱등성 키

  • 예약 이체 실행 시 같은 요청 중복 실행 방지
  • SET key value NX EX ttl 패턴 사용
    • NX : 키 없을 때만 생성 (중복 방지)
    • EX : TTL 설정 (멱등성 보장 기간 설정)
  • ex) idem:transfer:{reservationId}라는 키를 걸어두면 같은 예약 ID로 중복 실행 시도 시 Redis가 첫 실행 이후에는 무시

• 실패 처리 : 지수 백오프 (Exponential Backoff) + DLQ

  • 은행 송금 API나 타 금융 기관 API 호출 실패 시
  • Exponential Backoff로 재시도 시점 간격을 점점 늘려주면, 시스템 부하가 줄고 일시적인 장애 회복 가능성 상승
  • 재시도 횟수가 일정 한도 초과 시 Dead Letter Queue (DLQ로 이동
    • DLQ에 쌓인 데이터는 운영자가 수동 처리 or 별도 재처리 파이프라인으로 복구

• Stream 기반 결과 로그

  • 예약 이체 성공/실패 여부를 Redis Stream에 기록
    • 순차적으로 안전한 기록
    • 이후 단계에서 비동기 처리 (기록, 감사 로그 생성, 리스크 모니터링 등)
    • 장애 발생 시 처리 지점부터 재시작 가능

추가) 지연 큐 처리 시간을 지수 백오프로 구현한 예제이지만 지연 큐 전용 라이브러리가 있다! 깃허브 링크

3. Pub/Sub

목적

• Pub/Sub 모델은 발행자(Pub)가 특정 채널에 메시지 발행 시 해당 채널을 구독한 모든 구독자(Sub)가 메시지를 실시간으로 받는 구조
• 메시지 전송 시 Redis 저장 X (휘발성)
• 예약 이체 결과를 실시간으로 알림 (송금 완료, 이체 실패 - 잔액 부족 등)
• Stream과 차이점
  • Strema : 안전한 저장/재처리가 필요한 데이터
  • Pub/Sub : 일시적이고 빠른 알림

구현

1️⃣ 발행 (Publish)

@Service
@RequiredArgsConstructor
public class TransferPublisher {
    private final StringRedisTemplate redisTemplate;

    public void notifyTransfer(long userId, String status) {
        String message = String.format("{\"userId\":%d,\"status\":\"%s\"}", userId, status);
        // 채널 용도별/사용자별 분리 가능: transfer:notify[:{userId}]
        redisTemplate.convertAndSend("transfer:notify", message);
    }
}

2️⃣ 구독 (Subscrive)

@Component
@RequiredArgsConstructor
public class TransferSubscriber implements MessageListener {
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody(), StandardCharsets.UTF_8);
        System.out.println("알림: " + body);
        // 여기서 WebSocket/SSE를 통해 클라이언트에 전달
    }
}

4. 순위/리더보드

목적

• 거래별 리스크 점수를 산출 중
• ZADD로 fraud:rank:alerts (ZSET)에 (score=리스크, member=alertId) 반영
• 거래 분석 시 상위 N건을 즉시 확인 및 선점해 중복 처리 방지

키 설계

추가) 버킷 분할{bucket}은 트래픽에 따라 월별, 리전 등으로 나누면 효과적

구현

__ 1️⃣ 점수 반영__

@Service
@RequiredArgsConstructor
public class FraudPriorityWriter {
    private final StringRedisTemplate redis;

    // 리스크 점수 산출 결과를 반영
    public void upsert(String bucket, String alertId, double riskScore) {
        String key = "fraud:rank:" + bucket;
        redis.opsForZSet().add(key, "alert:" + alertId, riskScore);
    }

    // 누적 가중이 필요한 경우
    public Double incr(String bucket, String alertId, double delta) {
        String key = "fraud:rank:" + bucket;
        return redis.opsForZSet().incrementScore(key, "alert:" + alertId, delta);
    }
}

2️⃣ 조회: 상위 N, 순위, 점수 구간

@Service
@RequiredArgsConstructor
public class FraudPriorityReader {
    private final StringRedisTemplate redis;

    public List<String> topN(String bucket, int n) {
        var key = "fraud:rank:" + bucket;
        var r = redis.opsForZSet().reverseRange(key, 0, n - 1);
        return r == null ? List.of() : r.stream().toList();
    }

    public Long rankOf(String bucket, String alertId) {
        var key = "fraud:rank:" + bucket;
        return redis.opsForZSet().reverseRank(key, "alert:" + alertId); // 0 = 1위
    }

    public Set<ZSetOperations.TypedTuple<String>> byScore(String bucket, double min, double max, int limit) {
        var key = "fraud:rank:" + bucket;
        return redis.opsForZSet().reverseRangeByScoreWithScores(key, min, max, 0, limit);
    }
}

3️⃣ 선점(claim) : 경쟁 없이 한 건 뽑아 배정

@Component
@RequiredArgsConstructor
public class FraudClaimer {
    private final StringRedisTemplate redis;

    // KEYS[1]=ZSET fraud:rank:{bucket}
    // ARGV[1]=nowMillis, ARGV[2]=analystId, ARGV[3]=claimTtlSeconds
    private static final String CLAIM_LUA =
        "local k=KEYS[1] " +
        "local top = redis.call('ZREVRANGE', k, 0, 0) " +
        "if #top == 0 then return nil end " +
        "local member = top[1] " +
        "redis.call('ZREM', k, member) " +                       -- 우선순위 큐에서 제거
        "redis.call('SET', 'fraud:claimed:'..member, ARGV[2], 'EX', ARGV[3]) " + -- 선점 상태 기록
        "return member";

    // 상위 1건을 원자적으로 꺼내 선점. 없으면 null
    public String claimTopOne(String bucket, String analystId, Duration claimTtl) {
        String zset = "fraud:rank:" + bucket;
        return redis.execute((RedisConnection c) ->
            (String) c.scriptingCommands().eval(
                CLAIM_LUA.getBytes(StandardCharsets.UTF_8),
                ReturnType.VALUE, 1,
                zset.getBytes(StandardCharsets.UTF_8),
                String.valueOf(System.currentTimeMillis()).getBytes(StandardCharsets.UTF_8),
                analystId.getBytes(StandardCharsets.UTF_8),
                String.valueOf(claimTtl.toSeconds()).getBytes(StandardCharsets.UTF_8)
            )
        );
    }
}

• 상위 항목 분석 요청 겹칠 때 가장 높은 건수 1건을 꺼낸 후 선점하는 것을 Atomic 하게 처리
• Lua로 구현
• 선점 TTL이 지나면 fraud:claimed:alert:{id} 키가 만료되어 자동 반환
• 선점 후 실제 처리 완료 시 아카이브(Stream)에 기록 후 claimed키 삭제

@Service
@RequiredArgsConstructor
public class FraudArchive {
    private final StringRedisTemplate redis;

    public String resolve(String alertMember, String analystId, String decision, String reason) {
        // 1) 선점 해제
        redis.delete("fraud:claimed:" + alertMember);

        // 2) 처리 이력 스트림에 기록(누가/언제/결과)
        Map<String, String> body = Map.of(
            "alert", alertMember,
            "analyst", analystId,
            "decision", decision,      // APPROVE / REJECT / ESCALATE ...
            "reason", reason,
            "ts", String.valueOf(System.currentTimeMillis())
        );
        return redis.opsForStream().add("fraud:stream:resolved", body);
    }
}

Outro

이번에는 1탄과 다르게 은행권이라는.. 도메인을 설정해서 범위를 좀 좁혀보고 이해에 도움이 되도록 노력해봤습니다. Redis는 ZSET, Lua 등 활용 방법을 넓혀주는 다양한 기술이 있어서 무궁무진하게 사용할 수 있는 NoSQL이라 생각합니다. 캐싱을 위해서만이 아닌 다른 기능으로 Redis 한번 사용해보시는 건 어떨까요. 다음엔 저도 프로젝트 리팩토링 해보며 어떻게 활용되는지 더 자세한 예제 코드와 함께 돌아오겠습니다.

Redis는 초저지연(1ms 미만)과 다양한 자료구조 지원 덕분에 캐시, 세션 관리, 분산 락, 작업 큐, 실시간 순위 집계 등 정말 다양한 패턴을 소화합니다. 저는 프로젝트에서 주로 사용자 토큰 관리 같은 특정 기간이 지나면 사라지는 데이터들을 저장하는 용도로 자주 사용했던 것 같습니다.

많은 개발자가 Redis를 선호하고 있는데 왜 redis를 사용할까요?

이번 글에서는 자주 쓰이는 Redis 활용 패턴 7가지 중 Cache-Aside, 분산 락, Rate Limiting에 대해 정리하려 합니다.
언제 이 패턴을 쓰고, 왜 필요한지, 어떻게 구현하는지를 예제 코드와 함께 살펴보겠습니다. 예제는 모두 Spring Boot + Spring Data Redis 사용을 기본으로 합니다.

1. Cache-Aside (Lazy Loading캐시)

DB를 원본 (Source of Truth)으로 두고, 조회할 때만 캐시에 적재하는 가장 표준적인 캐싱 전략

• 조회가 많은 서비스에 효과적입니다.
• 캐시 장애 시, DB가 살아있다면 기능 유지가 가능합니다. (성능은 저하)
• 반대로 쓰기(Write)가 잦고 강한 일관성이 필요하면 다른 패턴을 고려해야 합니다.

How?

기본 동작 흐름

1. 캐시에서 데이터 조회
2. 없으면 DB에서 조회
3. 조회 결과를 캐시에 저장 (TTL 포함)
4. 결과 반환

[의존성 + 설정]

implementation("org.springframework.boot:spring-boot-starter-data-redis")
// JPA
implementation("org.springframework.boot:spring-boot-starter-data-jpa")
implementation("com.fasterxml.jackson.core:jackson-databind")

// RedisConfig.java
@Configuration
public class RedisConfig {

  @Bean
  public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory cf) {
    RedisTemplate<String, Object> tpl = new RedisTemplate<>();
    tpl.setConnectionFactory(cf);
    var serializer = new GenericJackson2JsonRedisSerializer();
    tpl.setKeySerializer(new StringRedisSerializer());
    tpl.setValueSerializer(serializer);
    tpl.setHashKeySerializer(new StringRedisSerializer());
    tpl.setHashValueSerializer(serializer);
    tpl.afterPropertiesSet();
    return tpl;
  }

  @Bean
  public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory cf) {
    return new StringRedisTemplate(cf);
  }
}

[서비스 : Cache-Aside]

@Service
@RequiredArgsConstructor
public class UserService {
  private final UserRepository userRepository;
  private final RedisTemplate<String, Object> redisTemplate;
  private static final Duration TTL = Duration.ofMinutes(10);

  private String keyForUser(Long id) { return "user:" + id; }

  @Transactional(readOnly = true)
  public UserDto getUser(Long id) {
    String key = keyForUser(id);

    // 1) 캐시 조회
    Object cached = redisTemplate.opsForValue().get(key);
    if (cached != null) return (UserDto) cached;

    // 2) DB 조회
    User user = userRepository.findById(id)
        .orElseThrow(() -> new NotFoundException("user not found"));

    UserDto dto = UserDto.from(user);

    // 3) 캐시에 저장 + TTL
    redisTemplate.opsForValue().set(key, dto, TTL);

    // 4) 응답
    return dto;
  }
}

문제 상황

1) 캐시 스탬피드 (Thundering Herd) TTL이 동시에 만료되면 다수의 요청이 한번에 DB로 몰립니다. 이를 해결하기 위해 잠금 / Redission 락, Soft TTL + Hard TTL을 활용할 수 있습니다.

1. 잠금 (SET NE EX) / Redisson 락

// Redisson 사용 예
@Service
@RequiredArgsConstructor
public class UserServiceWithLock {
  private final UserRepository userRepository;
  private final RedisTemplate<String, Object> redisTemplate;
  private final RedissonClient redisson;
  private static final Duration TTL = Duration.ofMinutes(10);

  @Transactional(readOnly = true)
  public UserDto getUser(Long id) {
    String key = "user:" + id;

    Object cached = redisTemplate.opsForValue().get(key);
    if (cached != null) return (UserDto) cached;

    RLock lock = redisson.getLock("lock:" + key);
    boolean locked = false;
    try {
      locked = lock.tryLock(200, 5_000, TimeUnit.MILLISECONDS); // 대기 200ms, 보유 5s
      // 잠금 획득 후 다시 캐시 확인(중복 DB 접근 방지)
      cached = redisTemplate.opsForValue().get(key);
      if (cached != null) return (UserDto) cached;

      User user = userRepository.findById(id).orElseThrow();
      UserDto dto = UserDto.from(user);
      redisTemplate.opsForValue().set(key, dto, TTL);
      return dto;
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RuntimeException(e);
    } finally {
      if (locked) lock.unlock();
    }
  }
}

2. Soft TTL + Hard TTL (stale-while-revalidate)

• Hard TTL : 데이터가 너무 오래되면 무조건 재계산
• Soft TTL : 만료 임박 시, 첫 번째 요청만 백그라운드에서 갱신 시도, 나머지는 오래된 값 (stale) 즉시 반환 -> 체감 지연 시간 감소
• 구현은 캐시 값에 fetchedAt을 같이 저장한 후 Soft TTL 비동기 리프레시 수행

2) 캐시 관통 (Cache Penetration) 존재하지 않는 키를 계속 요청 : DB 성능 타격

1. Null 캐싱 : DB에 없으면 NULL도 짧은 TTL로 캐시

   redisTemplate.opsForValue().set(key, NullMarker.INSTANCE, Duration.ofSeconds(30));

2. Bloom Filter : 존재 가능성 검사 후 DB 접근 (RedisBloom 모듈 사용)

쓰기 시나리오 + 일관성

Cache-Aside에선 Write 시 캐시를 직접 갱신하지 않고 무효화를 하는 것이 일반적입니다.

1. (기본) Update/Delete -> 트랜잭션 성공 후 캐시 삭제

   @Transactional
   public void updateUserName(Long id, String name) {
   userRepository.updateName(id, name); // DB 정본
   redisTemplate.delete("user:" + id);  // 캐시 무효화
   }

2. (레이스 컨디션 방지) 지연 이중 삭제 요청 A가 DB를 갱신하기 직전, 요청 B가 캐시 미스를 내고 DB의 이전 값을 가져와 캐시할 수 있습니다. 방지책으로 캐시를 한번 더 삭제합니다.

@Transactional
public void updateUserName(Long id, String name) {
  userRepository.updateName(id, name);
  String key = "user:" + id;
  redisTemplate.delete(key);           // 1차 삭제
  // 약간 지연 후 2차 삭제 (스케줄러/메시지 큐/간단한 sleep 등)
  CompletableFuture.runAsync(() -> {
    try { Thread.sleep(200); } catch (InterruptedException ignored) {}
    redisTemplate.delete(key);         // 2차 삭제
  });
}

3. 이벤트 기반 무효화 더 견고하게 하기 위해 DB 변경 이벤트를 메시지 큐 (kafka 등)로 발행할 수 있습니다.

TTL 설계 방법

• 랜덤 지터(±10~20%)를 섞어 동시 만료 폭을 줄이기 (예: 600초 ± 60초)
• 데이터 성격별로 TTL 차등 설정(프로필 10m, 카운트 30s …)
• Hot Key는 TTL을 더 짧게 + Soft TTL 리프레시

@Cacheable vs 수동 구현

2. 분산 락 (Distributed Lock)

분산 락은 여러 서버 인스턴스가 동시에 접근하는 공유 자원을 한번에 하나의 작업만 처리하도록 제어하는 기술입니다. 쇼핑몰 재고 관리, 예약 시스템 좌석 배정, 결제 중복 방지 등에 사용됩니다.

** Redis 기반 분산 락 특징**

• 단일 Redis 인스턴스에서 가능 (SET NX EX)
• 네트워크로 연결된 다수의 서버 간 동기화 가능
• 락 만료 시간 (TTL)로 영구 락 방지

How?

동작 흐름

1. 락 획득 시도 -> 키 없으면 생성 (SET NX) + 만료 시간 설정 (EX)
2. 작업 수행
3. 락 해제 -> 락 소유자만 해제

[서비스 - 분산 락 SET NX EX]

@Service
@RequiredArgsConstructor
public class InventoryService {
  private final RedisTemplate<String, String> redisTemplate;
  private static final String LOCK_KEY = "lock:inventory";
  private static final Duration LOCK_TTL = Duration.ofSeconds(5);

  public boolean tryLock(String lockValue) {
    Boolean success = redisTemplate.opsForValue()
        .setIfAbsent(LOCK_KEY, lockValue, LOCK_TTL);
    return Boolean.TRUE.equals(success);
  }

  public void unlock(String lockValue) {
    String currentValue = redisTemplate.opsForValue().get(LOCK_KEY);
    if (lockValue.equals(currentValue)) {
      redisTemplate.delete(LOCK_KEY);
    }
  }

  public void processOrder(Long productId) {
    String lockValue = UUID.randomUUID().toString();
    if (!tryLock(lockValue)) throw new RuntimeException("다른 서버에서 처리 중입니다.");

    try {
      // 재고 차감 로직
    } finally {
      unlock(lockValue);
    }
  }
}

[Redisson 기반 구현]

@Service
@RequiredArgsConstructor
public class InventoryServiceWithRedisson {
  private final RedissonClient redisson;

  public void processOrder(Long productId) {
    RLock lock = redisson.getLock("lock:inventory:" + productId);

    boolean acquired = false;
    try {
      // 최대 200ms 대기, 락 점유 5초, Watchdog이 자동 연장
      acquired = lock.tryLock(200, 5, TimeUnit.SECONDS);
      if (!acquired) throw new RuntimeException("다른 서버에서 처리 중입니다.");

      // 재고 차감 로직
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RuntimeException(e);
    } finally {
      if (acquired) lock.unlock();
    }
  }
}

락 해제 시 반드시 락 소유자만 삭제해야한다. (다른 스레드 락 오염 방지)

문제 상황

1. TTL보다 작업이 오래 걸리는 경우

• TTL 만료 -> 다른 서버가 락 획득 -> 중복 실행 가능성
• 해결책
  • Redisson의 락 자동 연장 (Watchdog)기능 사용
  • TTL을 충분히 길게 설정, 과도하게 길면 데드락 가능성이 있으니..적당히...

2. Redis 장애 시

• 단일 노드 다운 -> 락 무효화
• 해결책 : Redlock 알고리즘 (여러 노드 과반수 동의 시 락 획득)
  • 네트워크 지연, 시계 드리프트 문제로 논쟁이 많음
  • 단일 인스턴스 + 고가용성 구성이 일반적

설계 고려사항

• 락 키 설계 : lock:{resourceId}
• 락 소유자 식별 : UUID 등 고유값 저장
• TTL 설정 시 지터 적용 가능
• 장기 작업은 락 분할 고려 (큰 작업 -> 작은 청크로 나누기_

3. Rate Limiting

Rate Limiting은 일정 시간 동안 허용되는 요청 횟수를 제한하는 기법이다. 서버 자원 보호, API 남용 방지, 보안 강화 등에 활용된다.

효도란 무엇일까요. 그건 임영웅 콘서트 예매입니다. 물론 278469번째는 효도를 할 수 없습니다.

이런 상황에서 Rate Limiting이란 기법을 사용할 수 있습니다. 수 많은 사람이 콘서트 예매를 위해 몰린다 생각하면 백엔드 개발자는 머리가 아플 수 밖에 없습니다. 로그인 시도 제한, 구매 API 초당 호출 수 제한, SMS 인증 요청 제한 등 모든 과정의 API를 얼마나 제한해야할지를 고민해야합니다.

Redis 기반 Rate Limiting 특징

• 초저지연 : 메모리 기반이라 카운트 처리 속도가 빠름
• *분산 환경 대응 : 여러 서버에서 동시에 접근해도 Redis 단일 키로 제어 가능
• 다양한 구현 방식 : Fixed Window, Sliding Window, Token Bucket

구현 방식

1. Fixed Window

• 일정 시간 간격 (예: 1분) 내 요청 횟수 제한
• 간단하지만 시간 경계에서 순간 폭주 가능

  // 예: user:login:12345 → 60초 동안 5회 제한
  Boolean exists = redisTemplate.hasKey(key);
  Long count = redisTemplate.opsForValue().increment(key);
  if (Boolean.FALSE.equals(exists)) {
    redisTemplate.expire(key, Duration.ofSeconds(60));
  }
  if (count > 5) {
    throw new RuntimeException("요청 횟수 초과");
  }

2. Sliding Widnow (Sorted Set)

• 요청 타임 스탬프를 기록하고, 윈도우 범위 밖 데이터 제거
• 시간 경계 문제 해소 : 어느 시점이든 직전 N초 기준으로 계산

public boolean isAllowed(String userId) {
    String key = "rate:" + userId;
    long now = System.currentTimeMillis();
    long windowMillis = 60_000;
    long limit = 5;

    ZSetOperations<String, String> zSetOps = redisTemplate.opsForZSet();

    // 현재 요청 시간 기록
    zSetOps.add(key, String.valueOf(now), now);

    // 윈도우 밖 데이터 삭제
    zSetOps.removeRangeByScore(key, 0, now - windowMillis);

    // 현재 윈도우 요청 수 확인
    Long count = zSetOps.zCard(key);
    redisTemplate.expire(key, Duration.ofMillis(windowMillis));

    return count <= limit;
}

3. Token Bucket

• 버킷에 토큰이 일정 속도로 채워지고, 요청 시 토큰 사용
• 부드러운 속도 제어, burst 허용
• Redis 스크립트 (Lua)로 원자적 구현이 대다수

문제 상황

1. 시간 경계 폭주 (Fixed Window)

• 경계 시점에 요청, 직후에도 요청하면 2배 처리 가능
• 해결 : 다른 방법 사용 (Sliding Window, Token Bucket)

2. 원자성 문제

• INCR + EXPIRE를 따로 호출하면 중간에 장애 시 TTL 미설정
• 해결 : SET key value NX EX 또는 Lua 스크립트로 원자화

3. 대규모 사용자 처리

• 많은 유저가 동시 호출 시 Redis 키가 급증
• 해결 : TTL로 키 자동 삭제, 샤딩 또는 Cluster 분산

설계 고려사항

• 키 패턴 : rate:{userId}:{endPoint} 또는 rate:{IP}
• TTL은 제한 시간과 동일하게 설정
• 제한 값과 시간은 환경 변수나 DB에서 관리 -> 실시간 조정 가능
• 모니터링 : 제한 횟수, 차단 횟수, 차단된 사용자 리스트 추적 등

Outro

그저 토큰 저장소로 사용했던 Redis가 서비스를 고도화할수록 얼마나 다양한 역할을 할 수 있는지 살펴봤습니다. 다음 글에서는 세션/토큰 스토어, 작업 큐, Pub/Sub, 리더보드 등 이번에 다루지 못한 나머지 패턴도 살펴보겠습니다.

이번에도 GPT한테 썸네일을 부탁했는데 이분탐색과는 다르게 4+2=9가 되는 아쉬운 모습을 보여주네요. 이미지 생성은 여전히 어려운 과제인가봐요...👽

개발을 하면서 주식 시장에서 특정 기간의 거래량 합계를 구하기, 게임 속 캐릭터의 체력이나 경험치 등 누적되는 값을 관리해야 하는 이벤트 등 데이터의 특정 구간을 다루는 일은 생각보다 자주 있습니다.

이를 쉽게 구하는 방법은 단순 for문 사용입니다. 하지만 매번 N번을 M명의 사용자에게 시도해야한다면 당연히 성능이 좋지 않습니다. 이를 해결하기 위해 세그먼트 트리를 사용할 수 있습니다.

세그먼트 트리는 구간(Segment)에 대한 정보를 트리(Tree)에 저장하여, 특정 구간의 합이나 최댓값 등을 매우 빠르게 찾아내는 자료구조

이전에 작성한 펜윅 트리보다 포괄적으로 사용되기 때문에 세그먼트 트리 먼저 작성해야 했지만... 지금이라도 포스팅해보겠습니다.

기본 구조

세그먼트 트리는 개념적으로는 트리지만 실제 구현할 때는 배열을 사용합니다. 위 이미지는 세그먼트 트리가 배열에 어떻게 저장되는지를 보여줍니다.

트리 구조

• 루트 노드 (1번)은 배열 전체의 구간과 그 합(36)을 담당합니다.
• 부모 노드는 자신의 구간을 절반으로 나누어 두 자식 노드에게 물려줍니다. (분할 정복)
• 이 과정은 더 이상 나눌 수 없는, 즉 원소 1개짜리 구간에 도달할 때 까지 반복됩니다. 트리의 가장 아래에 있는 노드 (리프 노드)들이 원본 배열의 각 원소가 됩니다.

배열 구조

• 이미지의 보라색 배열이 이 트리 구조를 1차원 배열에 옮겨 담은 실제 구현 모습입니다.
• 트리를 배열로 표현하기 위해 인덱스 규칙을 사용합니다.
  • 부모 노드가 배열 n번 인덱스에 있다면
  • 왼쪽 자식은 2 * n번 인덱스
  • 오른쪽 자식은 2 * n + 1번 인덱스

이렇게 배열로 표현하여 특정 노드의 부모나 자식을 복잡한 연결 과정 없이 인덱스 계산으로 빠르게 찾아가게 됩니다.

핵심 동작 원리 및 구현

트리 생성 (init)

우선, 주어진 배열을 바탕으로 각 구간의 합을 미리 계산해 세그먼트 트리 배열에 저장해야 합니다. 이 과정은 재귀를 이용해 구현합니다.

💡 Logic

1. 탐색은 루트 노드 (인덱스 1)에서 시작하며, 루트는 전체 배열 구간 (0~N-1)을 담당
2. 현재 노드가 담당하는 구간을 자식들에게 절반씩 나눠주며 재귀적으로 더 깊이 탐색 (분할)
3. 더 이상 나눌 수 없는, 즉 원소가 1개인 리프 노드에 도달하면, 해당 배열의 원소를 트리 노드에 저장
4. 재귀 호출이 끝나고 돌아오면서, 자식 노드들의 값을 합쳐 현재 노드의 값을 계산 (정복)

import java.util.Arrays;

public class SegmentTree {

    long[] arr;  // 원본 배열
    long[] tree; // 세그먼트 트리 배열

    public SegmentTree(long[] arr) {
        this.arr = arr;
        this.tree = new long[arr.length * 4];

        // 트리 생성(초기화) 메서드 호출
        // 시작: 루트 노드(인덱스 1), 담당 구간: 배열 전체(0 ~ N-1)
        init(0, arr.length - 1, 1);
    }

    /**
     * 세그먼트 트리 초기화 메서드
     * @param start 현재 노드가 담당하는 원본 배열의 시작 인덱스
     * @param end   현재 노드가 담당하는 원본 배열의 끝 인덱스
     * @param node  현재 세그먼트 트리 노드의 인덱스
     * @return      생성된 노드의 값 (자식에게 값을 물려주기 위함)
     */
    private long init(int start, int end, int node) {
        // 1. 리프 노드일 경우 (더 이상 분할할 수 없을 때)
        if (start == end) {
            // 원본 배열의 값을 트리 노드에 저장
            return this.tree[node] = this.arr[start];
        }

        // 2. 내부 노드일 경우 (분할이 가능할 때)
        int mid = (start + end) / 2; // 구간을 나눌 중간 지점

        // 왼쪽 자식(node*2)과 오른쪽 자식(node*2+1)을 재귀적으로 호출
        long leftChildValue = init(start, mid, node * 2);
        long rightChildValue = init(mid + 1, end, node * 2 + 1);

        // 3. 자식들의 합을 현재 노드에 저장 (정복)
        return this.tree[node] = leftChildValue + rightChildValue;
    }
}

• tree 배열 크기를 4*(원본 배열 크기)로 상정한 이유

  • 세그먼트 트리는 자식에게 절반씩 나눠주기 때문에 완전 이진 트리형태로 구현
  • 트리의 높이를 h라 할 때, 필요 노드 개수는 N에 따라 다름
  • n이 2의 거듭제곱인 경우 : h = log2(n) + 1, 노드 수 = 2 * (n-1)
  • n이 2의 거듭제곱이 아닌 경우 : 완전 이진 트리 형태 유지를 위해 n 보다 큰 가장 가까운 2의 거듭제곱 까지 필요
    • n = 6이라면 2^2 < n < 2^3 이기 때문에 8이 필요
    • 이를 방지하기 위해 4*n으로 넉넉하게 설정

• 나머지는 모두 기존 logic을 따라가는 코드

구간 합 구하기 (query)

초기화가 완료된 세그먼트 트리를 이용해 특정 구간의 합을 구하는 메서드 입니다.

💡 Logic

1. (No Overlap) 현재 노드 구간이 내가 찾는 구간과 전혀 겹치지 않으면, 합계에 영향을 주지 않는 값(0)을 반환합니다.
2. (Full Overlap) 현재 노드 구간이 내가 찾는 구간 안에 완전히 포함되면, 이 노드에 저장된 값을 바로 반환합니다. 더 깊이 탐색할 필요가 없습니다.
3. (Partial Overlap) 일부만 겹치면, 왼쪽 자식과 오른쪽 자식에게 각각 탐색을 위임하고, 그 결과들을 더해서 반환합니다.

public class SegmentTree {

    // ....

    /**
     * 구간 합을 구하는 메서드
     * @param left  구하고자 하는 구간의 시작 인덱스
     * @param right 구하고자 하는 구간의 끝 인덱스
     * @return      구간의 합
     */
    public long query(int left, int right) {
        // 내부 재귀 함수를 호출하여 결과를 반환
        return query(0, arr.length - 1, 1, left, right);
    }

    private long query(int start, int end, int node, int left, int right) {
        // Case 1: No Overlap (찾는 구간과 현재 노드 구간이 겹치지 않음)
        if (left > end || right < start) {
            return 0;
        }

        // Case 2: Full Overlap (찾는 구간이 현재 노드 구간을 완전히 포함)
        if (left <= start && end <= right) {
            return this.tree[node];
        }

        // Case 3: Partial Overlap (일부만 겹침)
        // 왼쪽 자식과 오른쪽 자식 모두 확인 후 그 합을 반환
        int mid = (start + end) / 2;
        long leftSum = query(start, mid, node * 2, left, right);
        long rightSum = query(mid + 1, end, node * 2 + 1, left, right);
        return leftSum + rightSum;
    }
}

• if (left > end || right < start) : 찾으려는 구간 [left, right] 가 현재 노드가 담당하는 [start, end]를 완전히 벗어난 경우. 더 이상 볼 필요가 없기에 0을 반환
• if (left <= start && end <= right) : 현재 노드 구간 [start, end]가 찾으려는 구간 [left, right] 안에 딱 들어오는 경우. 이 노드 (tree[node])에는 해당 구간의 합이 이미 계산되어 있기에 그 값을 바로 사용.
• return leftSum + right Sum : 겹치지만 완전히 들어오진 않는 경우. 어쩔 수 없이 왼쪽 자식과 오른쪽 자식 모두에게 겹치는 부분이 있는지 확인 후 합쳐서 return

값 변경하기 (update)

배열의 특정 원소 값이 바뀌면 그 값을 포함하는 모든 구간의 합이 영향을 받습니다. 해당 원소를 포함하는 리프 노드부터 루트 노드까지의 경로에 있는 모든 노드 값을 갱신해야 합니다.

💡 Logic

1. 값을 변경할 원소의 인덱스와 값의 차이(diff = newValue - oldValue)를 계산
2. 루트부터 시작해 인덱스를 포함하는 노드들을 따라 리프 노드까지 탐색
3. 탐색 경로에 있는 모든 노드들의 값에 diff만큼 더함. 자식들의 값을 다시 계산할 필요 없이, 차이값만 반영 (이 반영을 늦추는 방식이 lazy propagation)

/**
* 특정 원소의 값을 변경하는 외부 호출용 메서드
* @param index     값을 변경할 원소의 인덱스
* @param newValue  새로운 값
*/
public void update(int index, long newValue) {
   long diff = newValue - this.arr[index]; // 기존 값과 새로운 값의 차이
   this.arr[index] = newValue; // 원본 배열의 값도 갱신해준다.
   update(0, arr.length - 1, 1, index, diff);
}

private void update(int start, int end, int node, int index, long diff) {
    // Case 1: 변경할 index가 현재 노드 구간과 관련 없는 경우
    if (index < start || index > end) {
       return;
    }

    // Case 2: 현재 노드 구간이 변경할 index를 포함하는 경우
    // -> 현재 노드 값에 차이(diff)를 반영한다.
    this.tree[node] += diff;

    // 리프 노드가 아니라면, 자식 노드도 재귀적으로 찾아 내려간다.
    if (start != end) {
        int mid = (start + end) / 2;
        update(start, mid, node * 2, index, diff);
        update(mid + 1, end, node * 2 + 1, index, diff);
    }
}

• if (index < start || index > end) : 변경하려는 인덱스가 현재 노드가 담당하는 구간에 포함되지 않으면 이 노드와 그 자식들은 값의 변화와 아무 관련 X. 더 깊이 탐색하지 않고 즉시 return
• if (start != end) : 현재 노드가 리프 노드가 아닌 경우에만 재귀 호출을 계속해 경로상의 모든 노드 갱신

[전체 코드]

import java.util.Arrays;

public class SegmentTree {

    long[] arr;
    long[] tree;

    public SegmentTree(long[] arr) {
        this.arr = arr;
        this.tree = new long[arr.length * 4];
        init(0, arr.length - 1, 1);
    }

    private long init(int start, int end, int node) {
        if (start == end) {
            return this.tree[node] = this.arr[start];
        }
        int mid = (start + end) / 2;
        return this.tree[node] = init(start, mid, node * 2) + init(mid + 1, end, node * 2 + 1);
    }

    public long query(int left, int right) {
        return query(0, arr.length - 1, 1, left, right);
    }

    private long query(int start, int end, int node, int left, int right) {
        if (left > end || right < start) {
            return 0;
        }
        if (left <= start && end <= right) {
            return this.tree[node];
        }
        int mid = (start + end) / 2;
        return query(start, mid, node * 2, left, right) + query(mid + 1, end, node * 2 + 1, left, right);
    }

    public void update(int index, long newValue) {
        long diff = newValue - this.arr[index];
        this.arr[index] = newValue;
        update(0, arr.length - 1, 1, index, diff);
    }

    private void update(int start, int end, int node, int index, long diff) {
        if (index < start || index > end) {
            return;
        }
        this.tree[node] += diff;
        if (start != end) {
            int mid = (start + end) / 2;
            update(start, mid, node * 2, index, diff);
            update(mid + 1, end, node * 2 + 1, index, diff);
        }
    }

시간 복잡도

세그먼트 트리가 빠른 이유는 인덱스 규칙을 활용한 트리 개념 활용에 있습니다.

• 트리 생성 (init) 시 모든 노드를 한 번씩 방문하기 때문에 시간 복잡도가 O(N) 입니다.
• 구간 합 & 값 변경 (query & update) 는 모두 트리의 높이에 비례하는 시간이 걸립니다 (리프 to 루트). 트리의 높이는 약 logN 이므로 두 연산 모두 O(logN)이 걸립니다.

백준 2243. 사탕상자

• 이분 탐색 문제를 풀기 위해 접한 문제라 당연히 이분 탐색으로 먼저 접근했습니다.
• 단순 이분 탐색으로 풀다보니 모든 구간에 대한 누적 개수를 구하기 어려웠습니다.
• 해당 문제의 핵심은 구간별 개수를 유지하면서 순위 (k번째 작은 원소) 조회를 지원하는 알고리즘입니다.

구간 합 문제라 생각해서 세그먼트 트리가 먼저 생각났으나 검색해보니 펜윅 트리가 있어서 해당 알고리즘에 대해 알게됐습니다.

기본 개념

펜윅 트리는 1994년 논문이 발표된 비교적 최신 알고리즘이다. PS에서는 세그먼트 트리로 풀 수 있는 문제 중 메모리를 절약할 수 있는 장점이 있어서 사용된다.

누적합 (prefix sum)

• 누적합 알고리즘은 1 ~ N 까지의 합 배열을 따로 만들어 구간합을 O(1)로 구할 수 있습니다.
• 1 ~ N 배열의 값이 변경(update)되면 누적합 배열을 전부 다시 계산해야하므로 O(N)의 복잡도를 갖게됩니다.

펜윅 트리 (fenwick tree)

• 펜윅 트리는 누적합에서 변경(update)을 더 빠르게 하기 위해 사용합니다.
• 구간합, 배열값 변경 모두 O(logN)의 복잡도를 갖습니다.
• 세그먼트 트리 구성을 위해 약 4N의 메모리가 필요한 것과 달리 기존 배열 크기인 N만큼의 메모리만 필요합니다.
• 세그먼트 트리보다 코드가 매우 간단합니다.

구현

⚙️ Setting

• 기존 배열 : arr = [1,2,3,4,...,16]
• 펜윅 트리 : tree = int[16]
• 핵심 연산 :
  • 구간 합 (query) : 1부터 특정 인덱스까지의 구간 합을 구합니다.
  • 값 변경 (update) : 특정 인덱스의 값을 변경 후 연관된 구간 합을 갱신합니다.

핵심 아이디어 : Range of Responsibility

펜윅 트리는 겉보기엔 단순 1차원 배열이지만, 내부적으로는 이미지처럼 계층적인 트리 구조 규칙을 따릅니다. tree 배열의 각 칸(노드)은 원본 배열(arr)의 특정 구간에 대한 합을 "책임"집니다.

• tree[8]은 arr[1]부터 arr[8]까지 8개 원소의 합을 책임짐.
• tree[12]는 arr[9]부터 arr[12]까지 4개 원소의 합을 책임짐.
• tree[7]은 arr[7]부터 arr[7]까지 1개 원소 (=자기 자신)값만 책임짐.

최하위 비트(LSB)가 이 Range of Responsibility를 정합니다.

최하위 비트 (LSB, RSB)

최하위 비트 (Least Significant Bit, LSB)는 어떤 숫자를 이진수로 표현했을 때 가장 오른쪽에 있는 1의 값을 의미한다. 1의 위치가 아닌, 1이 나타내는 실제 값임에 주의한다.

• 12 → 1100₍₂₎ → 가장 오른쪽의 1은 뒤에서 세 번째 자리에 있습니다. (LSB = 4)

• 7 → 0111₍₂₎ → 가장 오른쪽의 1은 맨 뒷자리에 있습니다. (LSB = 1)

• 8 → 1000₍₂₎ → 가장 오른쪽의 1은 뒤에서 네 번째 자리에 있습니다. (LSB = 8)

바로 이 LSB값이 각 tree 인덱스가 책임지는 원소의 개수가 됩니다. 이 LSB는 다음과 같은 비트 연산으로 간단하게 구할 수 있습니다.

LSB = i & -i

현재 값과 2의 보수의 AND 연산만으로 LSB값을 구할 수 있습니다.

[예시]
// 12의 LSB 구하기
12  = 00001100
-12 = 11110100  (2의 보수 표현)
-----------------
&     00000100  (결과는 4)

// 7의 LSB 구하기
7   = 00000111
-7  = 11111001
-----------------
&     00000001  (결과는 1)

이제 tree[i]가 몇 개의 원소를 책임지는지 알게 되었습니다. 이를 바탕으로 update와 query를 구현할 수 있습니다.

update : 값 변경 후 갱신

원본 배열 arr[i] 값을 변경 시 arr[i]를 포함하는 모든 구간 합들을 찾아 값을 갱신해야합니다. tree에서 i번 노드의 변경으로 영향을 받는 모든 노드를 수정해야합니다. 이 영향을 받는 노드 또한 LSB를 통해 구할 수 있습니다. 특정 인덱스 i에서 시작하여 자신의 LSB값을 더해주면 다음으로 갱신할 상위 노드의 인덱스로 이동할 수 있습니다.

다음 인덱스 = i + (i & -i)

[예제] update(3, 5) : arr[3] += 5

1. i=3: tree[3]에 5를 더하기 다음 인덱스 : 3 + (3 & -3) = 4

2. i=4: tree[4]에 5를 더하기 다음 인덱스 : 4 + (4 & -4) = 8

3. i=8: tree[8]에 5를 더하기 다음 인덱스 : 8 + (8 & -8) = 16

4. i = 16: tree[16]에 5를 더하기 다음 인덱스 : 16 + (16 & -16) = 32 (배열 범위 밖이라 종료)

이 예제를 통해 arr[3]이 변하면 tree[3], tree[4], tree[8], tree[16] 순서대로 전파되는 것을 알 수 있습니다.

이를 코드로 구현하면 다음과 같습니다.

public static void update(int i, long val) {
   // i가 배열 크기 N을 넘지 않을 때까지 LSB를 더해가면서 val을 더함
   while (i <= N) {
       tree[i] += val;
        i += lsb(i);
   }
}

query : 1부터 N까지의 합 구하기

1부터 i까지의 누적 합을 구하는 과정은 update와 반대입니다. i에서 시작해, 자신의 LSB값을 빼주면서 0이 될 때까지 만나는 모든 노드의 값을 더하면 됩니다. 각 노드들이 책임지는 구간들은 서로 겹치지 않기 때문에, 이들을 더하면 정확히 1부터 i까지의 합이 완성됩니다.

다음 인덱스 = i - (i & -i)

[예제] query(13) : arr[1] + arr[2] + ... + arr[13]

1. i = 13: sum에 tree[13]을 더하기. (tree[13]은 arr[13]의 합) 다음 인덱스: 13 - (13 & -13) = 13 - 1 = 12

2. i = 12: sum에 tree[12]를 더하기. (tree[12]는 arr[9]~arr[12]의 합) 다음 인덱스: 12 - (12 & -12) = 12 - 4 = 8

3. i = 8: sum에 tree[8]을 더하기. (tree[8]은 arr[1]~arr[8]의 합) 다음 인덱스: 8 - (8 & -8) = 8 - 8 = 0 (0이 되었으므로 종료)

이를 코드로 구현하면 다음과 같습니다.

public static long query(int i) {
    long result = 0;
    // i가 0보다 클 때까지 LSB를 빼가면서 tree 값을 더함
    while (i > 0) {
       result += tree[i];
        i -= lsb(i);
    }
    return result;
}

세그먼트 트리와 같이 부분 합을 구하기 위해서는 query 연산을 두번 해야합니다. 예를 들어 3~7 구간합을 구하기 위해서는 query(7) - query(2)을 해야합니다

사탕 상자 풀이

• B, C가 주어질 때 : C개의 B 맛 사탕 넣기 or 꺼내기. 이는 B라는 인덱스의 값을 C만큼 변화시키는 것 = query(B,C)
• A가 주어질 때 : A번째로 맛있는 사탕 꺼내기. 누적 합이 A 이상이 되는 최초의 맛(인덱스)가 무엇인가요?`
  • 기존의 query만으로는 부족
  • 해당 조건을 만족하는 값을 이분 탐색을 통해 해결해야함.
  • 펜윅 트리를 응용한 k번째 원소 찾기 기법
  • 가장 큰 2의 거듭제곱부터 시작해서, 현재 인덱스의 사탕 개수를 확인하며 이진 탐색처럼 아래로 내려오는 방식으로 원하는 순위의 사탕을 찾기

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.StringTokenizer;

public class Main {

    static final int MAX = 1_000_000;
    static int[] fenwick = new int[MAX+1];

    static void update(int idx, int val) {
        while(idx <= MAX) {
            fenwick[idx] += val;
            idx += idx & -idx;
        }
    }

    static int find(int rank) {
        int idx = 0;
        for(int i = (int)Math.pow(2, 20); i > 0; i /= 2) {
            if(idx + i <= MAX && fenwick[idx+i] < rank) {
                rank -= fenwick[idx + i];
                idx += i;
            }
        }
        return idx + 1;
    }
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringTokenizer st;
        StringBuilder sb = new StringBuilder();
        int N = Integer.parseInt(br.readLine());
        for(int i = 0; i < N; i++) {
            st = new StringTokenizer(br.readLine());
            int cmd = Integer.parseInt(st.nextToken());
            if (cmd == 1) {
                int rank = Integer.parseInt(st.nextToken());
                int taste = find(rank);
                sb.append(taste).append("\n");
                update(taste, -1);
            }else {
                int taste = Integer.parseInt(st.nextToken());
                int cnt = Integer.parseInt(st.nextToken());
                update(taste, cnt);
            }
        }
        System.out.println(sb);
    }
}

• 가중치가 없는 경우 선택

• 가중치가 0 / 1 로 이루어져 있을 때 선택

  • 가중치가 있지만 0 / 1 이라면 다익스트라가 아닌 BFS로 풀이가 가능

  • 다익스트라 시간 복잡도

    • 이진 힙 사용 (pq) = $O(ElogV)$
    • 피보나치 힙 사용 = $O( E+VlogV)$

  • 0-1 BFS

    • Queue 가 아닌, Deque 을 활용

    • 가중치가 0 이면 앞에 삽입, 1 이면 뒤에 삽입

      for all v in verticces:
        dist[v] = inf
      dist[start] <- 0
      deque d
      d.push_front(start)
      
      while d.empty() == false:
        vertex = get front element and pop as in BFS
        for all edges e of form(vertex, u):
            if travelling e relaxes distance to u:
                relax dist[u]
                if e.weight = 1:
                    d.push_back(u)
                else:
                    d.push_front(u)

    • 시간 복잡도 $O(E + V)$

Dijkstra

• 가중치에 음수가 없는 경우 선택
  • 그리디 기반인데, 음수 가중치가 있으면 그리디 논리에 모순이 발생
  • 사이클이 없으면 구현이 가능할 수 있지만 시간 복잡도가 큼 $O(V^2)$
  • 음수의 사이클이 존재할 때는 무한 루프 문제 발생
• 그리디 + DP 기반 ⇒ 우선순위 큐 사용

Bellman Ford

• 가중치에 음수가 있는 경우 선택
  • 벨만 포드도 음수 사이클이 있다면 최단 경로 구할 수 없고 “탐지”는 가능
• 모든 간선을 반복적으로 탐색하면서 최단 경로를 점진적 업데이트하는 방식
• 간선 수 (E)가 적을 때 사용
• 시간복잡도 : $O(VE)$

Floyd-Warshall

• 가중치에 음수가 있고 모든 정점 쌍의 최단 경로를 구할 때
  • 음수 사이클 X ⇒ 경로 길이 무한으로 수렴
• DP 기반의 알고리즘
• 시간복잡도 : $O(V^3)$

최소 스패닝 트리 Minimum Spanning Tree

우선 스패닝 트리가 무엇일까?

스패닝 트리 Spanning Tree

• 정의 = “n개의 정점으로 이루어진 무향 그래프에서 n개의 정점과 n-1개의 간선을 이루어진 트리”
• 1번 / 2번 / 3번 = 모든 노드를 지나는 경로 중 골라봄. (여러 스패닝 트리 존재 가능)
• 사이클이 없음
• 정점의 개수 n 일 때? 스패닝 트리의 간선 개수는? 당연히 n-1 개
• 사용하는 이유
  • 네트워크 설계 : 최소 케이블로 모든 컴퓨터 연결하는 네트워크 설계
  • 소셜 네트워크 분석 : 중요 노드 간의 연결 분석 후 군집 구조 단순화로 시각화
  • 데이터 압축 : 이미지 데이터를 그래프로 표현 후 스패닝 트리 생성해 연결된 주요 픽셀 정보만 보관하는 방식으로 데이터를 압축

최소 + 스패닝 트리 ➡️ 스패닝 트리 中 가장 작은 것 의미는 쉬운데 막상 코드 작성 시 어려움… 작성 방식은 크게 두 가지 ⇒ **KRUSKAL / PRIM

왜 완전탐색은 안될까?

• 최소 신장 트리를 완전 탐색으로 생각
  • 정점 30개, 간선 60개라 생각할 때 완전 탐색해서 MST 찾아보면?
  • ${60}C{29}$ ⇒ 터쳐버림

KRUSKAL

💡 [중요 정보]

• 구현 : 정렬 + 그리디 + 유니온 파인드

• 시간 복잡도 : $O(ElogE)$

• 간선이 적은 희소 그래프 에서 효율적

• 간선에 관심이 있음

• 💡아이디어💡 [1, 10, 2, 3, 4] 에서 3개를 뽑아 합이 최소가 되는 경우? ⇒ 정렬 (오름차순) ⇒ 앞쪽부터 3개 선택 : 그리디 ⇒ 근데 사이클이면 안되는데…. : 유니온 파인드

• 예제

  import java.util.*;
  

class Edge implements Comparable { int src, dest, weight;

public Edge(int src, int dest, int weight) {
    this.src = src;
    this.dest = dest;
    this.weight = weight;
}

@Override
public int compareTo(Edge other) {
    return this.weight - other.weight; // 가중치 오름차순 정렬
}

}

public class KruskalMST { static int[] parent;

// 유니온-파인드: Find 연산
public static int find(int node) {
    if (parent[node] == node) return node;
    return parent[node] = find(parent[node]); // 경로 압축
}

// 유니온-파인드: Union 연산
public static void union(int node1, int node2) {
    int root1 = find(node1);
    int root2 = find(node2);
    if (root1 != root2) {
        parent[root2] = root1; // 두 집합을 합침
    }
}

public static void main(String[] args) {
    int vertices = 5; // 정점 개수
    int edgesCount = 7; // 간선 개수

    // 간선 리스트 초기화
    List<Edge> edges = new ArrayList<>();
    edges.add(new Edge(0, 1, 1)); // A-B: 1
    edges.add(new Edge(0, 2, 3)); // A-C: 3
    edges.add(new Edge(1, 2, 3)); // B-C: 3
    edges.add(new Edge(1, 3, 6)); // B-D: 6
    edges.add(new Edge(2, 3, 4)); // C-D: 4
    edges.add(new Edge(2, 4, 2)); // C-E: 2
    edges.add(new Edge(3, 4, 5)); // D-E: 5

    // 유니온-파인드 초기화
    parent = new int[vertices];
    for (int i = 0; i < vertices; i++) {
        parent[i] = i; // 초기에는 각 정점이 자신이 부모
    }

    // 간선 가중치 기준 정렬
    Collections.sort(edges);

    // 크루스칼 알고리즘 수행
    List<Edge> mst = new ArrayList<>();
    int mstWeight = 0;

    for (Edge edge : edges) {
        // 사이클이 생기지 않으면 간선 선택
        if (find(edge.src) != find(edge.dest)) {
            union(edge.src, edge.dest);
            mst.add(edge);
            mstWeight += edge.weight;
        }
    }

    // 결과 출력
    System.out.println("Minimum Spanning Tree:");
    for (Edge edge : mst) {
        System.out.printf("Edge: %d-%d, Weight: %d\n", edge.src, edge.dest, edge.weight);
    }
    System.out.println("Total Weight: " + mstWeight);
}

}

```java
Minimum Spanning Tree:
Edge: 0-1, Weight: 1
Edge: 2-4, Weight: 2
Edge: 0-2, Weight: 3
Edge: 2-3, Weight: 4
Total Weight: 10

PRIM

💡 [중요 정보]

• 구현 : 그리디 + 우선순위 큐

• 시간 복잡도 : $O(ElogV)$

• 간선이 많은 밀집 그래프 에서 효율적

• 노드(정점)에 관심이 있음

• 💡 아이디어 💡

  1. 시작 정점(0번) 선택
  2. 연결된 간선 중에서 가중치가 가장 작은 간선 선택: 그리디 + 우선순위 큐
  3. 새로운 정점 추가 → 해당 정점에서 연결된 간선 추가
  4. 이미 방문한 정점으로의 간선은 무시: 방문 체크 (사이클 방지)
  5. 모든 정점이 연결될 때까지 반복

• 예제

  import java.util.*;
  

class PrimMST { static class Edge implements Comparable { int dest, weight;

    public Edge(int dest, int weight) {
        this.dest = dest;
        this.weight = weight;
    }

    @Override
    public int compareTo(Edge other) {
        return this.weight - other.weight; // 가중치 기준 오름차순 정렬
    }
}

public static void main(String[] args) {
    int vertices = 5; // 정점의 개수

    // 그래프를 인접 리스트로 초기화
    List<List<Edge>> graph = new ArrayList<>();
    for (int i = 0; i < vertices; i++) {
        graph.add(new ArrayList<>());
    }

    // 간선 추가 (무방향 그래프)
    graph.get(0).add(new Edge(1, 1)); // A-B: 1
    graph.get(1).add(new Edge(0, 1));
    graph.get(0).add(new Edge(2, 3)); // A-C: 3
    graph.get(2).add(new Edge(0, 3));
    graph.get(1).add(new Edge(2, 3)); // B-C: 3
    graph.get(2).add(new Edge(1, 3));
    graph.get(1).add(new Edge(3, 6)); // B-D: 6
    graph.get(3).add(new Edge(1, 6));
    graph.get(2).add(new Edge(3, 4)); // C-D: 4
    graph.get(3).add(new Edge(2, 4));
    graph.get(2).add(new Edge(4, 2)); // C-E: 2
    graph.get(4).add(new Edge(2, 2));
    graph.get(3).add(new Edge(4, 5)); // D-E: 5
    graph.get(4).add(new Edge(3, 5));

    // 우선순위 큐를 사용하여 최소 가중치 간선을 선택
    PriorityQueue<Edge> pq = new PriorityQueue<>();

    boolean[] visited = new boolean[vertices]; // 방문 여부 체크
    int mstWeight = 0; // MST 가중치 합
    List<String> mstEdges = new ArrayList<>(); // MST 간선 리스트

    // 시작 정점 (0번 정점)
    pq.add(new Edge(0, 0));

    while (!pq.isEmpty()) {
        Edge current = pq.poll(); // 가장 가중치가 작은 간선 선택

        if (visited[current.dest]) continue; // 이미 방문한 정점이면 스킵

        visited[current.dest] = true; // 정점 방문 처리
        mstWeight += current.weight; // MST 가중치 추가

        // 간선을 기록 (시작 정점이 있는 경우 출력)
        if (current.weight != 0) {
            mstEdges.add(String.format("Edge: %d, Weight: %d", current.dest, current.weight));
        }

        // 현재 정점에 연결된 간선을 큐에 추가
        for (Edge neighbor : graph.get(current.dest)) {
            if (!visited[neighbor.dest]) {
                pq.add(neighbor);
            }
        }
    }

    // 결과 출력
    System.out.println("Minimum Spanning Tree:");
    for (String edge : mstEdges) {
        System.out.println(edge);
    }
    System.out.println("Total Weight: " + mstWeight);
}

}

```

최소 경로(비용) 알고리즘 선택 가이드

출처

https://4legs-study.tistory.com/111

https://lotuslee.tistory.com/46

또! 다시 설계 이야기를 해볼까 합니다. 구현보다 어려운 설계, 언제쯤 알잘딱 잘하게 될지 참 궁금합니다. DB Deep은 기업 요구사항에 맞춰 진행한 프로젝트라 우선 요구사항에 맞추는 것이 ..급했습니다. 그러다 오늘 Mongo DB를 조금 파보니 여지없이 설계에 실수가 있는 것 같더라구요. 제대로 알지 못하는 상태에서 사용한다면 티가 나버리는 것 같습니다. 그래서 한번 대대적으로 갈아엎기 전에 고민의 흐름을 기록해두겠습니다.

초기 채팅 저장 구조

당시에는 단순하게 "채팅 메시지를 시간 순서대로 저장하고 불러오자!" 라는 생각으로 Google Cloud Platform의 Firestore에 메시지를 저장했습니다. 저장 방식은 다음과 같은 구조입니다.

def save_chat_message(
    chat_room_id: str,
    sender_type: Literal["user", "ai", "system"],
    message_type: Literal["text", "sql", "chart", "insight"],
    content: str | dict
):
    db = get_firestore_client()
    doc_ref = db.collection("chat_messages").add({
        "chat_room_id": chat_room_id,
        "sender_type": sender_type,
        "type": message_type,
        "content": content,
        "timestamp": datetime.utcnow()
    })

    return doc_ref[1].id

def get_chat_messages(chat_room_id: str, limit: int = 100):
    db = get_firestore_client()
    query = (
        db.collection("chat_messages")
          .where("chat_room_id", "==", chat_room_id)
          .order_by("timestamp")
          .limit(limit)
    )
    return [doc.to_dict() for doc in query.stream()]

모든 메시지를 chat_messages라는 하나의 컬렉션에 저장하고, 필드값으로 구분만 했습니다. 그리고 특정 채팅방(chat_room_id)의 기록을 불러오는 형식이었습니다. 겉보기에는 괜찮아 보이지만, 시간이 지나고 데이터가 쌓이기 시작한다면 여러 한계가 드러날 구조라 생각합니다.

문제점

성능 저하 : 전부 하나에 모여있는 단일 컬렉션

• chat_messages 컬렉션이 모든 채팅방의 메시지를 다 담고 있다 보니, 수십만 건이 쌓일수록 하나의 쿼리만으로도 비용이 커질 수 있습니다.
• chat_room_id를 기준으로 filter를 걸어도 Firestore에서는 항상 index scan이 발생하기 때문에 컬렉션 자체가 커질수록 조회 성능은 떨어집니다.

구조적 불일지 : content 필드

• content 필드는 타입에 따라 str일 수도 있고 dict일 수도 있는 비정형 필드입니다.
• message_type이 text일 때 string이지만, chart인 경우 json 형태의 dict가 됩니다.
• 이를 처리하는 코드에서는 매번 타입 체크 + 조건 분기를 해야하기 때문에 전체 흐름을 파악하거나 유지보수에 극악의 효율을 보여주고 있습니다😥

확장 불가능한 구조

• 단일 컬렉션의 모든 단점을 포함하고 있기에 당연히 향후 메시지 저장 형식의 확장이 불가능합니다.
• 답변 평가, 피드백, 첨부파일, system message에 대한 처리 등의 기능을 추가할 때 구조적인 제약이 발생합니다
  • 실제로 사용자 피드백을 받아 성능 개선 당시 Mongo DB 설계 미스로 구현에 어려움을 겪었습니다.
• Firestore에서 제공하는 subcollection이나 nested document로 충분히 해결 가능합니다.

Mongo DB의 설계 방법

RDB는 정규화, MongoDB는 사용 패턴 (Workloads)이 설계의 중심

그렇다면 Mongo DB는 스키마 설계를 어떻게 해야할까요? 이번 프로젝트를 하며 느낀점은 Mongo DB는 RDB와는 데이터 모델링 자체에 대한 철학이 다른 것을 인지해야한다는 점 입니다.

워크로드 기반 설계

Mongo DB에서는 데이터를 어떻게 쓰고, 읽고, 업데이트 할 것인지 실제 사용 패턴을 중심으로 데이터 모델을 설계해야합니다.

• 어떤 데이터를 자주 함께 조회하는가?
• 읽기와 쓰기 중 어떤 작업이 더 많은가?
• 특정 문서의 크기가 계속 커질 가능성이 있는지?

이러한 설계의 차이는 RDB는 데이터 정합성과 무결성이 중요하고 Mongo DB는 가장 큰 특징인 데이터 중복 허용과 쿼리 효율성 우선시 하기 때문입니다.

모델링 시 고려사항

Mongo DB에서 문서 설계 시 다음과 같은 기준을 고려합니다.

• 한 번의 요청으로 필요한 데이터를 모두 가져올 수 있는지?
• 문서 크기가 16MB (Mongo DB의 단일 문서 크기)를 초과할 위험이 있는지?

이 기준을 통해 문서 간 관계를 Embedding으로 할 지 Referencing으로 할 지 결정됩니다.

{
  "userId": "abc123",
  "name": "홍길동",
  "posts": [
    {"title": "첫 글", "createdAt": "..."},
    {"title": "두 번째 글", "createdAt": "..."}
  ]
}

MongoDB로 사용자 게시글을 조회한 예시입니다. 자주 같이 조회된다면 예시와 같이 임베딩이 적절하지만 게시글 수가 수천 개 이상으로 커진다면 분리한 후 레퍼런싱 하는 것이 더 적절합니다. 예시를 통해 더 알아봅시다.

• 유저 프로필 + 설정

  {
  "userId": "123",
  "name": "철수",
  "settings": {
    "theme": "dark",
    "notifications": true
  }
  }

  name과 settings는 유저 정보를 볼 때 항상 함께 조회됩니다. 따라서 지금처럼 임베딩이 적절합니다.

• 유저 + 게시글

  {
  "userId": "123",
  "name": "철수"
  }

{
  "postId": "999",
  "userId": "123",
  "title": "MongoDB 잘 쓰는 법",
  "content": "..."
}

사용자에 대한 정보는 자주 변경되지 않습니다. 하지만 게시글은 개수가 많고 조회, 추가, 삭제가 잦습니다 따라서 유저 데이터를 조회할 때 게시글이 항상 가져올 필요가 없기 때문에 레퍼런싱하는 것이 적절합니다.

대안 설계 방향

위 내용을 바탕으로 Firestore의 구조적 장점을 최대한 살릴 수 있는 방향으로 구조를 고민해봤습니다.

1. 채팅방 중심 구조

기존 구조

• 모든 채팅방의 메시지를 동일 컬렉션에 넣어 쿼리 성능이 점점 악화
• 인덱스 구성해도 전체 컬렉션 단위 인덱스 스캔이 빈번한 기능이 많기 때문에 부담 여전

개선안: 채팅방 하위 컬렉션 분리

• Firestore의 Subcollection을 활용해 각 채팅방에 종속적인 메시지 컬렉션 분리
• 탐색 범위 감소 : 각 채팅방 별로 데이터를 쿼리, 필요한 메시지만 빠르게 조회 가능
• 문서 수 제한 회피 : 메시지 수가 늘어도 단일 컬렉션에 데이터가 몰리지 않음
• 확장 유연성 : 향후 채팅방 메타 데이터, 참여자 관리 등 구조 확장 용이

적용 방법

• MongoDB의 16MB 제한에 맞춰야함
• 채팅방 별 메시지 컬렉션 분리
  • chat_messages_{chat_room_id} 식으로 컬렉션을 나누거나 chat_room_id로 샤딩 키를 설정
• 채팅 메시지를 채팅방에 임베딩
  • 메시지가 많으면 16MB를 초과할 수 있기에 테스트 단계에서 고려해볼만한 듯?

2. 메시지 타입에 따른 구조 정리

기존 구조

• content 필드 하나에 모든 메시지 데이터를 넣음
  • text일 땐 string
  • sql, chart, insight일 땐 dict(JSON)
• 타입에 따라 구조가 다르기 때문에:
  • 매번 타입 체크 → 파싱 로직 복잡
  • 프론트엔드에서 메시지 렌더링 시 로직 분기 증가
  • 추후 로그 분석, 백업 등 구조화된 데이터 필요 시 어려움 발생

예: type == "chart" → content["x"], content["y"] 체크 후 시각화
type == "sql" → content["query"] 존재 여부 확인 후 실행 요청 등
→ 모든 타입마다 if-else 분기처리 필요

개선안: 메시지 타입별 content 스키마 수정

message_type에 따라 content 필드를 명확한 구조로 구분합니다.

예시)

• 텍스트

{
  "type": "text",
  "sender_type": "user",
  "content": "이번 쿼리의 문제점은 뭐야?",
  "timestamp": "2024-01-01T12:00:00Z"
}

• SQL 생성 응답

{
  "type": "sql",
  "sender_type": "ai",
  "content": {
    "query": "SELECT name FROM employees WHERE salary > 5000"
  },
  "timestamp": "2024-01-01T12:01:00Z"
}

• 차트 (그래프)

{
  "type": "chart",
  "sender_type": "ai",
  "content": {
    "chart_type": "bar",
    "x": ["부서 A", "부서 B", "부서 C"],
    "y": [12, 23, 5]
  },
  "timestamp": "2024-01-01T12:02:00Z"
}

• 인사이트 요약

{
  "type": "insight",
  "sender_type": "ai",
  "content": {
    "summary": "부서 B는 전체 대비 2배 이상의 매출을 기록했습니다.",
    "related_kpi": ["매출", "부서"]
  },
  "timestamp": "2024-01-01T12:03:00Z"
}

추가 고려사항

• 메시지 유형에 따라 content 필드의 구조를 union type으로 설계
• 각 메시지를 별도 도큐먼트로 분리하고 message_type별 하위 필드로?

예: text, sql, chart, insight 각각의 필드 구성이 다를 경우 → content 하나로 두기보단, 아예 각 필드를 독립적으로 선언하는 것도 고려

{
  "type": "chart",
  "sender_type": "ai",
  "chart_data": {
    "chart_type": "line",
    "x": [...],
    "y": [...]
  },
  "timestamp": ...
}

Outro

항상 프로젝트가 끝나면 여기도 고쳐야되고 저기도 고쳐야할 것 같고 그렇습니다. 짧은 기간에 임팩트있게 구현하려다 보니 부족한 점이 먼저 눈에 띄는 것 같아요. 하지만? 이제.. 시간이 좀 강제로 생겼기 때문에 ^^ 차근차근 리팩토링을 진행해볼까 합니다.

슬슬 회고 쓰고 리팩토링하는 글을 자주 작성할 것 같습니다. 긴 글 읽어주셔서 감사합니다.👍

진짜 마지막으로 윈도우에서는 따봉으로 따봉티콘이 검색됩니다.

이번 프로젝트를 진행하면서 가장 많은... 고민을 한 부분입니다. 답변을 생성하는 과정을 어떻게 구상했는지 흐름과 함께 알아보겠습니다. AI에 대해 지식이 부족해서 짱 멋진 팀원들의 도움으로 아키텍쳐를 설계했습니다.

전체 아키텍쳐

저희 서비스의 답변 생성 전체 아키텍쳐입니다.

• 사용자의 질문 의도를 파악할 것

• 답변은 CoT 흐름 : NL2SQL -> Chart -> Insight
• 프롬프팅 효율을 위한 RAG 기반 스키마 검색

사용자 입력 ➡️ 질문 분류

• analysis : 분석 요청

• follow-up : 이전 흐름에 대한 후속 질문
• confused : 흐릿하거나 명확하지 않은 질문

일상 대화도 처리하는 Chat GPT와 다르게 저희 서비스는 기업 데이터베이스 분석에 특화되어 있는 "에이전트" 성격이 강했기 때문에 최초 질문을 분류하는 로직을 추가해야한다 생각했습니다. LangChain으로 구현한 프록시 역할의 LLM으로 분기 처리를 한 후 어떤 답변을 생성할지 결정하는 부분입니다. 해당 로직의 판단을 돕기 위해 이전 대화 기록이 있다면 대화 기록을 프롬프트에 추가하였고 사용자의 custom_dict를 조회해 더 정확한 판단을 내릴 수 있었습니다.

CoT : NL2SQL -> Chart -> Insight

핵심 기능이었던 답변 생성입니다. 이전 프로세스에서 analysis로 분류되면 다음과 같은 연쇄 처리가 일어납니다.

1. NL2SQL : 프롬프팅과 RAG를 통해 기업 DB 전용 쿼리 생성

• RAG를 통한 쿼리 정확도 향상
  • Schema Description을 Pinecone에 벡터 임베딩
  • 사용자 질문과 가장 유사한 Top-K 스키마 설명 검색
  • ReRank 과정을 거쳐 최종 Top-K 선정
  • RAGAS를 이용한 정확도 (precision@K, recall@K) 측정
• BigQuery 검증
  • 최대 5번의 재검증 (쿼리 정확도를 위함)
  • BigQuery의 고유 문법 프롬프팅에 추가
  • Pinecone에도 추가하여 최대한 비슷한 문법을 찾아가도록 유도

2. NL2Chart, NL2Graph : SQL 결과를 바탕으로 적절한 시각화 방식 선택 및 JSON 생성

• 초기 개발 단계에서는 이미지로 저장할 생각 (정적 리소스)
• 구글 멘토님이 동적으로 그래프를 볼 수 있는 방법이 있는 것이 좋을 것 같다 ➡️ JSON 형식으로 제공 및 최선의 그래프 형태 제공
• 최선의 그래프 형태를 제공하기 위한 LLM 프롬프팅 추가, 그래프가 필요없는 질문의 경우 (ex: 재직 중인 사원수 알려줘)도 판단할 수 있게 추가

3. Insight 도출 : NL2SQL, NL2Chart를 기반으로 LLM이 핵심 인사이트 요약 및 설명 작성

• 이전에 생성된 SQL, Chart를 CoT에 이용해 최종 인사이트 생성
• 차트에 대한 설명, 특이점, 추가 질문이나 개선 사항 등을 제공

Follow-up 답변 & 실패 대비 로직

질문이 Follow-up으로 분류되면 직전 질문의 SQL 결과와 인사이트를 포함한 대화 히스토리 기반으로 답변이 생성됩니다. 답변 이어쓰기가 아닌 대화 맥락을 유지한 분석 + 연속적인 데이터 탐색 유도를 위한 역할입니다.

또한 실패 로직도 많이 고민한 부분입니다. 실제로 ChatGPT의 초기에 재미를 넘어 범죄에 이용될만한 질문이 많았습니다.

윈도우키를 제공하거나 기업 기밀 유출 사고 등이 빈번하게 발생했습니다. 저희도 이러한 생성형 AI의 문제점을 자각하였고 1차적으로 Gemini 2.5 pro의 향상된 보안 탐지 정책에 더해 저희들만의 보안 감지를 고려했습니다. 사내 시스템에 맞게 우선 데이터 권한에 대한 탐색을 우선했습니다.

• 현재 로그인한 사람을 게이트웨이를 통해 Role을 파악
• 질문 시 프롬프트에 강한 제약 제공 (Role에 따른 각 스키마 접근 권한 부여)
• SQL 생성 단계에서 선택되는 Table의 접근 권한을 LLM 에서 제한 + 서버단에서 제한, 총 두번에 거친 검증

하지만 시간이,,, 촉박하여서 서버 검증 로직 구현을 추가하지 못햇습니다. LLM 프롬프팅을 통한 필터링은 생각보다 제어력이 약했습니다. 추가 리팩토링을 진행하여 권한 체크에 따른 보안 강화를 해야겠습니다.

WebSocket에 대해

답변 생성 파이프라인을 설계하며 동시에 채팅 로직에 대해 생각했습니다.

질문 ➡️ NL2SQL ➡️ SQL 실행 ➡️ Chart 생성 ➡️ Insight 추출

이 과정을 한번에 끝내는 동기 처리보다는 중간 단계마다 결과를 사용자에게 실시간으로 전달하는 방식이 훨씬 자연스럽다 생각했고 (ChatGPT 처럼) 통신 방식으로 SSE, WebSocket 중에 고민 했습니다.

SSE

이전 글에도 작성했듯이 SSE는 단방향 스트리밍 방식으로 서버에서 클라이언트로 실시간 데이터를 푸시할 수 있습니다. 하지만 서비스 구조와 적합하지 않다 생각했습니다. 그 이유는 다음과 같습니다.

• 사용자의 질문을 서버로 보내야함 (API 통신으로 따로 구현해야함)
• 중간 응답 (NL2SQL 결과, Chart, Insight)를 나눠서 보내야 하는데 SSE는 메시지 분류나 통제에 한계가 있음
• 향후 기능 확장 (답변 중단 요청, 실시간 피드백 등)에 제한적

모든 내용을 Markdown으로 실시간 파싱하는 ChatGPT와는 다르다 생각해서 Websocket을 생각했습니다. 또한 개발자와 관리자를 위해 콘솔과 유사한 기능 또한 존재했기 때문에 WebSocket을 생각했습니다.

WebSocket

WebSocket도 완벽히 적합하다고 생각하진 않았지만 클라이언트와 서버 간 연결을 지속하며 양방향 통신이 UX 측면에서 더 유리하다 생각했습니다.

• 실시간 질문 전달
• 중간 응답 스트리밍
• 사용자 상호작용
• 상태 관리
• 유연한 예외 핸들링

가장 중요한 건 상태 관리였습니다. WebSocket은 최초 연결 후 websocket.state를 통해 상태관리가 가능합니다. 저희 서비스는 사용자의 custom_dict, role 등이 필요했기 때문에 연결 후 해당 내용을 매번 가져오기보다 한번 저장 후 계속 사용할 수 있는 WebSocket이 더 효율적일 것이라 판단했습니다. 또한 어떤 파이프라인을 처리하다 에러가 발생했는지 파악하기 위해 예외 처리가 중요했는데 이 또한 WebSocket이 더 유연하다 생각하여 최종 통신 방식으로 WebSocket을 선택했습니다.

Outro

LLM 파이프라인을 처음 설계해보면서 정말 많이 배운 것 같습니다. 벡터 DB, LangChain, WebSocket 등 다양한 기술을 활용해보며 배울 수 있었습니다. 프로젝트 초기에 우아한 형제들의 물어보새 프로젝트를 많이 참고하였는데 많은 도움이 되었습니다. 정말 비슷한 프로젝트라 생각했는데 최신 LLM을 활용하니까 정확도가 높게 잘 나오더라구요. 나라에서 제공해주는 AI 학습용 데이터로 카드사 데이터가 약 1억개가 넘었는데 이 분량의 데이터도 빠르게 분석할 수 있는 프로젝트를 해냈다는 것이 매우 뿌듯합니다.

AI가 참 신기하고 재밌으면서도 개발자한텐 적인 것 같습니다. 직설적으로 AI의 등장으로 취업난이 가속화되고 있기 때문에 마냥 유쾌하진 않습니다. 그래도 새로운 기술의 등장에 움츠러들기보단 더욱 적극적으로 활용해보는 것이 좋은 것 같습니다. 퉁퉁퉁 사후르의 가질 수 없다면 부숴버리겠단 마인드로 AI를 활용할 수 있는 개발자가 되어야겠다는 생각을 가질 수 있는 프로젝트였습니다.

싸피의 마지막 프로젝트로 구글 기업 연계 DB Deep을 진행했습니다. DB Deep은 LLM을 활용한 기업 데이터 분석 서비스입니다. 데이터 분석은 전문성을 요구하는 영역이었으나 생성형 AI의 등장으로 여러 기업에서 쉽게 접근할 수 있는 방법을 모색 중 입니다. 이번 프로젝트를 진행하며 새로운 기술을 활용하고 웹소켓 채팅을 구현하려다보니 석연치 않은 부분이 많아 정리를 해보겠습니다.

Chat GPT는?

가장 먼저 찾아본 서비스는 Chat GPT입니다. 가장 처음 생성형 AI를 선보인 기업이고 지금도 압도적인 업계 1위를 지키고 있는 서비스인 Chat GPT는 SSE (Server-Sent Event)를 사용하고 있습니다.

SSE

서버에서 클라이언트로 단방향 실시간 데이터를 전송하는 HTML5 표준입니다. 브라우저에서 JavaScript를 수신하며, 실시간 알림이나 로그 스트리밍에 유리합니다.

[흐름]

• 클라이언트가 EventSource를 통해 서버 연결
• 서버는 text/event-stream Content-Type으로 응답
• 서버는 데이터를 일정 간격 or 이벤트 발생 시 전송
• 클라이언트는 자동으로 수신 및 처리
• 연결이 끊기면 클라이언트가 자동 재접속 시도

[장점]

• HTTP 기반으로 구현 및 디버깅 용이
• 자동 재연결, 이벤트 ID 재사용 등 기본 기능 제공
• 양방향 통신이 불필요한 경우 효율적

[단점]

• 단방향 통신이라 클라이언트는 서버에 요청 불가능
• 브라우저 탭이 많으면 연결 수 제한
• HTTP/2에서는 동시 연결 관리 유리, HTTP/1.1에서는 제약 존재

ChatGPT가 SSE를 선택한 이유는 아마도? 다음과 같습니다.

• HTTP 통신으로 요청, SSE로 응답
  • 양방향 통신 불필요
  • 단방향 스트리밍 응답이 필요
• 방화벽 차단 및 프록시 통과에 SSE가 Websocket 대비 유리
• 핸드셰이크 복잡 (이건 아마 HTTPS와 함께 구성하기..애매해서가 아닐까...요???)

API 통신이 아닌 SSE를 선택한 이유는 생성형 AI의 응답이 완성까지 굉장히 느리기 때문에 사용자를 붙잡아 두기 위한 전략이라 생각합니다. 실제로 GPT 쓰면서도 와다닥 답변 하는 느낌이라 가만 있는 것이지 그마저도 요즘엔 답변 생성하는 도중에 다른 짓 합니다.

DB Deep은

저희는 많이.... 고민했지만 Websocket을 이용해 구현하기로 했습니다. 근거는 다음과 같습니다.

• 생성형 AI 응답 전까지 사용자의 입력 자체가 트리거, 단방향 SSE보다 양방향 통신이 더 자연스럽다 생각
• DB Deep은 사용자의 질문과 피드백을 서버로 계속 전송해야함 (응답 품질 상승 요인)
• SSE는 text/event-stream 기반 단일 메시지 타입만 지원하기 때문에 개발 단계에서 구조화된 메시지 포맷을 통한 쉬운 유지 보수를 의도함

WebSocket

클라이언트와 서버 간 양방향 통신 (Full-Duplex)을 가능하게 해주는 프로토콜, 한번 연결되면 양측 모두 자유롭게 데이터를 주고 받기 가능

[흐름]

• 클라이언트가 서버로 Websocket 연결 요청 ws:// or wss://
• 서버가 핸드셰이크 수락 시 연결 성립
• 이후 클라이언트와 서버는 서로 필요할 때마다 자유롭게 메시지 송수신
• 연결 종료 시 클라이언트 혹은 서버가 명시적으로 close

[장점]

• 양방향 통신 지원
  • 채팅, 실시간 협업, 알림 등 즉각적인 피드백 필요할 때
• 낮은 오버헤드
  • HTTP 요청/응답 처럼 헤더 반복 필요 X

[단점]

• 초기 핸드셰이크 문제
  • HTTP를 이용한 핸드셰이크 이후 Websocket으로 업그레이드
• 방화벽/프록시 제한
  • 일부 네트워크 환경에서는 Websocket 연결 차단
• 연결 수 관리 필요
  • 브라우저/서버에서 동시 유지할 수 있는 연결 수 제한 존재
• 구현 복잡

Outro

6주라는 짧은 기간에 가장 중요한 핵심 기능을 어떻게 구현할까 고민이 너무 짧았던 것 같기도 하고 아쉬운 선택은 아니었나..싶기도 합니다. 사내 서비스라 인프라 요소를 좀 더 고려해서 SSE를 선택했어도 괜찮았을 것 같고... 피드백 기반 성능 개선을 위해선 WebSocket이 더 좋은 것 같기도하고... 아직 웹에서 실시간 통신은 장단점이 있는 것 같습니다. 좀 더 다양한 방법으로 프로젝트를 리팩토링 해보고 싶네요.

출처

ChatGPT - 채팅 MDN - SSE MDN - WebSocket

개념

이분 탐색은 정렬된 데이터에서 구간을 절반씩 나눠가며 좁혀가는 방식으로 원하는 값을 빠르게 찾는 알고리즘 시간 복잡도 : $$O(log N)$$

1 ~ 100 사이의 숫자를 맞추는 Up&Down 게임을 한다 했을 때 어떻게 찾으시나요?

가장 단순한 방법은 선형 탐색입니다. 주어진 데이터의 처음부터 끝까지 순차적으로 돌면서 원하는 값이 있는지 확인하는 방법으로 정렬이 필요없지만 최악의 경우 모든 데이터를 다 확인해봐야하기 때문에 $$O(N)$$의 시간 복잡도를 갖게 됩니다.

하지만 실제로 사람들은 처음으로 50을 외칠 것 입니다. 범위를 반씩 줄여가며 탐색하는 이분 탐색이 더 빠르다는 것을 체감하고 있기 때문입니다. 100이 아닌 1,000,000 개의 데이터가 있다고 하면 더 체감이 됩니다. 선형 탐색은 100만개의 데이터 모두 시도해봐야하지만 이분 탐색은 $$2^{19}<1,000,000<2^{20}$$ 이기 때문에 최대 20번에 찾을 수 있습니다.

이분 탐색은 다음과 같은 흐름으로 동작합니다.

1. 탐색 구간 초기화 left = 0, right = len(arr) - 1 2. 중간값 (mid) 계산) mid = (left + right) / 2 3. 값 비교 및 범위 조정 arr[mid] == target : 답 arr[mid] < target : 왼쪽은 제외 ➡️ left = mid + 1 arr[mid] > target : 오른쪽 제외 ➡️ right = mid - 1

위 과정을 java 코드로 옮기면

public class BinarySearch {
    public static int binarySearch(int[] arr, int target) {
        int left = 0;
        int right = arr.length - 1;

        while (left <= right) {
            int mid = (left + right) / 2;

            if (arr[mid] == target) {
                return mid;
            } else if (arr[mid] < target) {
                left = mid + 1;
            } else {
                right = mid - 1;
            }
        }

        return -1;
    }
}

lower_bound, upper_bound

이분 탐색의 구현 자체는 단순합니다. 하지만 특정 값을 찾을 때만 단순 구현으로 ps가 가능합니다. 다음과 같은 경우엔 lower_bound, upper_bound 개념이 필요합니다.

• 어떤 값이 몇 개 있는지 세기
• 어떤 값 이상인 첫 위치 찾기
• 어떤 값보다 큰 값이 처음 나오는 지점 찾기
• 데이터를 정렬 상태로 유지하면서 적절한 삽입 위치 찾기

이 상한선, 하한선 내용이 굉장히 헷갈립니다. 백준에 잘 정리된 글이 있으니 같이 읽어보셔도 좋을 것 같아요.

개념부터 정리하고 가겠습니다.

두 함수 모두 "해당 조건을 만족하는 첫 번째 위치" 를 찾습니다.

이분 탐색을 값 자체를 찾기 위한 목적이 아닌, 특정 조건을 만족하는 최소 인덱스를 찾기 위해 사용합니다.

• left : 항상 유효 후보군
• right : 항상 무효 후보군
• left < right : 조건을 만족하는 가장 왼쪽 경계로 수렴

[lower_bound 구현]

// arr[i] >= target인 첫 번째 위치를 반환
public static int lowerBound(int[] arr, int target) {
    int left = 0;
    int right = arr.length;

    while (left < right) {
        int mid = left + (right - left) / 2;

        if (arr[mid] < target) {
            // target보다 작은 값은 제외해도 됨
            left = mid + 1;
        } else {
            // target 이상이면 범위에 포함될 수 있음 → 계속 왼쪽으로 이동
            right = mid;
        }
    }

    return left; // !!!!left가 항상 유효 후보군!!!!
}

[upper_bound 구현]

// arr[i] > target인 첫 번째 위치를 반환
public static int upperBound(int[] arr, int target) {
    int left = 0;
    int right = arr.length;

    while (left < right) {
        int mid = left + (right - left) / 2;

        if (arr[mid] <= target) {
            // target 이하인 값은 후보 제외
            left = mid + 1;
        } else {
            // target 초과 가능성 있음 → 계속 왼쪽으로 이동
            right = mid;
        }
    }

    return left;
}

이분 탐색 코드에서는 while(left <= right), return mid 였는데 왜 upper_bound, lower_bound에서는 while(left < right) , return left일까요? while 문의 종료 조건에 따라 달라지는 것이므로 개념을 확실히 알고 사용해야합니다.

1. while (left <= right) = 정답이 정확히 배열 안에 있을 때

   public static int binarySearch(int[] arr, int target) {
    int left = 0;
    int right = arr.length - 1;
   
    while (left <= right) { // 등호 포함
        int mid = left + (right - left) / 2;
   
        if (arr[mid] == target) {
            return mid; // 정답
        } else if (arr[mid] < target) {
            left = mid + 1;
        } else {
            right = mid - 1;
        }
    }
   
    return -1; // 정답 없음
   }

• 종료 조건 : left > right ➡️ 범위 다 돌아도 못 찾음
• 탐색 대상 : 정확히 일치하는 값

💡 left == right인 순간도 비교하는 경우 <= 사용

2. while(left < right) = 조건을 만족하는 최소 위치를 찾는 경우

   // 조건을 만족하는 "가장 왼쪽 경계"를 찾기 위함
   while (left < right) {
    int mid = left + (right - left) / 2;
   
    if (조건을 만족하는가?) {
        right = mid;
    } else {
        left = mid + 1;
    }
   }
   return left; // 또는 right (같음)

• 종료 조건 : left >= right ➡️ left = right 일 때 while문 종료
• left == right일 때 해당 값을 반환함
• 정확한 값의 존재 여부 X, 조건을 만족하는 최초 인덱스를 찾음 (그 값이 배열의 끝일 수도 있기 때문에 left == right 까지 조사)
• lower_bound나 upper_bound는 조건을 만족하는 값이 배열에 존재하지 않아도, 그 값이 정렬을 유지하며 들어갈 수 있는 위치를 반환합니다. 예를 들어 lower_bound(arr, 10)에서 arr이 {1, 2, 5}라면 결과는 3입니다.

매개변수 탐색 (Parametric Search)

항상 이분탐색하면 같이 나와주는 매개변수 탐색입니다. 이분탐색은 하나의 값을 찾는 알고리즘이라면 매개변수 탐색은 이분탐색을 이용하여 가장 적절한 값을 찾아주는 알고리즘입니다. 예제를 통해 확실한 구분이 가능합니다.

[백준 2110] 공유기 설치

대표적인 매개 변수 탐색 문제인 공유기 설치입니다.

• 정답이 명확하지 않음 : 값을 직접 찾는 것이 아닌 ((특정 조건을 만족하는 값 중에 가장 적절한 값))을 바랍니다. 해당 문제에선 가장 적절한 값 = 최대값 입니다.
• 이분 탐색을 통해 범위를 좁혀 나갈 수 있습니다.
• isValid() 같이 유효성을 검사하는 함수를 사용하게 됩니다.

import java.io.*;
import java.util.*;

public class Main {
    static int N, C, ans;
    static int[] nums;

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringTokenizer st = new StringTokenizer(br.readLine());
        N = Integer.parseInt(st.nextToken());
        C = Integer.parseInt(st.nextToken());
        nums = new int[N];
        for(int i = 0 ; i < N; i++) {
            nums[i] = Integer.parseInt(br.readLine());
        }
        Arrays.sort(nums);
        int left = 1, right = nums[N-1] - nums[0];
        while(left <= right) {
            int mid = (left + right) / 2;
            if(isValid(mid)){
                ans = mid;
                left = mid + 1;
            }else {
                right = mid -1;
            }
        }
        System.out.println(ans);
    }
    private static boolean isValid(int dist) {
        int cnt = 1;
        int last = nums[0];
        for(int i = 1; i < N; i++) {
            if(nums[i] - last >= dist) {
                cnt ++;
                last = nums[i];
            }
            if(cnt >= C) return true;
        }
        return false;
    }
}

활용 사례

Arrays.binarySearch()

int result = Arrays.binarySearch(int[] arr, int key) result >= 0 : key의 인덱스 result < 0 : key가 존재 X ➡️ 삽입 위치를 반환

Java에서는 배열에서 이분 탐색을 수행할 때 직접 구현하지 않고 사용할 수 있는 표준 라이브러리를 제공합니다.

key 값을 찾으면 인덱스를 리턴하고 없으면 -Insertion point - 1값을 반환한다고 합니다. 이 값은 key를 배열에 삽입해서 정렬 상태를 유지하려할 때 어디에 삽입해야 하는지 알려주는 위치입니다.

int[] arr = {1, 3, 5, 7};

System.out.println(Arrays.binarySearch(arr, 5)); // 2 (존재)
System.out.println(Arrays.binarySearch(arr, 4)); // -3 (→ 삽입 위치 2)

즉, 음수 값이면 -(result + 1)를 하면 어디에 삽입해야하는지가 나옵니다. 이렇게 복잡하게 설정한 이유는 아마 삽입 위치가 0일 경우를 고려해서라 생각합니다. 이 라이브러리의 key 값은 lower_bound 처럼 항상 최소 인덱스를 보장하지 않기 때문에 중복 요소가 있을 경우 보장된 위치가 아닙니다.

LIS

최장 증가 수열인 LIS 문제에서도 이분 탐색이 사용됩니다. 현재 숫자가 들어갈 위치를 이분 탐색으로 찾은 후 그 위치에 값을 덮어 씌우는 방식입니다.

List<Integer> lis = new ArrayList<>();
for (int num : arr) {
    int pos = Collections.binarySearch(lis, num);
    if (pos < 0) pos = -(pos + 1);
    if (pos == lis.size()) lis.add(num);
    else lis.set(pos, num);
}

Intro

벌써 SSAFY의 마지막 세번째 프로젝트를 진행 중 입니다. 마지막인만큼 엄청난 프로젝트를 하고 싶었는데 멋진 팀원들 덕분에 구글과의 기업 연계에 선정되었습니다. SSAFY에서 기본적으로 괜찮은 사양의 AWS EC2를 제공해주긴 하지만 구글 기업연계인만큼 GCP를 활용하려합니다. 또한 호환성이... 두려워 AI 모델을 사용하는 어플리케이션 서버는 GCP를 활용하기로 했습니다. SSAFY는 완전 멋있기 때문에 마지막 자율 프로젝트에서는 지원을 많이 해줍니다. 그래서 GCP를 정말 열심히 써보기로 결심했는데 역시..! 클라우드 사업은 너무 다양한 종류의 서비스가 있어서 처음이 가장 어려운 것 같습니다. 이러한 어려움을 겪고 있을 미래의 저를 포함한 사람들을 위해 GCP가 제공하고 있는 다양한 서비스에 대해 알아보겠습니다. 주로 제가 사용하게 될 서비스 위주로 넓고 얕게 알아보겠습니다!

GCP vs AWS

캬 자극적인 제목 벌써 두렵습니다. 그래도 궁금하잖아요. 비슷한 서비스를 엄청...많이... 하고 있는데 어떤 차이가 있는지 가볍게 알아보겠습니다.

기본 비교

AWS

• 클라우드 업계 1등
• 서비스 종류, 기능, 안정성 모두 압도적 선두주자
• 기업용 대규모 인프라 구축에 강점
• 문서화 친절해서 사용하기 좋음

GCP

• 후발주자는 어쩔 수 없다.
• AL/ML, 데이터 분석, 쿠버네티스 같은 특정 영역은 업계 최고 수준
• 관리형 서비스 (GKE, BigQuery, Vertex AI 등)가 매우 강력함

Gen AI

가장 큰 차이점은 역시 구글은 Gemini가 있다는 것이고 AWS는 없다는 것입니다. 저희는 AI가 주된 서비스를 개발할 것이기 때문에 당연히 GCP가 유리하다고 판단했습니다. 아부 아니고 진짜로 AI 개발에 GCP가 좀 더 낫습니다. 물론 Gemini를 쓸 때의 이야기지만. AWS는 SageMaker라는 관리형 AI 플랫폼이 있습니다. 타 AWS 서비스와 마찬가지로 초기 설정이 조금 복잡하고 진입 장벽이 높습니다. GCP는 Vertex AI, AutoML, BigQueryML 등 다양한 AI 플랫폼을 서비스 중이고 모든 과정을 쉽게 접근할 수 있다는 장점이 있습니다. 저희는 Gemini라는 사전 학습된 모델 API를 사용하기 때문에 Vertex AI Studio를 통해 Gemini API를 직접 파인 튜닝없이 호출하는 기능, 빅쿼리의 AI Assistance 등 기능 호환성 덕분에 GCP가 훨씬 유리하다 판단했습니다.

GCP 서비스

이번 프로젝트에서 꽤 다양한 GCP 서비스를 사용해보고 싶어 잔뜩 준비해봤습니다. 기존의 AWS 서비스로 배포한 과정에 필요한 서비스에 더해서 좀 더 관리를 빡세게 가져가는 버전이라 생각해주시면 될 것 같습니다.

Cloud DNS

• 역할 : 이름 그대로 DNS 관리를 해줍니다.

도메인 구매가 자체적으로 가능한 AWS Route 53과 달리 GCP의 Cloud DNS는 외부에서 구매한 DNS만 사용 가능합니다.

Cloud CDN

• 역할 : 정적 파일 (React 빌드 결과 등)의 빠른 배포 담당

역시.. 이름 그대로 CDN 역할을 맡습니다. 굳이 필요없긴 하지만 저희 서비스는 이미지와 파일을 굉장히 많이 필요하기 때문에 CDN이 UX에 필수라 생각했습니다.

GCP Storage

• 역할 : 저장소, Bucket

저희 서비스에서 제작되는 이미지, 정적 빌드 파일 등을 저장하는 곳입니다. AWS S3와 거의 똑같습니다.

BigQuery 🌟

- 역할 : 대용량 데이터 분석

요즘 아주 핫하고 엄청 밀어주는 서비스 중 하나입니다. 서버리스 데이터 웨어하우스 서비스로 사용자 활동 기록, 대규모 로그 데이터 등을 분석할 때 사용합니다. 기존 DB로 감당하기 힘든 대량 데이터 (저희 더미 셋이 엄청 많습니다...)를 초고속 쿼리가 가능합니다. 또 쿼리 Validation이 내부적으로 이루어지고 캐싱까지 해주는 등 기능이 대단합니다. 쿼리문으로 AI 학습까지 된다는데 너무 신기합니다. 근데 저희가 모델 학습을 하는 서비스는 아니라 사용해보고 싶은데 조금 아쉽습니다. 좋은 기능이 많기에 가격이 살짝 나가는 편이긴합니다. 그래도 접근 DB의 용량별 가격 측정으로 가성비가 나쁘다는 생각은 안듭니다. 저희 서비스의 가장 핵심 기능이라 별표 달았습니다.

GCP Cloud Run 🌟

- 역할 : FastAPI 백엔드 배포

굉장히 신박한 기능을 가진 서비스입니다. Cloud Run은 컨테이너 이미지를 가져와서 자동 배포 및 필요할 때만 인스턴스 실행이 가능합니다. 서버리스 컨테이너라는 말 그대로 서버 관리 부담 없이 어플리케이션 운영이 가능합니다.

• 트래픽 없으면 0원
• 오토 스케일링에 매우매우 유리
• 배포가 너무..쉬움

GCP 정말 잘하는데 AWS한테 밀리기만 할 플랫폼이 아닌데 라는 생각이 듭니다.

Firestore

• 역할 : 실시간 데이터 저장 및 읽기

Document NoSQL을 위한 서비스입니다. 유저 세션, 상태 관리, 메타데이터 저장 등에 사용합니다. 실시간 업데이트 기능이 강력한 편이라 채팅..기능에 사용해보려 합니다.

Cloud SQL

• 역할 : RDB 관리

완전 관리형 관계형 DB입니다. DB 설치/운영이 필요 없고 백업, 스케일링, 모니터링 모두 다 해주기 때문에 할 일이 많이 줄어듭니다. 일반적인 트랜잭션 기반 데이터 관리에 최적화 되어 있다 생각됩니다. 근데 조금 비싼 것 같습니다. AWS RDB와 비슷한 서비스인데 RDB도 비싼데 왜 관계형 DB 서비스는 비싼지 궁금합니다.

Vertex AI (Gemini)

• 역할 : AI 어플리케이션 학습 및 배포를 위한 GCP 중앙 집중식 플랫폼

사실 이거 안씁니다. 근데 궁금해서 작성합니다. ML/DL 엔지니어링의 워크플로우를 간소화하고 데이터셋 생애 주기 단순화가 주된 서비스입니다. AutoML, Pipeline, Pretrained API, Integration, Monitoring/Management 등을 모두 하나의 서비스에서 관리할 수 있기 때문에 많은 AI 개발자가 사용한다고 합니다.

추가) Pinecone (Vector DB)

• 역할 : Vector DB

저는 AI 개발자가 아니기 때문에 정확히 벡터 DB가 무엇인지 몰랐습니다. 근데 이전에 작성한 추천 알고리즘 글과 관련이 있는 것이 벡터 임베딩해서 코사인 유사도를 구하지 않았습니까 저희가? 물론 내부에서 돌려서 라이브러리만 구동한 저는 정확한 원리는 몰랐지만, 이 과정을 도와주고 저장까지 되는 도구라고 합니다. Pinecone을 활용해서 DB의 스키마 및 컬럼 메타데이터 등을 넣어서 ...유사도를 분석할 것 같습니다.

Outro

오늘은 뭔가 개요를 작성한 것 같은데 실제로 서비스를 이용해보며 어떤 것이 좋고 어떤 것이 불편했는지 정리해보겠습니다. 어째 작성하고 보니 GCP가 지금보다는 더 많은 파이를 가져야 한다는 생각을 하게됩니다. 아마 아이콘이 너무 못생겨서 그런 것이 아닐까요? 이번 프로젝트 매우 기대가 되는데 기대만큼 열심히 해보겠습니다.