• 트랜잭션이란 작업의 완전성을 보장해 주는 것이다. 하나의 논리적인 작업을 모두 완전하게 처리하거나 처리하지 못할 때 원 상태로 복구하여 일부만 적용되는 현상을 만들어주지 않게 한다.
• 아래는 트랜잭션과 관련해서 헷갈릴 수 있는 개념이다.

  잠금(Lock)은 동시성을 제어하기 위한 기능으로 여러 트랜잭션이 같은 데이터에 접근할 때 데이터 충돌이나 불일치를 방지한다.

• 트랜잭션*은 데이터 정합성을 보장하기 위한 기능이다. 데이터 정합성(Constistency)이란 데이터가 일관성 있게 유지되는 것을 말한다. 이 외에도 원자성(Atomicity), 고립성(Isolation), Durability(지속성) 특징을 가진다.
• 데이터 무결성*은 데이터가 정확한 상태를 유지하는 것을 말한다.

세 가지 부정합 문제

1. DIRTY READ

• 트랜잭션이 아직 커밋되지 않은 다른 트랜잭션의 변경사항을 읽는 경우를 말한다.

2. NON-REPEATABLE READ

• 같은 트랜잭션 내에서 동일한 쿼리를 여러 번 실행했을 때, 그 결과가 다른 경우를 말한다.

3. PHANTOM READ

• 트랜잭션이 동일한 범위 쿼리를 두 번 이상 실행할 때, 다른 트랜잭션이 데이터를 삽입하거나 삭제함으로써 결과 집합이 달라지는 현상이다.

트랜잭션 격리 수준

1. READ-UNCOMMITTED

• 각 트랜잭션 변경 내용이 COMMIT이나 ROLLBACK 여부에 상관없이 다른 트랜잭션에서 보인다.

• 사용자 A는 emp_no가 50000이고 first_name이 Lara인 사원을 INSERT 한다.
• 사용자 B가 변경된 내용을 커밋하기도 전에 사용자 B는 emp_no = 50000인 사원을 검색한다.
• 사용자 A가 처리 도중 알 수 없는 문제가 발생해 롤백한다고 해도 여전히 사용자 B는 "Lara"가 정상적인 사원이라고 생각하고 계속 처리할 것이 문제다.

2. READ-COMMITTED

• COMMIT이 완료된 데이터만 다른 트랜잭션에서 조회할 수 있다.
• 오라클 DBMS에서 기본적으로 사용되는 격리 수준이다.

• 사용자 A는 emp_no=50000 사원의 first_name을 "Lara"에서 "Toto"로 변경했다. 이때, 새로운 값인 "Toto"는 테이블에 즉시 기록되고 이전값인 "Lara"는 언두 영역으로 백업된다.
• 사용자 A가 커밋하기 전에 B가 emp_no=50000을 조회하면 first_name 칼럼의 값은 "Toto"가 아니라 "Lara"로 조회된다. (언두 영역에 있는 데이터 조회)

NON-REPEATABLE READ 부정합 문제

• 사용자 B가 BEGIN 명령으로 트랜잭션을 시작하고 first_name이 "Toto"인 사원을 검색했는데 결과가 없었다. 하지만 사용자 A가 "Toto"로 변경 후 사용자 B가 똑같은 쿼리로 다시 조회하면 결과가 1건 조회된다.

3. REPEATABLE-READ

• MySQL에선 MVCC를 이용해 언두 영역에 있는 백업된 이전 데이터를 이용해 동일 트랜잭션 내에서 동일한 결과를 보여주는 것을 보장한다.
• MySQL InnoDB 스토리지 엔진에서 기본으로 사용하는 격리 수준이다.

• 사용자 A의 트랜잭션 번호는 12, 사용자 B의 트랜잭션 번호는 10이다.
• 사용자 A는 사원의 이름을 "Toto"로 변경하고 커밋한다.
• 사용자 B는 A 트랜잭션의 변경 전후 각각 한 번씩 SELECT 했지만, 결과는 항상 "Lara"라는 값을 가져온다. B는 자신의 트랜잭션 번호(10)보다 작은 트랜잭션 번호에서 변경한 것만 보게 된다.

PHANTOM READ 부정합 문제

A가 employees 테이블에 INSERT를 실행하는 도중에 사용자 B가 SELECT FOR UPDATE 쿼리로 테이블을 조회하는 상황

• B는 BEGIN 명령으로 트랜잭션을 시작한 후 SELECT를 수행한다. 위에서처럼 두 번의 SELECT 쿼리 결과는 똑같아야 한다.
• 하지만 첫 번째 SELECT FOR UPDATE에선 A가 커밋하기 전이므로 "Lara" 결과 한 건만 검색되고, 두 번째 SELECT FOR UPDATE에선 A가 커밋한 이후이므로 새롭게 추가된 "Georgi" 결과까지 추가로 검색된다.
• 쓰기 잠금을 거는 것인데, 언두 레코드에는 쓰기 잠금을 걸 수 없다. SELECT FOR UPDATE, SHARE 로 조회되는 레코드는 언두 영역의 레코드를 가져오는 것이 아니라 현재 레코드의 값을 가져오기 때문이다.
• ** 위의 예시는 PHANTOM READ의 예외적인 상황이라 설명하고, MySQL REPEATABLE-READ 격리 수준에선 팬텀 리드가 발생하지 않는다고 한다. 그럼 일반적인 팬텀 리드 상황을 생각해 보고, MYSQL에선 어떻게 방지되는지 찾아보자.**

일반적인 PHANTOM READ 부정합 문제

• 사용자 A는 id > 50000 유저 조회, 결과 (50001, 50002)
• 사용자 B는 INSERT 완료
• 사용자 A는 id > 50000 유저 다시 조회, 결과 (50001, 50002, 50003)
• 초기 SELECT 쿼리와 다른 결과, PHANTOM READ 발생

하지만, MySQL에서는 갭 락과 넥스트 키 락 덕분에 PHANTOM READ 발생하지 않음

• 즉, id > 50000 조회할 때 읽기 락(갭락, 넥스트 키 락)이 걸리며 INSERT가 되어도 추가되지 않고, 트랜잭션 내에서 일관된 읽기를 할 수 있는 것이다.

4. SERIALIZABLE

• 한 트랜잭션에서 읽고 쓰는 레코드를 다른 트랜잭션에서는 절대 접근할 수 없는 격리 수준이다.

MySQL 잠금(락)

• MySQL 잠금은 크게 스토리지 엔진 레벨과 MySQL 엔진 레벨로 나눌 수 있다. MySQL 엔진 레벨의 잠금은 모든 스토리지 엔진에 영향을 미치지만, 스토리지 엔진 레벨의 잠금은 스토리지 엔진 간에는 영향을 미치지 않는다.

1. MySQL 엔진 잠금(락)

1) 글로벌 락

• 한 세션에서 글로벌 락을 획득하면 다른 세션에서 SELECT를 제외한 대부분의 DDL, DML을 실행할 때 글로벌 락이 해제될 때까지 해당 쿼리는 대기 상태가 된다.
• FLUSH TABLES WITH READ LOCK 명령으로 획득할 수 있으며, MySQL에서 제공하는 잠금 가운데 가장 범위가 넓다.

2) 백업 락

• 백업 락은 MySQL에서 백업 수행 중 데이터의 일관성을 유지하고, 복제 및 백업 실패를 방지하기 위해 도입된 기능이다.
• InnoDB 스토리지 엔진은 트랜잭션을 지원하므로 데이터 변경 작업을 멈출 필요 없다. 백업 락을 획득한 경우 테이블의 스키마나 사용자 인증 관련 정보를 변경할 수 없다. 즉, 일반적인 데이터 변경은 가능하지만, 스키마 변경은 불가능한 락이라고 생각하면 된다. (글로벌 락 보다 가벼운)

3) 테이블 락

• 개별 테이블 단위로 설정되는 락이다.
• 명시적 또는 묵시적으로 특정 테이블 락을 획득할 수 있다. 명시적으로는 LOCK TABLES table_name [ READ | WRITE] 해당 명령어를 통해 락을 획득하고, UNLOCK TABLES로 락을 반납한다.
• InnoDB 테이블의 경우 스토리지 엔진 차원에서 레코드 기반의 락을 제공하기에 단순 데이터 변경은 묵시적인 테이블 락이 설정되지 않는다. DDL의 경우에만 테이블 락이 설정된다.

4) 네임드 락

• GET_LOCK() 함수를 이용해 임의의 문자열에 대해 락을 설정할 수 있다.
• 잠금 대상이 테이블이나 레코드 또는 AUTO_INCREMENT와 같은 데이터베이스 객체가 아니고, 단순히 사용자가 지정한 문자열에 대해 획득하고 반납하는 잠금이다.
• 많은 레코드에 대해 복잡한 요건으로 레코드를 변경하는 트랜잭션에 유용하게 사용할 수 있다. 배치 프로그램처럼 한꺼번에 많은 레코드를 변경하는 쿼리는 자주 데드락의 원인이 되곤 하는데, 동일 데이터를 변경하거나 참조하는 프로그램끼리 분류해서 네임드 락을 걸고 쿼리를 실행하는 게 아주 간단히 해결하는 방법이다.

5) 메타데이터 락

• 데이터베이스 객체(테이블, 뷰 등)의 이름이나 구조를 변경하는 경우에 획득하는 락이다.
• 명시적으로 획득하거나 해제할 수 없다. RENAME TABLE tab_a TO tab_b 같이 테이블의 이름을 변경하는 경우 자동으로 획득한다.

2. InnoDB 스토리지 엔진 잠금

• 과거에는 MySQL 명령을 이용해 InnoDB 스토리지 엔진의 락 정보를 얻기 어려웠으나, 최근에는 MYSQL 서버의 infomation_schema에 있는 INNODB_TRX, INNODB_LOCKS, INNODB_LOCK_WAITS라는 테이블을 조인하여 트랜잭션 정보를 확인할 수 있다.

1) 레코드 락

• 레코드 자체만을 잠그는 것을 레코드 락이라고 한다. InnoDB에서는 인덱스의 레코드를 잠근다. 인덱스가 없는 테이블이더라도 내부적으로 자동 생성된 클러스터 인덱스를 이용해 잠금을 설정한다.

• 비관적 쓰기 락(Select For Update), 비관적 읽기 락(Select For Share) 모두 포함하는 개념이다. 아래와 같이 특정 레코드만을 잠그는 쿼리를 생각해볼 수 있다.

  SELECT * FROM time_slot WHERE id = 5 FOR UPDATE;

비관적 락을 사용하지 않더라도 REPEATABLE_READ 격리 수준에서 일관된 읽기를 위해 레코드에 읽기 락이 걸린다.

2) 갭 락

• 갭 락은 레코드 자체가 아니라 인접한 레코드 사이의 간격만을 잠그는 것을 의미한다.

• 레코드와 레코드 사이의 간격에 새로운 레코드가 생성되는 것을 제어하며, 갭 락 그 자체보다 넥스트 키 락의 일부로 자주 사용된다.

• 아래와 같이 1, 5 사이에 있는 레코드(2, 3, 4)를 잠그는 경우를 생각해 볼 수 있다.

  SELECT * FROM time_slot WHERE id > 1 AND Id < 5 FOR UPDATE;

비관적 락을 사용하지 않더라도 REPEATABLE_READ 격리 수준에서 일관된 읽기를 위해 특정 조건에 속한 레코드가 락이 걸린다.

3) 넥스트 키 락

• 레코드 락과 갭 락을 합쳐 놓은 형태의 잠금을 넥스트 키 락이라고 한다.

• 아래와 같이 1 ~ 5 레코드 (1, 2, 3, 4, 5)를 잠그는 경우를 생각해 볼 수 있다.

  SELECT * FROM time_slot WHeRE id >= 1 AND id <= 5 FOR UPDATE; 

비관적 락을 사용하지 않더라도 REPEATABLE_READ 격리 수준에서 일관된 읽기를 위해 특정 조건에 속한 레코드가 락이 걸린다.

4) 자동 증가 락

• 자동 증가하는 숫자 값을 추출하기 위해 AUTO_INCREMENT라는 컬럼 속성값을 제공한다. 한 번에 여러 레코드가 INSERT 되는 경우 각 레코드는 중복되지 않고 저장된 순서대로 증가하는 일련번호 값을 가져야 하기에 AUTO_INCREMENT 락이 필요하다.
• AUTO_INCREMENT 락을 명시적으로 획득하고 해제하는 방법은 없다. 아주 짧은 시간 동안 걸렸다가 해제되는 잠금이다.

인덱스와 락 동작 방식

• InnoDB 잠금은 레코드를 잠그는 것이 아니라 인덱스를 잠그는 방식으로 처리한다. 즉, 변경해야 할 레코드를 찾기 위해 검색한 인덱스의 레코드를 모두 락을 걸어야 한다.

mysql > SELECT COUNT(*) FROM employees WHERE first_name='Georgi';
+----------+
| COUNT(*) |
+----------+
|      253 |
+----------+

mysql > SELECT COUNT(*) FROM employees WHERE first_name='Georgi' AND last_anme='Klassen';
+----------+
| COUNT(*) |
+----------+
|        1 |
+----------+

mysql > UPDATE employees SET hire_date=NOW() WHERE first_name='Georgi' AND last_name='Klassen';

위와 같이 Update 쿼리를 실행하면 1건의 레코드가 변경될 것이다. 그럼 몇 개의 레코드에 락을 걸어야 할까?

• 조건은 first_name='Georgi'이며, last_name 컬럼은 인덱스에 없기에 first_name='Georgi'인 레코드 253건의 레코드가 모두 잠긴다.
• 만약 테이블에 인덱스가 하나도 없다면? 테이블을 풀 스캔하면서 UPDATE 작업을 하게 된다. 즉 전체 30여만 건의 모든 레코드를 잠그게 된다.

잠금 대기 순서 파악하기

• performance_schema의 data_locks 테이블과 data_lock_waits 테이블을 조인해서 잠금 대기 순서를 살펴볼 수 있다.

  select 
    r.trx_id waiting_trx_id,
    r.trx_mysql_thread_id waiting_thread,
    r.trx_query waiting_query,
    b.trx_id blocking_trx_id,
    b.trx_mysql_thread_id blocking_thread,
    b.trx_query blocking_query
  FROM
    performance_schema.data_lock_waits w
  INNER JOIN 
    information_schema.innodb_trx b
  ON
    b.trx_id = w.blocking_engin_transaction_id
  INNER JOIN
    information_schema.innodb_trx r
  ON
    r.trx_id = w.requesting_engine_transaction_id;

  

  *(Update 쿼리 3개를 실행 후 프로세스 목록을 조회한 상황) *

• 현재 대기 중인 스레드는 18, 19번이다.

• 18번 스레드는 17번 스레드를 기다리고 있다.

• 19번 스레드는 17번 스레드와 18번 스레드를 기다리고 있다.

• 17번 스레드가 어떤 잠금을 가졌는지 더 상세히 확인하고 싶다면 performance_schma의 data_locks 테이블이 가진 칼럼을 모두 살펴본다.

  SELECT * FROM performance_schema.data_locks\G

  

• KILL 17 로 강제 종료하면, 나머지 UPDATE 들이 진행되면서 락이 풀린다.

참고 자료

• RealMySQL 8.0

• MySQL 서버는 사람의 머리 역할을 하는 MySQL 엔진과 손발 역할을 담당하는 스토리지 엔진으로 구분한다. 스토리지 엔진은 핸들러 API를 이용해 MySQL 엔진과 데이터를 주고받는다.

1. MySQL 엔진

1) 커넥션 핸들러 (Connection Handler)

• 클라이언트(JDBC, ODBC, ...)와 MySQL 서버 간 연결을 관리하는 MySQL 내부의 컴포넌트이다. 쉽게 말해 SQL 쿼리를 수신하고, 이를 MySQL 엔진에 전달하며 결과를 클라이언트로 반환한다.

2) SQL 파서 (Parser)

• 사용자 요청으로 들어온 쿼리 문장을 최소 단위로 분리(토큰)해 트리 형태의 구조로 만들어내는 작업을 한다.
• 쿼리 문장의 기본 문법 오류는 이 과정에서 발견되고 사용자에게 오류 메시지를 전달한다.

3) 전처리기 (PreProcessor)

• 파서 트리를 기반으로 쿼리 문장에 구조적인 문제가 있는지 확인한다. 토큰을 테이블, 칼럼, 내장 함수와 같은 객체로 매핑하여 해당 객체의 존재 여부와 접근 권환 등을 확인한다.

4) 옵티마이저 (Optimizer)

• SQL 쿼리 요청을 가장 효율적으로 처리하기 위해 쿼리 변환, 비용 최적화, 실행 계획 수립한다. 각각을 간단히 살펴보면,
  1. 쿼리 변환: 쿼리의 구조를 변경하여 더 나은 성능을 제공할 수도 있다.
  2. 비용 최적화: 여러 실행 계획을 시뮬레이션하고, 각 실행 계획에 대한 비용을 계산한다. 비용은 대체로 처리할 데이터의 양, 필요한 I/O, CPU 사용량 등을 기준으로 평가하며 가장 비용이 적은 실행 계획을 선택한다.
  3. 실행 계획 수립: 사용자가 작성한 SQL 쿼리를 분석한 후, 이를 가장 효율적으로 실행할 방법을 결정한다. 인덱스 선택, 조인 순서 결정, 필터 조건 적용 순서 등을 포함한다.

5) 실행 엔진 (Query Execution Engine)

• 만들어진 계획대로 각 핸들러에게 요청해서 받은 결과를 또 다른 핸들러 요청의 입력으로 연결하는 임무를 수행한다.
• 옵티마이저가 두뇌라면 실행 엔진과 핸들러는 손과 발에 비유할 수 있다. 옵티마이저가 group by를 사용하기로 했다면,
  1. 실행 엔진이 핸들러에게 임시 테이블을 만들라고 요청
  2. 다시 실행 엔진은 WHERE 절에 일치하는 레코드를 읽어오라고 핸들러에게 요청
  3. 읽어온 레코드들을 1번에서 준비한 임시 테이블로 저장하라고 다시 핸들러에게 요청
  4. 데이터가 준비된 임시 테이블에서 필요한 방식으로 데이터를 읽어오라고 핸들러에게 요청
  5. 최종적으로 실행 엔진은 최종 결과를 사용자나 다른 모듈로 넘긴다.

위 쿼리가 실행되는 과정을 그림으로 나타내면 아래와 같다.

Handler API

• MySQL과 스토리지 엔진 간의 인터페이스 역할로, 실행 엔진(쿼리 실행기) 요청을 스토리지 엔진으로 전달한다. 아래에서 Handler API가 어떤 식으로 동작하는지 파악할 수 있다. https://github.com/mysql/mysql-server/blob/trunk/sql/sql_handler.cc

2. 스토리지 엔진(InnoDB)

• MyISAM 등 다양한 스토리지 엔진이 존재하지만, 거의 유일하게 레코드 기반 잠금을 제공하는 InnoDB 스토리지 엔진에 대해서 알아본다.
• InnoDB 모든 테이블은 기본적으로 기본 키(Primary Key)를 기준으로 클러스터링 되어 저장된다. 즉, 기본 키값 순서대로 디스크에 저장되며 세컨더리 인덱스는 레코드의 주소 대신 기본 키의 값을 논리적인 주소로 사용한다.

1) MVCC(Multi Version Concurrency Control) 지원

• MVCC는 하나의 레코드에 대해 여러 개의 버전을 동시에 관리하여 잠금을 사용하지 않고 일관된 읽기를 제공한다. InnoDB는 언두로그를 이용해 이 기능을 구현하고 있다.

{12, 홍길동, 서울} 에서 '서울'을 '경기'로 변경해 보자. InnoDB 버퍼 풀은 새로운 값인 '경기'로 즉시 업데이트된다. 언두 로그에는 변경 전 값인 '서울'만 복사하여 저장한다. 이때, 해당 컬럼을 조회했을 때, 어떻게 보일까?
답은 트랜잭션 격리 수준(Transaction Isolation Level)에 따라 다르다. 격리 수준이 READ_UNCOMMITED 라면 InnoDB 버퍼 풀이 현재 가지고 있는 변경된 데이터를 읽어서 반환한다. 격리 수준이 READ_COMMITED 이상이라면 아직 커밋되지 않았기에 언두 로그에 있는 변경 전의 데이터인 '서울'을 반환하는 것이다.

• 이처럼 InnoDB 버퍼 풀 또는 언두 로그 데이터를 반환하는 식으로 여러 개의 버전이 존재하므로 이를 MVCC라고 한다.
• 변경되기 전의 데이터는 언두 로그 영역에서, 변경 후의 데이터는 InnoDB에서 읽어오면 되기에 잠금 없는 읽기가 가능한 것이다.

2) InnoDB 버퍼 풀

• 디스크의 데이터 파일이나 인덱스 정보를 메모리에 캐시해두는 공간이다. 쓰기 작업을 지연시켜 일괄 작업으로 처리할 수 있게 해주는 버퍼 역할도 같이 한다. 버퍼는 변경 작업을 모아서 처리하여 랜덤 디스크 접근을 최소화하는 역할을 한다고 보면 된다.
• InnoDB 버퍼 풀 메모리 공간은 전체 메모리에서 50%에서 시작하여 조금씩 올려가며 최적점을 찾는 것을 권장한다. (innodb_buffer_pool_size)

InnoDB 버퍼 풀 구조

• InnoDB 스토리지 엔진은 버퍼 풀이라는 거대한 메모리 공간을 페이지 크기 조각으로 쪼개어 데이터를 각 페이지에 저장한다.
• 버퍼 풀 페이지 조각을 관리하기 위해 LRU(Least Recently Used) 리스트와 플러시(Flush) 리스트, 그리고 프리(Free) 리스트 자료 구조를 사용한다.
  • LRU 리스트: 가장 최근에 사용되지 않은 페이지 목록이며, 가장 오래 사용되지 않은 페이지일수록 리스트 하단에 위치한다. 메모리 공간이 부족할 때 LRU 리스트의 하단에 위치한 페이지부터 FREE 리스트로 이동한다.
  • FLUSH 리스트: 변경된 데이터 페이지 목록이며, 디스크 변경 작업이 필요한 페이지를 효율적으로 관리하기 위해 사용한다. 주기적으로 플러시 리스트의 플러시 함수를 호출해서 오래전에 변경된 데이터 페이지를 순서대로 디스크에 동기화하는 작업을 수행한다.
  • FREE 리스트: 비어 있는 페이지 목록이며, 사용자의 쿼리가 새롭게 디스크의 데이터 페이지를 읽어와야 하는 경우 사용한다.

InnoDB 스토리지 엔진에서 데이터를 찾는 과정

1. 필요한 레코드가 저장된 데이터 페이지가 버퍼 풀에 있는지 검사 1) InnoDB 어댑티브 해시 인덱스를 이용해 페이지 검색 2) 해당 테이블 인덱스를 이용해 버퍼 풀에서 페이지 검색 3) 버퍼 풀에 이미 데이터 페이지가 있었다면 페이지 포인터를 LRU 상단 방향으로 승급
2. (없었다면) 디스크에서 필요한 데이터 페이지를 버퍼 풀에 적재하고, 적재된 페이지 포인터를 LRU 리스트에 추가
3. 해당 데이터가 실제로 읽히면 페이지 포인터를 LRU 상단 방향으로 승급
4. 버퍼 풀에 상주하는 데이터 페이지는 나이(Age)가 부여되어 오랫동안 사용되지 않으면 버퍼 풀에서 제거
5. 해당 데이터가 자주 접근됐다면 해당 페이지의 인덱스 키를 어댑티브 해시 인덱스에 추가

버퍼 풀과 리두 로그

• InnoDB 버퍼 풀은 서버의 메모리가 허용되는 만큼 크게 설정하면 쿼리의 성능이 빨라진다. (데이터 캐시 역할)
• 또 다른 기능은 InnoDB 버퍼 풀 역할은 쓰기 버퍼링인데, 버퍼 풀의 크기가 크다고해서 쓰기 버퍼링 성능이 좋아지지 않는다. 그렇다면 해당 기능은 어떻게 향상시킬 수 있을까?

고성능 저장 장치를 사용하거나 쓰기 작업을 처리하는 쓰레드 수를 늘리는 방법을 생각해 볼 수 있겠지만, 로그 파일 크기와 관련지어보자.
데이터 수정 작업이 일어나면 리두 로그에 변경 내용이 기록된다. 리두 로그에 기록이 쌓이면 디스크에 반영해야 하는 동기화 문제가 발생한다. MySQL에서는 주기적으로 체크포인트를 발생시켜 변경된 페이지, 리두 로그, 디스크 간 동기화 작업을 진행한다. InnoDB 버퍼 풀이 매우 크지만, 리두 로그 파일의 크기가 매우 적다면 아주 적은 더티 페이지만 버퍼 풀에 보관할 수 있기에 쓰기 버퍼링의 효과는 거의 볼 수 없다. 반대로, InnoDB 버퍼 풀이 매우 작지만, 리두 로그 파일의 매우 크다면 매우 많은 더티 페이지를 한 번에 기록해야 하는 상황이 올 수 있다.
따라서, 리두 로그 파일의 전체 크기를 버퍼 풀 크기의 대략 5~10%로 설정하고 조금씩 늘려가며 최적값을 늘리는 것이 권장된다.

Double Writer Buffer

• 리두 로그는 리두 로그 공간의 낭비를 막기 위해 페이지의 변경된 내용만을 기록한다. 이로 인해 InnoDB 스토리지 엔진에서 더티 페이지를 디스크에 플러시할 때 일부만 기록되는 문제가 발생하면 그 페이지 내용은 복구할 수 없을 수도 있다. 이 같은 문제를 막기 위해 Double Writer Buffer를 사용한다.
• 실제 데이터 파일에 변경 내용을 기록하기 전에 'A'~'E'까지의 더티 페이지를 우선 묶어 한 번의 디스크 쓰기로 DoubleWrite 버퍼에 기록한 후 각 더티 페이지를 파일의 적당한 위치에 하나씩 랜덤 쓰기를 실행한다.
• 이는 실제로 'A'~'E' 더티 페이지가 정상적으로 기록되면 필요 없어진다. 데이터 파일의 쓰기가 중간에 실패한 경우에만 DoubleWriter 버퍼의 내용과 비교하여 복구에 사용된다.

3) 언두 로그

• 트랜잭션 격리 수준을 보장하기 위해 DML(INSERT, UPDATE, DELETE)로 변경되기 이전 버전의 데이터를 별도로 백업한다.
• 트랜잭션이 롤백되면 언두 로그에 백업해 둔 이전 버전의 데이터를 이용해 복구하며(트랜잭션 보장), 특정 격리 수준에 맞게 언두 로그에 백업해둔 데이터를 읽어서 반환한다.(격리 수준 보장)

4) 체인지 버퍼

• RDBMS에서 인덱스를 업데이트하는 작업은 랜덤하게 디스크를 읽는 작업이므로 테이블에 인덱스가 많다면 상당한 자원을 소모하게 된다.
• InnoDB는 변경해야 할 인덱스 페이지가 버퍼 풀에 있으면 바로 업데이트를 수행하지만, 그렇지 않고 디스크에서 읽어와 변경해야 한다면 이를 즉시 실행하지 않고, 이를 임시 공간에 저장해두고 바로 사용자에게 반환하는 형태로 성능을 향상시킨다. 이때 임시 메모리 공간을 체인지 버퍼라고 한다.

4) 리두 로그 및 리두 버퍼

• 서버가 비정상적으로 종료됐을 때 디스크에 기록되지 못한 데이터를 잃지 않게 해주는 안전 장치이다.
• 커밋됐지만 데이터 파일에 기록되지 않은 데이터는 단순히 리두 로그에 저장된 데이터를 데이터 파일에 다시 적용하면 된다.
• 롤백됐지만 데이터 파일에 이미 기록된 데이터는 언두 로그의 내용을 가져와 데이터 파일에 다시 적용한다. 이때, 그 변경이 커밋, 롤백, 트랜잭션 실행 중간 상태인지 파악하기 위해 리두 로그를 활용한다.

5) 어댑티브 해시 인덱스

• B-TREE 인덱스에서 특정 값을 찾기 위해서 루트 노드, 브랜치 노드, 리프 노드를 거쳐야 원하는 레코드를 읽을 수 있다. 이러한 BTREE 검색 시간을 줄이기 위해 어댑티브 해시 인덱스가 도입되었다.

• 수동으로 생성하는 인덱스가 아니라 InnoDB 스토리지 엔진에서 사용자가 자주 요청하는 데이터에 대해 자동으로 생성하는 인덱스이다.

• 해시 인덱스는 '인덱스 키 값'과 해당 인덱스 키 값이 저장된 '데이터 페이지 주소' 쌍으로 관리된다.

• 성능에 유리하지 않은 상황

  1. 디스크 읽기가 많은 경우
  2. 특정 패턴의 쿼리가 많은 경우(조인, LIKE 패턴)
  3. 매우 큰 데이터를 가진 테이블의 레코드를 폭넓게 읽는 경우

• 성능에 유리한 상황

  1. 디스크 데이터가 InnoDB 버퍼 풀 크기와 비슷한 경우(디스크 읽기가 많지 않은 경우)
  2. 동등 검색 조건(동등 비교와 IN 연산자)이 많은 경우
  3. 쿼리가 데이터 중에서 일부 데이터에만 집중되는 경우

어댑티브 해시 인덱스는 메모리(버퍼 풀) 내에서 접근하는 것을 더 빠르게 만드는 기능이기 때문에 디스크에서 읽어오는 경우가 빈번하다면, 아무런 도움이 되지 않는다는 것을 명심하자!

• 어댑티브 해시 인덱스 또한 메모리 공간을 추가로 차지하며, 생성 및 제거 시에 추가적인 작업이 필요하다. 따라서, 어댑티브 해시 인덱스의 도움을 많이 받을수록 테이블 삭제 또는 변경 작업은 더 치명적인 작업이 된다는 것이다.

참고 자료

• https://dev.mysql.com/doc/refman/8.4/en/pluggable-storage-overview.html
• RealMySQL 8.0

전체 코드 링크: https://github.com/ji-jjang/Learning/commit/9b080a5e2cdaea4fbad2cf6cbc2c1559ebc47052 특정 경쟁 조건을 만족시키기 위해 sleep으로 조정한 코드가 있습니다. 운영체제, 컴퓨터 사양, 실행 상태 등에 따라 쓰레드 스케줄링이 달라 다른 결과가 나올 수 있는 점 참고해 주세요.

병행 제어 기본 용어

1) 임계 영역

• 변수나 자료구조 같은 공유 자원에 접근하는 코드의 일부분

2) 경쟁 조건

• 여러 쓰레드가 거의 동시에 임계 영역에 접근하는 상황

3) 비결정적

• 경쟁 조건이 생긴다면 그 실행 결과가 각 쓰레드가 실행된 시점에 의존하므로 매번 프로그램의 결과가 다른 현상

4) 상호 배제

• 하나의 쓰레드가 임계 영역 내의 코드를 실행 중일 때는 다른 쓰레드가 실행할 수 없도록 하는 것

5) 락

• 상호 배제를 구현하기 위한 매커니즘으로 락을 획득한 쓰레드만 임계 영역에 접근할 수 있게 하는 것

예약 시스템 병행 제어 필요성

• 아래와 같이 30분 단위 예약 슬롯이 8개 존재하는 간단한 공간 예약 서비스를 생각해 보자.

• 유저가 앞에 슬롯부터 4개를 예약하면 슬롯의 상태는 아래와 같을 것이다.

• 사용자가 한 명이라면, 크게 문제가 없다. 하지만 실제로는 여러 사용자가 동시에 예약할 수 있는 슬롯을 보게 된다.

• 세 명의 사용자가 거의 동시에 4개의 슬롯(10:00 ~ 12:00)을 예약했다고 해보자. 그럼 무슨 일이 일어날까? 이 상황을 간단하게 자바 코드로 구현해 보자.

public class TimeSlot {

  private Long id;
  private LocalDateTime startTime;
  private LocalDateTime endTime;
  private Boolean isReserved;
  private Integer price;
}

public class Reservation {

  private Long id;
  private LocalDateTime startTime;
  private LocalDateTime endTime;
  private Integer price;
}

// 예약 서비스, 예약 생성 메서드
@Transactional
public Reservation createReservationRaceCondition(List<Long> timeSlotIds) {

  LocalDateTime startTime = null;
  LocalDateTime endTime = null;
  for (var timeSlotId : timeSlotIds) {

    TimeSlot timeSlot = getTimeSlotRaceCondition(timeSlotId);

    startTime = getStartTime(startTime, timeSlot.getStartTime());
    endTime = getEndTime(endTime, timeSlot.getEndTime());

    timeSlot.setIsReserved(true);
    timeSlotRepository.updateIsReservedTrue(timeSlot);
  }

  Reservation reservation = new Reservation(null, startTime, endTime, 10000);
  reservationRepository.save(reservation);

  return reservation;
}

private TimeSlot getTimeSlotRaceCondition(Long timeSlotId) {
  TimeSlot timeSlot = timeSlotRepository
    .findById(timeSlotId)
    .orElseThrow(
      () -> new RuntimeException(String.format("TimeSlot %d not found", timeSlotId)));

  if (timeSlot.getIsReserved()) {
    throw new RuntimeException("TimeSlot is reserved");
  }
  return timeSlot;
}

• 예약 생성 메서드는 해당 슬롯이 예약되었다면, 이미 예약된 슬롯이라는 메시지와 함께 에러를 발생시킨다. 그러면 두 개의 쓰레드를 생성하여, 예약 생성이 거의 동시에 일어나는 테스트 코드를 작성해 보자.

  @Test
  @DisplayName("예약 생성 경쟁 조건이 발생하여 같은 슬롯에 2개의 예약이 생성된다.")
  void createReservationRaceCondition() throws InterruptedException {
  
  List<Long> timeSlotIds = List.of(1L, 2L, 3L, 4L);
  
  ExecutorService executor = Executors.newFixedThreadPool(2);
  
  List<Reservation> reservations = Collections.synchronizedList(new ArrayList<>());
  
  Runnable task1 = () -> {
    Reservation reservation = reservationService.createReservationRaceCondition(timeSlotIds);
    reservations.add(reservation);
  };
  
  Runnable task2 = () -> {
    Reservation reservation = reservationService.createReservationRaceCondition(timeSlotIds);
    reservations.add(reservation);
  };
  
  executor.execute(task1);
  executor.execute(task2);
  
  executor.shutdown();
  executor.awaitTermination(5, TimeUnit.SECONDS);
  
  Assertions.assertThat(reservations.size()).isEqualTo(2);
  }

• 위 코드를 실행하면서 발생한 쿼리는 아래와 같다. ```java

1. thread2 슬롯 조회 (슬롯 예약되지 않은 상태)
2. thread1 슬롯 조회 (슬롯 예약되지 않은 상태)
3. thread1 슬롯 1, 2, 3, 4 UPDATE 및 예약 생성
4. thread2 슬롯 1, 2, 3, 4 UPDATE 및 예약 생성 ```

• 예약되지 않은 슬롯에 2개의 쓰레드가 거의 동시에 접근했기 때문에 2개의 예약이 생성된다. 이처럼 임계 영역에 경쟁 조건이 발생하는 상황은 상호 배제 수단인 락을 통해 해결할 수 있다.

낙관적 락 (Optimistic Lock)

• 낙관적 락은 데이터를 읽을 때 락을 걸지 않고, 데이터를 수정할 때 버저닝된 값의 비교를 통하여 충돌 여부를 판단한다. 처음 데이터를 읽을 때 버저닝 값과 수정할 때 버저닝 값이 일치하지 않는다면 충돌이 발생한 것이다. 충돌이 발생하면 롤백시킨다. 충돌이 빈번한 상황에서 낙관적 락을 사용한다면 계속해서 롤백될 수 있기 때문에 롤백 오버헤드가 락 오버헤드보다 더 클 수 있다.

<update id="updateIsReservedTrue">
  UPDATE time_slots_versioning
  SET is_reserved = TRUE,
      version     = version + 1
  WHERE id = #{id}
    AND version = #{version}
</update>

• 그럼, 낙관적 락을 사용해서 병행 제어를 해보자. 먼저 버저닝 필드를 추가한다.

  public class TimeSlotVersioning {
  
  private Long id;
  private LocalDateTime startTime;
  private LocalDateTime endTime;
  private Boolean isReserved;
  private Integer price;
  private Integer version;
  }
  

@Transactional public Reservation createReservationOptimistic(List timeSlotIds) {

List timeSlots = new ArrayList<>(); for (var timeSlotId : timeSlotIds) { TimeSlotVersioning timeSlot = timeSlotVersioningRepository .findById(timeSlotId) .orElseThrow( () -> new RuntimeException(String.format("TimeSlot %d not found", timeSlotId))); timeSlots.add(timeSlot); }

LocalDateTime startTime = null; LocalDateTime endTime = null; for (var timeSlot : timeSlots) { startTime = getStartTime(startTime, timeSlot.getStartTime()); endTime = getEndTime(endTime, timeSlot.getEndTime());

timeSlot.setIsReserved(true);
int rows = timeSlotVersioningRepository.updateIsReservedTrue(timeSlot);
if (rows == 0) {
  throw new RuntimeException("Optimistic reservation failed");
}

} Reservation reservation = new Reservation(null, startTime, endTime, 10000); reservationRepository.save(reservation);

return reservation; }

- 버저닝 값이 같으면 Where 조건으로 인해 row가 수정되지 않고, 애플리케이션에서 사용자 설정 예외를 발생시킨다. 4개의 슬롯을 100개의 쓰레드로 예약해 보자. 이미 눈치챘을 수도 있지만, 위 코드는 제대로 병행 제어를 할 수 없다.

```java
@Test
@DisplayName("낙관적 락을 사용할 때 조회한 슬롯들을 한 번에 수정하지 않으면 병행 제어가 되지 않는다.")
void createReservationOptimisticFailed() throws InterruptedException {

  List<Long> timeSlotIds = List.of(1L, 2L, 3L, 4L);

  ExecutorService executor = Executors.newFixedThreadPool(100);

  List<Reservation> reservations = Collections.synchronizedList(new ArrayList<>());

  for (int i = 0; i < 100; ++i) {
    executor.execute(
        () -> {
          try {
            Reservation reservation = reservationService.createReservationOptimistic(timeSlotIds);
            reservations.add(reservation);
            System.out.println("ThreadInfo: " + Thread.currentThread());
          } catch (Exception e) {
            System.err.println(Thread.currentThread().getName() + " failed: " + e.getMessage());
          }
        });
  }

  executor.shutdown();
  executor.awaitTermination(5, TimeUnit.SECONDS);

  System.out.println("reservations = " + reservations);
  Assertions.assertThat(reservations.size()).isEqualTo(1);
}

• 계속해서 충돌이 발생하면 예약이 0개가 되거나, 1개가 되어야 한다. 하지만 결과는 예상과 달리 예약이 10개가 생성된다. 이유는 4개의 슬롯을 부분적으로 업데이트하는 과정에서 여러 쓰레드가 수정 전의 상태를 보고 값을 덮어씌우는 현상이 발생하기 때문이다. 아래와 같이 한 번에 업데이트 해야 한다.

  <update id="bulkUpdateIsReservedTrue">
    UPDATE time_slots_versioning
    SET is_reserved = TRUE, version = version + 1
    WHERE id IN
    <foreach item="id" collection="list" open="(" separator="," close=")">
      #{id}
    </foreach>
    AND is_reserved = FALSE
  </update>

  @Transactional
  public Reservation createReservationOptimisticBulk(List<Long> timeSlotIds) {

    List<TimeSlotVersioning> timeSlots = new ArrayList<>();
    LocalDateTime startTime = null;
    LocalDateTime endTime = null;
    for (var timeSlotId : timeSlotIds) {
      TimeSlotVersioning timeSlot = timeSlotVersioningRepository
        .findById(timeSlotId)
        .orElseThrow(
          () -> new RuntimeException(String.format("TimeSlot %d not found", timeSlotId)));
      timeSlots.add(timeSlot);
      startTime = getStartTime(startTime, timeSlot.getStartTime());
      endTime = getEndTime(endTime, timeSlot.getEndTime());
    }

    int rowsUpdated = timeSlotVersioningRepository.bulkUpdateIsReservedTrue(timeSlotIds);
    if (rowsUpdated < timeSlotIds.size()) {
      throw new RuntimeException("Some TimeSlots failed to reserve due to optimistic locking");
    }

    Reservation reservation = new Reservation(null, startTime, endTime, 10000);
    reservationRepository.save(reservation);

    return reservation;
  }

• 위 코드를 100, 1000개의 쓰레드로 테스트해도 여러 개의 예약이 생성되지 않는 걸 확인할 수 있다.

이후 설명 할 비관적 쓰기 락 (Select For Update)에서는 조회와 동시에 락을 걸기 때문에 다른 쓰레드에서 조회 및 수정이 불가능하다. 이 경우 하나씩 업데이트해도 경쟁 조건이 발생하지 않는다.

비관적 락 (Pessimistic Lock)

• 비관적 락은 이러한 동시성 문제를 사전에 해결하기 위해 Lock을 걸어 다른 트랜잭션의 접근을 차단하는 방식이다. 충돌이 빈번하게 발생하는 상황에서 락을 획득한 쓰레드만 임계 영역에 접근할 수 있기에 효율적으로 동작한다.
• 크게 비관적 쓰기 락과 비관적 읽기 락으로 구분한다.

1.비관적 쓰기 락(Pessimisic Write Lock)

• Select For Update 명령어를 통해 조회 시 비관적 쓰기 락을 건다. 다른 트랜잭션은 해당 데이터에 대해 읽기 및 수정을 하지 못한다. 강력한 잠금 방식이며, 충돌이 빈번하지 않은 상황에서는 성능상 오버헤드가 크다. 4개의 슬롯을 조회하며 비관적 쓰기 락을 걸어보자.

<select id="findByIdsForUpdate" resultType="com.juny.locktest.TimeSlot">
  SELECT
    id,
    start_time AS startTime,
    end_time AS endTime,
    is_reserved AS isReserved,
    price
  FROM time_slots
  WHERE id IN
  <foreach item="id" collection="list" open="(" separator="," close=")">
    #{id}
  </foreach>
  FOR UPDATE
</select>

@Transactional
public Reservation createReservationPessimisticWriteLock(List<Long> timeSlotIds)
  throws InterruptedException {

  List<TimeSlot> timeSlots = timeSlotRepository.findByIdsForUpdate(timeSlotIds);

  for (var timeSlot : timeSlots) {
    if (timeSlot.getIsReserved()) {
      throw new RuntimeException("TimeSlot is reserved");
    }
  }

  LocalDateTime startTime = null;
  LocalDateTime endTime = null;
  for (TimeSlot timeSlot : timeSlots) {
    startTime = getStartTime(startTime, timeSlot.getStartTime());
    endTime = getEndTime(endTime, timeSlot.getEndTime());

    timeSlot.setIsReserved(true);
    timeSlotRepository.updateIsReservedTrue(timeSlot);
  }
  Reservation reservation = new Reservation(null, startTime, endTime, 10000);
  reservationRepository.save(reservation);

  return reservation;
}

• 100개, 1000개의 쓰레드를 실행해도 하나의 예약만 성공하는 것을 볼 수 있다.

  @Test
  void createReservationPessimisticWriteLock() throws InterruptedException {
  List<Long> timeSlotIds = List.of(1L, 2L, 3L, 4L);
  
  ExecutorService executor = Executors.newFixedThreadPool(100);
  
  List<Reservation> reservations = Collections.synchronizedList(new ArrayList<>());
  
  for (int i = 0 ; i < 100; ++i) {
    executor.execute(
      () -> {
        try {
          Reservation reservation =
            reservationService.createReservationPessimisticWriteLock(timeSlotIds);
          reservations.add(reservation);
        } catch (Exception e) {
          System.err.println(Thread.currentThread().getName() + " failed: " + e.getMessage());
        }
      });
  }
  executor.shutdown();
  executor.awaitTermination(5, TimeUnit.SECONDS);
  
  Assertions.assertThat(reservations.size()).isEqualTo(1);
  }

• 여기서 또 하나 중요한 것은 Select For Update가 호출되었을 때 다른 쓰레드가 대기하는지 확인하는 것이다. 아래처럼 출력문을 작성하고 2개의 쓰레드를 실행해보자.

@Transactional
public Reservation createReservationPessimisticWriteLock(List<Long> timeSlotIds)
  throws InterruptedException {

  System.out.println("비관적 쓰기 락 조회 전" + Thread.currentThread().getName());
  List<TimeSlot> timeSlots = timeSlotRepository.findByIdsForUpdate(timeSlotIds);
  System.out.println("비관적 쓰기 락 조회 후" + Thread.currentThread().getName());
  Thread.sleep(3000);

  ... 생략
}

• 로그로 실행 순서를 파악해보면, 비관적 쓰기 락이 걸리면 다른 쓰레드는 대기 중인 것을 확인할 수 있다. ```

1. 비관적 쓰기 락 조회 전pool-1-thread-1
2. 비관적 쓰기 락 조회 전pool-1-thread-2
3. thread-1 SELECT FOR UPDATE (쓰레드 2는 대기 중)
4. thread-1 개별 UPDATE 쿼리
5. thread-2 SELECT FOR UPDATE

2. 비관적 읽기 락(Pessimisic Read Lock)

• Select For Share, 비관적 읽기 락은 데이터를 읽는 동안 다른 트랜잭션이 해당 데이터를 수정하지 못하도록 락을 거는 방식이다.
• 예약 생성 메서드(슬롯을 조회 및 수정하고, 예약을 생성함)에서 슬롯을 읽어올 때 비관적 읽기 락을 적용하면 어떻게 될까?

  <select id="findByIdsForShare" resultType="com.juny.locktest.TimeSlot">
    SELECT
    id,
    start_time AS startTime,
    end_time AS endTime,
    is_reserved AS isReserved,
    price
    FROM time_slots
    WHERE id IN
    <foreach item="id" collection="list" open="(" separator="," close=")">
      #{id}
    </foreach>
    FOR SHARE
  </select>

  @Transactional
  public Reservation createReservationPessimisticReadLock(List<Long> timeSlotIds)
    throws InterruptedException {


    List<TimeSlot> timeSlots = timeSlotRepository.findByIdsForShare(timeSlotIds);

    for (var timeSlot : timeSlots) {
      if (timeSlot.getIsReserved()) {
        throw new RuntimeException("TimeSlot is reserved");
      }
    }

    LocalDateTime startTime = null;
    LocalDateTime endTime = null;
    for (TimeSlot timeSlot : timeSlots) {
      startTime = getStartTime(startTime, timeSlot.getStartTime());
      endTime = getEndTime(endTime, timeSlot.getEndTime());

      timeSlot.setIsReserved(true);
      timeSlotRepository.updateIsReservedTrue(timeSlot);
    }
    Reservation reservation = new Reservation(null, startTime, endTime, 10000);
    reservationRepository.save(reservation);

    return reservation;
  }

• 위와 같이 조회를 하고, 수정하는 쿼리에서 비관적 읽기 락을 사용하면 데드락이 발생한다.

  Exception in thread "pool-1-thread-2" org.springframework.dao.DeadlockLoserDataAccessException: 
  ### Error updating database.  Cause: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction

• 동작 과정을 살펴보면 쓰레드 1이 쓰기 락(배타적 락)을 가지고 있을 때 쓰레드 2가 배타적 락을 가지려고 하면 데드락이 발생한다. ```

1. Thread-1 슬롯 1,2,3,4 비관적 읽기 락
2. Thread-2 슬롯 1,2,3,4 비관적 읽기 락
3. Thread-1 슬롯 1,2,3,4 수정 (쓰기 락)
4. Thread-2 슬롯 1 수정하려고 할 때 DeadLock ```

• 이게 왜 데드락일까? Thread-1과 Thread2는 슬롯 1,2,3,4에 대해 공유 락(S)을 설정하고, 쓰레드 1이 수정할 때 쓰기 락(EX)은 쓰레드 2의 공유 락(S)을 해제하길 원하고, Thread-2가 슬롯 1을 수정(EX)하려고 할 때 Thread-1의 읽기 락(s)을 해제하길 원하니, 자원을 가진채 서로의 자원을 원하는 상황(Hold And Wait)이라 데드락이 발생한다.

• 이처럼 비관적 읽기 락을 사용하면, 데이터를 읽는 시점에 다른 트랜잭션이 해당 데이터를 수정하는 것을 막아주는 효과가 있다.

참고 자료

• https://f-lab.kr/insight/understanding-optimistic-and-pessimistic-locking
• https://dev.mysql.com/doc/refman/8.0/en/innodb-transaction-isolation-levels.html
• https://docs.spring.io/spring-framework/docs/4.1.5.RELEASE/spring-framework-reference/html/transaction.html
• https://dev.mysql.com/doc/refman/8.4/en/innodb-locking.html

1. 도커로 redis-cli 접속

• 아래 명령어를 통해 redis-cli 에 접근할 수 있다. docker run --name my-redis -d -p 6379:6379 redis docker exec -it my-redis redis-cli —-raw

2. SpringBoot 설정

build.gradle

implementation ("org.springframework.boot:spring-boot-starter-data-redis")

• 스프링 부트 2.X 부터는 lettuce를 기본 클라이언트로 선택하여 redis와 통신한다.
• TPS, CPU, Connection 개수, 응답속도 등에서 jedis 보다 성능이 좋다고 한다. 자세한 내용은 여기를 참고하자. https://jojoldu.tistory.com/418

레디스 인터페이스

• 레디스의 모든 기능을 이용하기 위해 RedisTemplate으로 사용할 수도 있고, 엔티티 중심으로 CRUD 연산을 쉽고 빠르게 할 수 있는 RedisRepository 있다. 또한, Spring Cache와 통합하여 어노테이션 기반으로 캐싱을 쉽게 적용할 수 있다.

1. RedisTemplate

• key, value 연산을 직접 할 수 있으며, 레디스 자료 구조에 따른 세밀한 제어가 필요한 경우에 사용한다.
• key, value 직렬화를 어떤 방식으로 할지 먼저 설정해 주어야 한다.

@Configuration
public class RedisConfig {

  @Bean
  public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
    RedisTemplate<String, Object> template = new RedisTemplate<>();

    template.setConnectionFactory(connectionFactory);

    template.setKeySerializer(new StringRedisSerializer());
    template.setValueSerializer(new GenericJackson2JsonRedisSerializer()); 

    template.setHashKeySerializer(new StringRedisSerializer());
    template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());

    return template;
  }
}

• Redis Hash 자료구조를 사용한다면 해시 자료구조에 대한 직렬화 방법도 등록해준다. GenericJackson2JsonRedisSerializer를 사용하면 값을 Json 형태로 직렬화할 수 있다.

@Getter
public class Product {
  private String name;
  private int price;
  private int quantity;

  public Product(
      @JsonProperty("name") String name,
      @JsonProperty("price") int price,
      @JsonProperty("quantity") int quantity) {
    this.name = name;
    this.price = price;
    this.quantity = quantity;
  }
  ...
}

• Setter를 열어두지 않는다면 @JsonProperty로 매핑 정보를 알려주어야 한다.

Redis 자료 구조 제어하기

• 키의 추가는 자료구조에 따라 아래 메서드를 사용한다.

1) String

(1) 스프링

@Autowired RedisTemplate redisTemplate;

@Test
void setString() {
  redisTemplate.delete("juny");
  redisTemplate.delete("지니");

    // 키 쓰기
  redisTemplate.opsForValue().set("juny", "쭈니야");
  redisTemplate.opsForValue().set("지니", "쭈니");

  Product product1 = new Product("Monitor", 120, 15);
  redisTemplate.opsForValue().set("product:1", product1);

    // 키 읽기
  String value1 = (String) redisTemplate.opsForValue().get("juny");
  String value2 = (String) redisTemplate.opsForValue().get("지니");
  Product value3 = (Product) redisTemplate.opsForValue().get("product:1");

  assertThat(value1).isEqualTo("쭈니야");
  assertThat(value2).isEqualTo("쭈니");
  assertThat(value3).isEqualTo(product1);
}

(2) 레디스

• 한글로 된 Key, Value 값을 redis-cli에서 이스케이프 형식으로 보고 싶지 않다면, docker exec -it my-redis redis-cli —-raw 로 접속한다.

127.0.0.1:6379> get juny
"쭈니야"

127.0.0.1:6379> get 지니
"쭈니"

127.0.0.1:6379> get product:1
{"@class":"redis.RedisTemplateTest$Product","name":"Monitor","price":120,"quantity":15}

2) List

(1) 스프링

@Test
void setList() {

  redisTemplate.delete("user");

    // 키 쓰기
  redisTemplate.opsForList().leftPush("user", "1");
  redisTemplate.opsForList().leftPush("user", "2");
  redisTemplate.opsForList().leftPush("user", "3");

    // 키 읽기
  List<String> list = redisTemplate.opsForList().range("user", 0, -1);
  assertThat(list).containsExactly("3", "2", "1");
}

(2) 레디스

127.0.0.1:6379> LRANGE user 0 -1
"3"
"2"
"1"

3) Set

(1) 스프링

@Test
void setSet() {
  redisTemplate.delete("numbers");

    // 키 쓰기
  redisTemplate.opsForSet().add("numbers", "1", "1", "2", "2", "3", "4", "5", "5");

    // 키 읽기
  Set<Object> numbers = redisTemplate.opsForSet().members("numbers");

  assertThat(numbers).contains("1", "2", "3", "4", "5");
  assertThat(numbers).hasSize(5);
}

(2) 레디스

127.0.0.1:6379> SMEMBERS numbers
"1"
"2"
"3"
"4"
"5"

4) Sorted Set

(1) 스프링

@Test
void setSortedSet() {
  redisTemplate.delete("scores");

    // 키 쓰기
  redisTemplate.opsForZSet().add("scores", "korean", 100);
  redisTemplate.opsForZSet().add("scores", "science", 80);
  redisTemplate.opsForZSet().add("scores", "math", 90);

    // 키 읽기
  Set scores = redisTemplate.opsForZSet().rangeWithScores("scores", 0, -1);

  scores.forEach(System.out::println);

  assertThat(scores).hasSize(3);

  assertThat(scores)
    .extracting("value")
    .containsExactly("science", "math", "korean");

  assertThat(scores)
    .extracting("score")
    .containsExactly(80.0, 90.0, 100.0);
}

(2) 레디스

127.0.0.1:6379> ZRANGE scores 0 -1 WITHSCORES
"science"
80
"math"
90
"korean"
100

5) Hash

(1) 스프링

@Test
void setHash() {

    // 키 쓰기
  redisTemplate.opsForHash().put("product", "name", "상품 1");
  redisTemplate.opsForHash().put("product", "price", "50000");
  redisTemplate.opsForHash().put("product", "quantity", "30");

    // 키 읽기
  Map<String, String> map = redisTemplate.opsForHash().entries("product");

  assertThat(map).containsEntry("name", "상품 1");
  assertThat(map).containsEntry("price", "50000");
  assertThat(map).containsEntry("quantity", "30");
}

(2) 레디스

127.0.0.1:6379> HGETALL product
name
"상품 1"
price
"50000"
quantity
"30"

2. RedisRepository

• RedisRepository 인터페이스를 이용하면, Spring Data JPA와 유사하게 레디스 CRUD를 할 수 있다.
• 기본적으로 Redis Hash 자료구조를 사용하여 저장한다.

(1) 스프링

@RedisHash(value = "RefreshToken", timeToLive = 60) // 초 단위
@Getter
public class RefreshToken {

  @Id 
  private Long id;
  private Long userId;
  private String refresh;

  public RefreshToken(Long id, Long userId, String refresh) {
    this.id = id;
    this.userId = userId;
    this.refresh = refresh;
  }
}

public interface RefreshTokenRepository extends CrudRepository<RefreshToken, Long> {}

@Autowired
private RefreshTokenRepository refreshTokenRepository;

@Test
void saveRefreshToken() {

  RefreshToken refreshToken = new RefreshToken(1L, 1L, "asdkfaisxk291j28");

    // 키 쓰기
  RefreshToken savedToken = refreshTokenRepository.save(refreshToken);

    // 키 읽기
  RefreshToken findToken = refreshTokenRepository
    .findById(1L)
    .orElseThrow(() -> new RuntimeException("Refresh token not found"));

  assertThat(savedToken.getId()).isEqualTo(findToken.getId());
  assertThat(savedToken.getRefresh()).isEqualTo(findToken.getRefresh());
  assertThat(savedToken.getUserId()).isEqualTo(findToken.getUserId());
}

(2) 레디스

127.0.0.1:6379> HGETALL RefreshToken:1
_class
com.juny.jspboardwithmybatis.redis.RefreshToken
id
1
refresh
asdkfaisxk291j28
userId
1

3. Spring Cache

• Spring Cache를 Redis와 통합하여 redis의 빠른 읽기, 쓰기 성능을 이용하는 방법이다.

• 먼저, Spring Cache 의존성을 추가한다.

  gradle

  implementation 'org.springframework.boot:spring-boot-starter-cache'

• 레디스 캐시 매니저를 통해 Redis를 캐시 저장소로 사용하도록 설정한다.

@Configuration
@EnableCaching
public class RedisConfig {

@Bean
public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {

  RedisCacheConfiguration defaultConfig = RedisCacheConfiguration.defaultCacheConfig()
    .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
    .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()))
    .entryTtl(Duration.ofMinutes(5));

  return RedisCacheManager.builder(connectionFactory)
    .cacheDefaults(defaultConfig)
    .build();
}

• Spring Cache에 공통 유효기간 외 특정 키에 대한 유효기간을 별도로 지정하고 싶다면 아래와 같이 설정한다.

    @Bean
    public RedisCacheManager cacheManager(RedisConnectionFactory connectionFactory) {
  
      RedisCacheConfiguration defaultConfig = RedisCacheConfiguration.defaultCacheConfig()
        .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
        .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()))
        .entryTtl(Duration.ofMinutes(5));
  
      Map<String, RedisCacheConfiguration> cacheConfigurations = new HashMap<>();
      cacheConfigurations.put("boardDetails", defaultConfig.entryTtl(Duration.ofMinutes(1)));
  
      return RedisCacheManager.builder(connectionFactory)
        .cacheDefaults(defaultConfig)
        .withInitialCacheConfigurations(cacheConfigurations)
        .build();
    }

• DB 쿼리를 사용하지 않고, 캐시된 데이터를 재사용하고 싶다면 서비스에서 캐싱을 적용한다.

@Cacheable(cacheNames = "boardDetails", key = "#id")
public ResBoardDetail getBoard(Long id) {

  Map<String, Object> board = boardMapper.findBoardDetailById(id);

  if (board == null) {
    throw new BoardNotFoundException(ErrorCode.BOARD_NOT_FOUND);
  }

  return BoardDTOConverter.convertToResBoardDetail(id, board);
}

• 키를 삭제하고 싶다면, @CacheEvict를 이용해 삭제할 수 있다.

@CacheEvict(cacheNames = "boardDetails", key = "#reqBoardDelete.boardId")
@Transactional
public void deleteBoard(ReqBoardDelete reqBoardDelete) {

  for (var comment : reqBoardDelete.getComments()) {
    commentMapper.deleteCommentById(comment.getId());
  }
  for (var att : reqBoardDelete.getAttachments()) {
    attachmentMapper.deleteAttachmentById(att.getId());
  }
  for (var image : reqBoardDelete.getBoardImages()) {
    boardImageMapper.deleteBoardImageById(image.getId());
  }
  boardMapper.deleteBoardById(reqBoardDelete.getBoardId());
}

레디스 적용 사례 찾아보기

• 레디스를 캐시로 사용했을 때, 성능 개선 효과가 크려면 아래 조건에 해당해야 한다.

1. 검색하는 시간이 오래 걸리거나, 매번 계산을 통해 데이터를 가져와야 하는 경우

2. 데이터가 잘 변하지 않는 경우

3. 자주 검색되는 데이터인 경우

• 위 경우에 해당하면, Redis를 도입했을 때 성능효과가 클 수 있다. 여태까지 프로젝트를 진행하며 적용할 수 있는 부분을 찾아보면…

1. 상품 상세페이지, 게시판 상세 페이지 등

• 도메인 특성상 해당 페이지가 자주 조회되고, 잘 변경되지 않는다면 레디스 적용하기에 적합하다.
• 좋아요, 조회수 등도 매번 반영하기보다 캐시에 저장해놓고, 일정 주기로 DB에 반영한다면 성능 효과가 클 것으로 보인다.

2. 리프레시 토큰

• 사용자가 로그인할 때마다 해당 토큰이 유효한지 검증해야 하므로 레디스 적용하기 적합하다.
• MySQL을 토큰 저장소로 사용할 경우 주기적으로 유효기간이 지난 토큰을 배치나 스케줄러로 삭제해 주어야 하지만, 레디스는 TTL을 통해 유효기간이 만료된 키를 자동으로 삭제할 수 있다.
• 최근 검색 목록도 자주 조회되고, 오래된 데이터를 삭제할 때 sorted set을 이용하면 쉽고 효율적으로 구현할 수 있다.

3. 실시간 시스템 (알림, 랭킹, 지도 등)

• 랭킹 시스템(ex: 일간, 주간, 월간 가장 인기 있는 상품 10개)
  • sorted set을 이용하면 정렬이 되어있기 때문에 매번 계산을 할 필요가 없다. 물론 레디스에 매번 데이터를 추가해주어야 하는데, 이와 관련해서 아래 글을 참고하면 좋을 것이다. https://gosuda.org/ko/blog/posts/improving-responsiveness-with-redis-client-side-caching-zb711e502
• 실시간 알림(레디스 메시징 큐)
• 지도
  • geo set을 이용해 경도와 위도 연산 효율적으로 처리할 수 있다.

다루지 않았지만, 중요한 내용들

• 레디스 자료 구조, 캐시 전략
• 레디스 백업 AOF, RDB (가용성)
• 레디스 복제 (가용성)
• 페일오버, 센티널 (가용성)
• 레디스 클러스터 (확장성)

• Apache에서 제공하는 성능 테스트 도구로, 오픈소스 프로젝트다.
• *어떤 기술을 도입하기 전에 해당 기술이 왜 필요한지, 해당 기술로 인해 어떠한 개선이 있는지 정량적으로 말할 수 있어야 한다. *
• ex) 레디스를 왜 도입하셨나요?
  • 빠르다고 해서요(X)
  • 빨라서요(X) -> 얼마나 빠른가요? (...?)
• Jmeter를 사용하면 아래처럼 답변할 수 있다.

  <Redis 캐시 미적용>
  

평균 응답 시간: 19ms 처리량(Throughput): 3516.2 요청/초 전송 데이터량(Received KB/sec): 18964.61 KB/sec

<Redis 캐시 적용>

평균 응답 시간: 5ms 처리량(Throughput): 7052.2 요청/초 전송 데이터량(Received KB/sec): 38037.87 KB/sec

```

• 특정 쿼리에서 다수 엔티티를 조인할 때 발생하는 비용이 크다고 생각했고, 변경이 빈번히 일어나지 않은 데이터라 캐시 효율성이 높아 레디스를 도입하게 되었습니다.
• 레디스를 도입한 결과 평균 응답 시간이 약 70% 감소(19ms → 5ms)하여 사용자 경험이 좋아지고, 처리량은 2배(3516.2 요청/초 → 7052.2 요청/초)로 증가하여 성능이 개선되었습니다. 로컬 환경이 아닌, 실제 운영 환경에서 테스트했을 경우 네트워크 통신 비용이 절약되어 더 큰 성능 개선 효과를 볼 수 있습니다.

그럼, Jmeter를 어떻게 사용해야 하는지 알아보자!

Jmeter 주요 구성요소

1. Test Plan

• JMeter 테스트 최상위 요소로, 성능 테스트 모든 설정을 포함한다.
• Test Plan은 항상 하나이고, 테스트를 구성하는 모든 데이터가 .jmx 파일로 저장된다.
• 테스트 환경에서 공통으로 사용할 환경변수를 등록해서 사용할 수 있다.

• 기본적으로 여러 개의 Thread Group은 병렬로 실행된다. 하지만, Run Thread Groups consecutively선택하면 위에서 나열된 대로 순차적으로 실행할 수 있다. (ex: 로그인 테스트 후에 상품 목록 조회 테스트)

2. Thread Group

• 실제 요청을 생성하는 가상 사용자(스레드)를 정의한다.
  • 아래 설정은 100개의 스레드가 1초 안에 순차적으로 실행하면서, 각 작업(샘플러)을 10번 반복한다고 이해하면 된다.

• 요청(Sameplers), 설정(Config Elements), 타이머(Timers), 어설션(Assertions) 등을 추가할 수 있다.

1) setUp Thread Group

• 테스트 전에 필요한 준비 작업을 수행한다. 인증 토큰 발급이나 테스트에 사용할 데이터를 추가하는 등의 작업을 할 수 있다.
• 가장 먼저 실행되는 스레드 그룹이다.

2) tearDown Thread Group

• 테스트가 완료된 후 정리 작업을 수행한다. 토큰 무효화, 테스트 데이터 삭제 등의 작업을 할 수 있다.
• 모든 스레드 그룹이 실행이 완료된 후 실행되는 스레드 그룹이다.

3) Open Model Thread Group

• 일반 스레드 그룹에서는 사용자 수, ram-up 기간, loop count 만을 지정할 수 있다.
• 해당 스레드 그룹을 사용하면 요청 빈도(rate), 무작위 도착 패턴(random arrival), 특정 시간 동안 요청 패턴(pause) 등의 트래픽 설정이 가능하다. 즉, 좀 더 현실적이고 유연한 테스트가 가능하다.

3. Sampler

• 특정 작업을 수행하거나 요청을 보내는 역할을 담당하는 구성요소다.
• HTTP뿐만 아니라 아래와 같이 JDBC, FTP 등 다양한 요청을 생성할 수 있다.

• HTTP 요청을 선택했다면, 해당 프로토콜 내용을 프로젝트 환경에 맞게 구성한다.

4. 로직 컨트롤러

• 로직 컨트롤러를 통해 특정 시나리오 기반 테스트를 구성할 수 있다. 예를 들어, If Controller는 응답 데이터의 ${status} 필드가 ‘success’라면 특정 작업을 처리할 수 있다.

5. Listener

• 테스트 결과를 시각적으로 보여주는 역할을 하는 구성요소다.

• View Results Tree는 각 요청과 응답 데이터를 상세하게 보여준다.

• Summary Report는 테스트 결과의 요약 데이터를 보여준다.
  • 위에서 100개의 스레드에 게시판 목록 조회 작업을 10번 시킨 결과이다.

• 1000번의 요청이 실행되었고, 모든 요청의 평균 응답시간은 2ms, 가장 짧은 것은 1ms, 가장 긴 것은 48ms, 실패비율은 0.1%이며 처리량은 초당 약 975개의 요청을 처리한다는 것을 확인할 수 있다.

• 추가로 Aggreagate Report 는 테스트 집계 결과를 보여주고, Backend Listener는 결과 데이터를 외부 모니터링 시스템으로 전송할 수 있다.

• 빌더 패턴이란 객체의 생성 과정을 별도의 빌더 객체로 분리하여 유연하게, 단계별로 객체를 생성할 수 있게 하는 패턴이다.

필요한 이유

• 객체를 생성할 때 초기화할 필드가 많으면 생성자 매개변수가 많아져 가독성이 떨어진다. 심지어 같은 타입의 필드를 초기화할 때 다른 생성자를 실수로 호출할 위험도 있다.

  public User(String name, int age, String email, String phoneNumber, String address ...)
  

public User(String name, int age, String referer, String email, String PhoneNumber, String Address ...)

- 생성자 매개변수에 **의존성이 생기는 문제, 가독성이 떨어지는 문제**를 해결하기 위해 빌더를 사용할 수 있다.

> 전체 코드: https://github.com/ji-jjang/Learning/commit/b0c39e36b8f339079d5539d0d3399489ef9dbfec

```java
public class User {

    private String name;
    ...

    private class User() {} // private으로 제한할 수도 있음

    public static class Builder {
        private String name;
        ...

        public Builder(String name) {
            this.name = name;
        }

        public Builder(int age) {
            this.age = age;
        }
        ...
        public User build() {
            return new User(this);
        }
    }
}

public static void main(String[] args) {
    User user = new Builder("홍길동")
      .age(20)
      .email("juny@gmail.com")
      .address("seoul")
      .build();
}

• Setter를 열어두지 않기 위해 Builder 클래스를 내부 클래스로 선언한다.

빌더 패턴 구조

• 위와 같이 빌더를 구성해도 되지만, Director를 두어 좀 더 유연하게 사용할 수도 있다. 유키 히로시 디자인 패턴 책에 나온 구성을 참고해 보자.

  1. Builder 인터페이스

  public interface Builder {
  public void makeTitle(String title);
  public void makeString(String string);
  public void makeItems(String[] items);
  public void close();
  }

  2. Director (객체 생성 과정 관리)

  public class Director {
  private Builder builder;
  public Director(Builder builder) {
    this.builder = builder;
  }
  public void construct() {
    builder.makeTitle("Greeting");
    builder.makeString("일반적인 인사");
    builder.makeItems(new String[]{
      "How are you?",
      "Hello.",
      "Hi."
    });
    builder.makeString("시간대별 인사");
    builder.makeItems(new String[]{
      "Good morning.",
      "Good afternoon.",
      "Good evening.",
    });
    builder.close();
  }
  }

3. TextBuilder (Builder Interface 구현)

public class TextBuilder implements Builder {
  private StringBuilder sb = new StringBuilder();
  @Override
  public void makeTitle(String title) {
    sb.append("===============================\n");
    sb.append("[");
    sb.append(title);
    sb.append("]\n\n");
  }
  @Override
  public void makeString(String str) {
    sb.append("*");
    sb.append(str);
    sb.append("\n\n");
  }
  @Override
  public void makeItems(String[] items) {
    for (String s : items) {
      sb.append(".");
      sb.append(s);
      sb.append("\n");
    }
    sb.append("\n");
  }
  @Override
  public void close() {
    sb.append("================================\n");
  }
  public String getTextResult() {
    return sb.toString();
  }
}

4. HTML Builder(Builder Interface 구현)

public class HTMLBuilder implements Builder {
  private String filename = "untitled.html";
  private StringBuilder sb = new StringBuilder();
  @Override
  public void makeTitle(String title) {
    filename = title + ".html";
    sb.append("<!DOCTYPE html>\n");
    sb.append("<html>\n");
    sb.append("<head>\n\t<title>");
    sb.append(title);
    sb.append("</title>\n");
    sb.append("\t<meta charset=\"UTF-8\">\n");
    sb.append("</head>\n");
    sb.append("<body>\n");
    sb.append("<h1>");
    sb.append(title);
    sb.append("</h1>\n\n");
  }
  @Override
  public void makeString(String str) {
    sb.append("<p>");
    sb.append(str);
    sb.append("</p>\n\n");
  }
  @Override
  public void makeItems(String[] items) {
    sb.append("<ul>\n");
    for (String s: items) {
      sb.append("<li>");
      sb.append(s);
      sb.append("</li>\n");
    }
    sb.append("</ul>\n\n");
  }
  @Override
  public void close() {
    sb.append("</body>");
    sb.append("</html>\n");
    try {
      Writer writer = new FileWriter(filename);
      writer.write(sb.toString());
      writer.close();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
  public String getHTMLResult() {
    return filename;
  }
}

5. Main 함수

public class Main {
  public static void main(String[] args){

    TextBuilder textbuilder = new TextBuilder();
    Director director = new Director(textbuilder);
    director.construct();
    String result = textbuilder.getTextResult();
    System.out.println(result);

    HTMLBuilder htmlbuilder = new HTMLBuilder();
    director = new Director(htmlbuilder);
    director.construct();
    String filename = htmlbuilder.getHTMLResult();
    System.out.println("HTML파일 " + filename + "이 작성되었습니다.");
  }
}

• Director는 빌더를 사용하여 객체 생성 과정을 관리하고, Builder 인터페이스를 구현한 빌더들은 자신의 목적에 맞는 객체를 생성하는 것을 볼 수 있다.

순수 JDBC에서 게시글 검색

• 등록일과 제목, 내용, 작성자를 입력하면 게시글을 검색해서 게시글을 가져오는 게시판을 생각해보자. jdbc를 사용한다면 아래와 같은 코드를 작성할 수 있다.

전체 코드: https://github.com/ji-jjang/ebrainsoft/tree/main/JspBoard

  @Override
  public List<Board> getBoardSearchList(int page, Map<String, String> searchConditions) {

    List<Board> boards = new ArrayList<>();

    StringBuilder sql =
        new StringBuilder(
            """
        SELECT
          b.id, b.title, b.view_count, b.created_at, b.updated_at
        FROM
          boards b
        LEFT JOIN
          categories c
        ON
          b.category_id = c.id
        """);

    int limit = Constants.BOARD_LIST_PAGE_SIZE;
    int offset = (page - 1) * limit;

    boolean hasWhere = false;

    if (searchConditions.containsKey(Constants.START_DATE)
        && searchConditions.containsKey(Constants.END_DATE)) {
      sql.append("WHERE b.created_at BETWEEN ? AND ? ");
      hasWhere = true;
    }

    if (searchConditions.containsKey(Constants.CATEGORY)) {
      String connector = hasWhere ? "AND" : "WHERE";
      sql.append(connector).append(" c.name = ? ");
    }

    if (searchConditions.containsKey(Constants.KEYWORD)) {
      String connector = hasWhere ? "AND" : "WHERE";
      sql.append(connector).append(" (b.title LIKE ? OR b.created_by LIKE ? OR b.content LIKE ?) ");
    }
    sql.append(" ORDER BY b.created_at DESC ");
    sql.append(" LIMIT ? OFFSET ?");

    try (Connection conn = DriverManagerUtils.getConnection();
        PreparedStatement pstmt = conn.prepareStatement(sql.toString())) {

      int index = 0;

      if (searchConditions.containsKey(Constants.START_DATE)
          && searchConditions.containsKey(Constants.END_DATE)) {
        pstmt.setString(
            ++index, searchConditions.get(Constants.START_DATE) + Constants.START_DATE_START_TIME);
        pstmt.setString(
            ++index, searchConditions.get(Constants.END_DATE) + Constants.END_DATE_END_TIME);
      }

      if (searchConditions.containsKey(Constants.CATEGORY)) {
        pstmt.setString(++index, searchConditions.get(Constants.CATEGORY));
      }

      if (searchConditions.containsKey(Constants.KEYWORD)) {
        String keyword =
            Constants.PERSENT_SIGN
                + searchConditions.get(Constants.KEYWORD)
                + Constants.PERSENT_SIGN;
        pstmt.setString(++index, keyword);
        pstmt.setString(++index, keyword);
        pstmt.setString(++index, keyword);
      }

      pstmt.setInt(++index, limit);
      pstmt.setInt(++index, offset);

      try (ResultSet rs = pstmt.executeQuery()) {
        while (rs.next()) {
          boards.add(
              new Board(
                  rs.getLong(Constants.ID_COLUMN),
                  rs.getString(Constants.TITLE_COLUMN),
                  rs.getString(Constants.CONTENT_COLUMN),
                  rs.getString(Constants.PASSWORD_COLUMN),
                  rs.getInt(Constants.VIEW_COUNT_COLUMN),
                  rs.getTimestamp(Constants.CREATED_AT_COLUMN).toLocalDateTime(),
                  rs.getString(Constants.CREATED_BY_COLUMN),
                  rs.getTimestamp(Constants.UPDATED_AT_COLUMN) != null
                      ? rs.getTimestamp(Constants.UPDATED_AT_COLUMN).toLocalDateTime()
                      : null,
                  rs.getLong(Constants.CATEGORY_ID_COLUMN)));
        }
      }
    } catch (SQLException | ClassNotFoundException e) {
      throw new RuntimeException(e);
    }
    return boards;
  }

• 복잡해보일 수 있지만, 조건에 따라 쿼리를 붙이는 부분만 이해하면 된다. QueryBuilder를 만들어 빌드 패턴을 적용해 볼 수 있지 않을까? 여기서 BoardDAO는 위 구조에서 Director 역할을 하며, 객체 생성 책임(쿼리)을 지니는 QueryBuilder를 만들어보자.

  public class QueryBuilder {
  
  private final StringBuilder sql;
  private final List<Object> parameters;
  
  private QueryBuilder(String baseQuery) {
    this.sql = new StringBuilder(baseQuery);
    this.parameters = new ArrayList<>();
  }
  
  public static QueryBuilder create(String baseQuery) {
    return new QueryBuilder(baseQuery);
  }
  
  public QueryBuilder addCondition(String condition, Object... params) {
    if (!Objects.isNull(condition)) {
      sql.append(parameters.isEmpty() ? " WHERE " : " AND ").append(condition);
      if (!Objects.isNull(params)) {
        parameters.addAll(Arrays.asList(params));
      }
    }
    return this;
  }
  
  public QueryBuilder orderBy(String orderBy) {
    if (!Objects.isNull(orderBy)) {
      sql.append(" ORDER BY ").append(orderBy);
    }
    return this;
  }
  
  public QueryBuilder limitOffset(int limit, int offset) {
    sql.append(" LIMIT ? OFFSET ?");
    parameters.add(limit);
    parameters.add(offset);
    return this;
  }
  
  public String buildQuery() {
    return sql.toString();
  }
  
  public List<Object> getParameters() {
    return parameters;
  }
  }

  @Override
  public List<Board> getBoardList(int page, Map<String, String> searchConditions) {
  
    List<Board> boards = new ArrayList<>();
  
    String baseQuery = """
        SELECT
          b.id, b.title, b.content, b.view_count, b.created_at, b.created_by, b.updated_at, b.category_id
        FROM
          boards b
        LEFT JOIN
          categories c ON b.category_id = c.id
        """;
  
    int limit = Constants.BOARD_LIST_PAGE_SIZE;
    int offset = (page - 1) * limit;
  
    QueryBuilder queryBuilder = QueryBuilder.create(baseQuery);
  
    String startDate = searchConditions.getOrDefault(Constants.START_DATE, TimeFormatterUtils.getDefaultStartDate()) + Constants.START_DATE_START_TIME;
    String endDate = searchConditions.getOrDefault(Constants.END_DATE, TimeFormatterUtils.getDefaultEndDate()) + Constants.END_DATE_END_TIME;
    queryBuilder.addCondition("b.created_at BETWEEN ? AND ?", startDate, endDate);
  
    if (searchConditions.containsKey(Constants.CATEGORY)) {
      queryBuilder.addCondition("c.name = ?", searchConditions.get(Constants.CATEGORY));
    }
  
    if (searchConditions.containsKey(Constants.KEYWORD)) {
      String keyword = Constants.PERSENT_SIGN + searchConditions.get(Constants.KEYWORD) + Constants.PERSENT_SIGN;
      queryBuilder.addCondition("(b.title LIKE ? OR b.created_by LIKE ? OR b.content LIKE ?)",
        keyword, keyword, keyword);
    }
  
    queryBuilder.orderBy("b.created_at DESC").limitOffset(limit, offset);
  
    String sql = queryBuilder.buildQuery();
    List<Object> parameters = queryBuilder.getParameters();
  
    try (Connection conn = DriverManagerUtils.getConnection();
      PreparedStatement pstmt = conn.prepareStatement(sql)) {
  
      for (int i = 0; i < parameters.size(); i++) {
        pstmt.setObject(i + 1, parameters.get(i));
      }
  
      try (ResultSet rs = pstmt.executeQuery()) {
        while (rs.next()) {
          boards.add(
            new Board(
              rs.getLong(Constants.ID_COLUMN),
              rs.getString(Constants.TITLE_COLUMN),
              rs.getString(Constants.CONTENT_COLUMN),
              null,
              rs.getInt(Constants.VIEW_COUNT_COLUMN),
              rs.getTimestamp(Constants.CREATED_AT_COLUMN).toLocalDateTime(),
              rs.getString(Constants.CREATED_BY_COLUMN),
              rs.getTimestamp(Constants.UPDATED_AT_COLUMN) != null
                ? rs.getTimestamp(Constants.UPDATED_AT_COLUMN).toLocalDateTime()
                : null,
              rs.getLong(Constants.CATEGORY_ID_COLUMN)));
        }
      }
    } catch (SQLException | ClassNotFoundException e) {
      throw new RuntimeException(e);
    }
    return boards;
  }

• 예외 처리를 빼놓고 보면, 큰 차이가 없어 보이지만 조건을 처리하는 로직을 Builder가 담당하니 단순해진다. 이렇게 QueryBuilder를 한 번 만들어두면 검색 조건이 있는 다른 쿼리에 대해서 적용할 수 있다. 전체적으로 쿼리를 붙이는 로직이 QueryBuilder에 집중되니 DAO나 Service 계층은 단순해진다.

• 만약에 어떤 시스템에선 JDBC가 아니라 QueryDsl을 사용한다면? 위의 예시에서 나온 구조로 확장하여 사용할 수도 있을 것이다.

  public class SQLQueryBuilder implements QueryBuilder {
  }
  

public class QueryDSLQueryBuilder implements QueryBuilder { }

- 물론 QueryDSl 자체가 타입 안전한 쿼리 빌더이기 때문에 실제로 사용 가능한 구조는 아니지만 이렇게 생각해볼 수도 있다.

커맨드 패턴이란

• 커맨드 패턴이란 요청을 객체의 형태로 캡슐화하여 사용자가 보낸 요청을 나중에 이용할 수 있도록 매서드 이름, 매개변수 등 요청에 필요한 정보를 저장 또는 로깅, 취소할 수 있게 하는 패턴이라고 한다. (위키백과)
• 쉽게 말해, 실행할 작업(명령)을 객체로 만들어 호출자와 수신자의 의존성을 낮추는 것이다.

필요한 이유

• JPA를 사용하지 않고 서블릿을 이용해 게시판을 구현해보는 예시를 생각해보자. HttpServlet을 상속받는 BoardServlet이 앞단에서 /boards/ 로 시작하는 요청을 처리한다고 할 때, 아래 코드를 쉽게 떠올릴 수 있다.

  @WebServlet(urlPatterns = "/boards/*")
  public class BoardServlet extends HttpServlet {
  
  @Override
  protected void service(HttpServletRequest request, HttpServletResponse response)
      throws ServletException, IOException {
  
  if (requestURI.equals("/boards/free")) { // 게시판 목록
    viewListBoard(request, response);
  } else if (requestURI.equals("/boards/free/boardId")) { // 게시글 상세보기
    viewDetailBoard(request, response);
  } else if (requestURI.equals("/board/free/write")) { // 게시글 생성
    createBoard(request, response);
  } else if (requestURI.equals("/board/free/update")) { // 게시글 수정
    updateBoard(request, response);
  } else if ...
  {

• 특정 기능(게시글 삭제)이 추가될 때마다 if 문이 추가되어야 한다. Service() 메서드가 단순히 HTTP 요청을 처리하는 게 아닌 어떤 컨트롤러를 호출해야 하는지 결정하여 단일 책임 원칙에 위배된다.

• 커맨드 패턴을 적용하면 아래와 같이 요청을 캡슐화하고, 요청을 실행할 객체가 게시판 생성 객체라면 게시판 생성을, 게시판 수정 객체라면 게시판 수정을 처리하게 할 수 있다.

  protected void service(HttpServletRequest request, HttpServletResponse response)
      throws ServletException, IOException {
  
    String requestURI = request.getRequestURI();
  
    // 컨트롤러를 찾는 책임은 ControllerResolver 에게 위임
    BoardController controller = controllerResolver.resolveController(requestURI);
  
    if (controller == null) {
      throw new RuntimeException(ErrorMessage.NO_HANDLER_MSG + request.getRequestURI());
    }
    // 단순히 컨트롤러를 실행하는 책임
    controller.execute(request, response);
  }

커맨드 패턴 구조

전체 코드: https://github.com/ji-jjang/ebrainsoft/tree/main/JspBoard

• 먼저 실행될 명령에 대한 인터페이스를 정의한다. URL에 따라 특정 작업을 하는 컨트롤러 메서드를 만드는 것이므로 아래와 같이 선언한다.

  public interface BoardController {
  
  public void execute(HttpServletRequest req, HttpServletResponse res)
      throws ServletException, IOException;
  }

• 위 인터페이스를 구현하는 게시판 생성 객체는 아래와 같이 선언할 수 있다.

  public class BoardCreateController implements BoardController {
  
  private final CategoryDAO categoryDAO;
  
  public BoardCreateController(CategoryDAO categoryDAO) {
    this.categoryDAO = categoryDAO;
  }
  
  @Override
  public void execute(HttpServletRequest req, HttpServletResponse res)
      throws ServletException, IOException {
  
    List<String> categories = categoryDAO.getCategories();
  
    req.setAttribute(Constants.CATEGORIES, categories);
    req.getRequestDispatcher("/createBoard.jsp").forward(req, res);
  }
  }

• execute 메서드 안에서 게시판 생성 작업을 처리한다. 즉, 특정 기능마다 BoardController 인터페이스를 상속하여 기능을 캡슐화하고, execute() 메서드를 통해 URL에 따라 기능이 호출되게 한다. 호출자 코드는 위에 본 것처럼 서블릿이 담당한다.

팩토리 메서드 패턴이란

• 객체 생성의 책임을 서브클래스나 별도의 클래스에 위임하여 객체 생성 과정과 사용하는 코드를 분리하는 패턴이다.

  public class BoardModifyController implements BoardController {
  private final BoardDAO boardDAO = new BoardImpl();
  private final BoardValidator validator = new BoardValidator();

• 위는 필드 주입 방법이다. 필드 주입 단점은 테스트 코드를 작성할 때 단위 테스트(모킹)이 어렵다는 점이다.

필요한 이유

private final BoardDAO boardDAO;
  private final BoardValidator validator;

public BoardModifyController(BoardDAO boardDAO, BoardValidator validator) {
  this.boardDAO = new boardDAOImpl();
  this.validator = new Validtor();
}

• 생성자 방식으로 수정해도 직접 인스턴스화하면 여전히 강한 결합되어 변경에 취약해지고, 모킹이 어렵다. 그럼 외부에서 인스턴스화해서 넣어주면 되지 않을까?

public class BoardControllerFactory {
  private final CategoryDAO categoryDAO;
  private final BoardDAO boardDAO;
  private final BoardValidator validator;

  public BoardControllerFactory(
      CategoryDAO categoryDAO, BoardDAO boardDAO, BoardValidator validator) {
    this.categoryDAO = categoryDAO;
    this.boardDAO = boardDAO;
    this.validator = validator;
  }

  public Map<String, BoardController> createExactMappings() {
    Map<String, BoardController> exactMappings = new HashMap<>();
    exactMappings.put("/boards/free/write", new BoardCreateController(categoryDAO));
    exactMappings.put(
        "/boards/free/list", new BoardListController(boardDAO, categoryDAO, validator));
    exactMappings.put("/boards/free/delete", new BoardDeleteController(validator));

    return exactMappings;
  }

  public Map<Pattern, BoardController> createRegexMappings() {
    Map<Pattern, BoardController> regexMappings = new HashMap<>();
    regexMappings.put(
        Pattern.compile("^/boards/free/view/[0-9]+"),
        new BoardDetailController(boardDAO, validator));
    regexMappings.put(
        Pattern.compile("^/boards/free/modify/[0-9]+"),
        new BoardModifyController(boardDAO, validator));
    regexMappings.put(
        Pattern.compile("^/boards/[0-9]+/comments$"),
        new CommentCreateController(boardDAO, validator));
    regexMappings.put(
        Pattern.compile("^/boards/free/delete/[0-9]+"),
        new BoardDeleteExecutionController(boardDAO, validator));

    return regexMappings;
  }

  public BoardDAO createBoardDAO() {
    return boardDAO;
  }

  public BoardValidator createBoardValidator() {
    return validator;
  }
}

전체 코드: https://github.com/ji-jjang/ebrainsoft/tree/main/JspBoard

• BoardControllerFactory 는 다양한 BoardController 객체 생성을 책임진다. 필요한 의존 관계를 받아 인스턴스를 생성한 후 요청이 오면 해당 객체를 반환한다. BoardModifyController 에서 BoardDAO와 Validator가 필요한데, 내부적으로 이미 해당 객체를 가지고 있어 컨트롤러를 생성할 때 주입해줄 수 있으며, 별도로 해당 객체를 반환해주는 메서드까지 있어 유연하게 사용할 수 있다.

옵저버 패턴이란

• 객체의 상태 변화를 관찰하는 관찰자들, 즉 옵저버들의 목록을 객체에 등록하여 상태 변화가 있을 때마다 메서드 등을 통해 객체가 직접 목록의 각 옵저버에게 통지하도록 하는 디자인 패턴이다. (위키 백과)

• 잘 와닿지 않는다. 간단한 빙고 게임을 생각해보자. 심판(Referee)이 임의의 숫자를 하나씩 부르고, 플레이어(Player)들은 심판이 부른 숫자를 보드판에 마킹하여 빙고를 찾아내는 게임이다. 단순하게 구현하면 아래와 같은 코드를 생각해볼 수 있다.

while (!hasWinner()) {

int CalledNumber = Referee.getCalledNumber();
player.markBoard(calledNumber);
player.checkBingo();
for (var player : players) {
    if (player.isBingo()) {
        winners.add(player);
        hasWinner = true;
    }
}

if (hasWinner) {
    System.out.printf("우승자는 ");
    for (var winner : winners) {
        System.out.printf("winner.getName()), ");
    }
    System.out.println(" 입니다.");

} }

## 필요한 이유
- 이런식으로 작성하면 제대로 동작하지 않을까? 간단한 게임에선 크게 문제가 되지 않는다. 하지만 기능이 추가될수록 해당 **while문 안에 코드를 점점 추가되어 가독성이 떨어지는 문제**가 생긴다.
- 예를 들어, 특정 플레이어는 marking을 한 번에 두 번 할 수 있다거나 특정 숫자를 맞출 때 보너스를 주거나 하는 등의 기능을 추가할 때 코드 로직이 복잡해진다.
- **옵저버 패턴을 사용하면 해당 객체의 기능을 캡슐화**할 수 있다. 특히 이벤트(심판이 부르는 숫자)가 발생하고, 해당 이벤트를 다른 객체들(플레이어)이 바로 처리해야 할 때 유용하다. 새로운 기능이 추가될 때 단순히 관찰자만 추가하고 행동만 다르게 정의하면 되기 때문에 확장성에도 좋다.

## Observer 패턴 구조

> 전체 코드: https://github.com/ji-jjang/Learning/commit/cee8f544da3e5d368cb260d012d4ac5dad96bbfd

- **Observer들을 관리하는 주체 Class를 만든다. 관찰자를 등록, 해제, 그리고 이벤트를 발행하는 역할**을 한다. **이벤트가 발생하면 Observer는 단순히 자신의 역할에 맞게 행동**하면 된다.

```java
public class BingoGame {
    private List<Observer> observers = new ArrayList<>();

    public void addObserver(Observer observer) {
        observers.add(observer);
    }

    public void deleteObserver(Observer observer) {

        observers.remove(observer);
      }
    public void notifyObservers() {
        for (var o : observers) {
            o.update(this);
        }
    }
}

• 만약 관찰자들을 관리하는 주체가 위의 기본 기능을 가지면서, 상황에 따라 다른 기능을 포함해야 한다면 추상 클래스로 선언하고 해당 추상 클래스를 상속한 클래스에서 재정의하게 설계하면 좋다.

• 관찰자는 자신의 역할에 따라 행동을 다르게 할 수 있도록 인터페이스만 제공한다.

  public interface Observer {
  
  public abstract void update(NumberGenerator generator);
  }

• 빙고 게임에서 크게 번호를 부르고, 자신이 부른 번호를 관리하는 심판(refree) 객체, 보드판에 불린 숫자를 마킹하고 빙고를 외치는 플레이어(player) 객체, 승리자를 말해주는 Checker가 있다고 해보자. 그럼, Observer 인터페이스를 구현한 후 update 메소드에서 각자 역할을 수행하면 된다.

1. 플레이어

public class Player implements Observer {

  private String name;
  private int[][] board;
  private boolean isBingo;

  @Override
  public void update(BingoGame game) {

     markBoard(game.getBingoNumber());
     checkBingo();
  }
}

2. 심판

public class Referee implements Observer {

  private List<Integer> calledNumbers = new ArrayList<>();
  Random random = new Random();

  @Override
  public void update(BingoGame game) {

    int number = 0;
    do {
      number = random.nextInt(game.getMaxNumber() + 1);
    } while (calledNumbers.contains(number));

    System.out.println("call: " + number);
    game.setBingNumber(number);
    calledNumbers.add(number);
  }
}

3. WinnerChecker

public class WinnerChecker implements Observer {

  private List<String> winners = new ArrayList<>();

  @Override
  public void update(BingoGame game) {

    for (var observer : game.getObservers()) {
      if (observer instanceof Player) {
        if (((Player) observer).isBingo()) {
          winners.add(((Player) observer).getName());
        }
      }
    }
    if (winners.size() > 0) {
      game.setHasWinner(true);
      System.out.printf("우승자는 {");
      for (var e : winners) {
        System.out.printf(e + " ");
      }
      System.out.println("} 입니다.");
    }
  }
}

4. Main

public static void main(String[] args) throws IOException {

    BingoGame bingoGame = new BingoGame();

    for (int i = 0; i < bingoGame.getPlayerCount(); ++i) {
      Player player = new Player("player " + i, bingoGame);
      bingoGame.addObserver(player);
    }

    Referee referee = new Referee();
    bingoGame.addObserver(referee);

    WinnerChecker winnerChecker = new WinnerChecker();
    bingoGame.addObserver(winnerChecker);

    bingoGame.startGame();
}

public void startGame() {

    while (!hasWinner) {
      notifyObservers();
    }
}

• 메인 함수에서는 단순히 Player, Refree, WinnerChecker를 관찰자로 등록하고 게임을 시작한다. 이제는 클라이언트 코드가 변경될 때 단순히 클라이언트 코드만 수정하면 된다. BingoGame 객체는 관찰자들이 어떤 메소드를 가지고 있는지, 어떤 상태값을 주는지 등 알 필요가 없어진다. 즉, 유연하고 느슨한 설계가 된다.

여담 (React)

코드 링크: https://github.com/ji-jjang/ebrainsoft/tree/main/BingoGame

• React로 빙고 게임을 작성하면서 위 내용을 그대로 구현해 보았다. 옵저버 패턴을 적용해 보려고 했지만, react는 이미 상태 변화와 리랜더링으로 옵저버 패턴을 이미 유사하게 적용하고 있기에 적용할 부분을 찾지 못했다. useEffect()로 Referee 컴포넌트가 호출한 숫자들이 변경되면(notify) Player 컴포넌트는 화면을 리랜더링하는 식으로 동작하기 때문이다.

      {gameStart && (
        <>
          <Referee
            callNumber={callNumber}
            isGameOver={isGameOver}
            calledNumbers={calledNumbers}
          />
          <div className="boards-container">
            {!isNaN(players) &&
              Array(players)
                .fill()
                .map((undefined, playerIndex) => (
                  <Player
                    key={playerIndex}
                    playerIndex={playerIndex}
                    rows={rows}
                    cols={cols}
                    maxNumber={maxNumber}
                    calledNumbers={calledNumbers}
                    handleWin={() => handleWinners(playerIndex)}
                    isGameOver={isGameOver}
                  />
                ))}
          </div>
          <WinnerChecker
            winningPlayers={winningPlayers}
            players={players}
            setIsGameOver={setIsGameOver}
          />
        </>
      )}

데이터가 백만개, 천만개라면 아래 쿼리는 어떻게 동작할까?

SELECT * FROM users WHERE name = 'juny';

• name에 Index가 걸려있지 않다면 전체 테이블을 찾는 과정(full scan)을 거친다. name에 Index가 있다면, 마치 이진 탐색처럼 동작하여 더 빨리 데이터를 찾아낼 수 있다.

왜 인덱스를 사용해야 할까?

SELECT * FROM users WHERE name = 'juny';
DELETE FROM spaces WHERE deleted_at < '2024.10.12';
UPDATE users SET is_totp_enabled = true WHERE id = 10;
SELECT * FROM users U JOIN question Q ON U.id = Q.id;

• 조건을 만족하는 튜플(들)을 빠르게 조회, 정렬(order by), 그룹핑(group by)하기 위해 사용한다.

인덱스 사용하기

SELECT * FROM users WHERE name = 'juny';

CREATE INDEX name_idx ON users (name); 

• 만약 인덱스가 적용된 컬럼에 고유성을 보장하고 싶다면 UNIQUE 제약 조건을 부여한다.

  SELECT * FROM users WHERE is_totp_enabled = false AND phone_number IS NOT NULL; 
  

CREATE UNIQUE INDEX totp_enabled_phone_number_idx ON users (is_totp_enabled, phone_number);

- 단순히 인덱스를 설정할 땐 테이블 밖에서 인덱스가 선언되지만, UNIQUE 제약 조건으로 인덱스를 부여했다면, 고유성을 보장하기 위해 테이블 안에서 인덱스가 생성되는 것을 확인할 수 있다.
![](https://velog.velcdn.com/images/ji-jjang/post/934745ab-b5e1-4244-a08c-47dd06ae48a1/image.png)

### 사용 중인 인덱스 조회하기
```sql
SHOW INDEX FROM users;

• Index가 Unique 여부, 멀티 컬럼 인덱스 여부 등을 알려준다.

주의할 점

CREATE UNIQUE INDEX totp_enabled_phone_number_idx ON users (is_totp_enabled, phone_number);

SELECT * FROM users WHERE phone_number = '010-1234-5678';
SELECT * FROM users WHERE is_totp_enabled = false OR phone_number IS NOT NULL;

• index가 먼저 is_totp_enabled 기준으로 정렬되고나서 phone_number 기준으로 정렬된다. 위의 phone_number만으로 쿼리를 조회하게 되면 index를 제대로 사용할 수 없기에 full scan을 하게된다.
• 마찬가지로 OR 조건일 때도 한 쪽 조건에 대해선 index가 제대로 동작하지만, 뒤에 있는 조건에 대해선 제대로 동작하지 않는다.
• *또한, 인덱스 키 값에 변형이 가해진 후 비교되는 경우 절대 B-TREE의 빠른 검색 기능을 사용할 수 없다. 이미 변형된 값은 B-TREE 인덱스에 존재하는 값이 아니다. 따라서 함수나 연산을 수행한 결과로 정렬한다거나 검색하는 작업은 B-TREE의 장점을 이용할 수 없다. *

즉, 사용되는 query에 맞춰서 적절하게 Index를 걸어주어야 한다. 하지만, Index를 생성하면 데이터가 저장될 때마다 매번 정렬해야 하므로 저장하는 과정이 복잡하고 느려지게 된다. 인덱스는 저장 성능을 희생하고, 그 대신 데이터의 읽기 속도를 높이는 기능이라고 볼 수 있다.

Full Scan이 더 좋은 경우

• 인덱스를 통해 테이블의 레코드를 읽는 것은 인덱스를 거치지 않고 바로 테이블의 레코드를 읽는 것보다 높은 비용이 드는 작업이다.
• DBMS 옵티마이저에서는 인덱스를 통해 레코드 1건을 읽는 것이 테이블에서 직접 레코드 1건을 읽는 것보다 45배 정도 비용이 더 많이 드는 것으로 예측한다. **전체 테이블 레코드의 2025%를 넘어서면 인덱스를 이용하지 않고 풀 스캔이 더 효율적**일 수 있다.

BTREE 인덱스 동작 방식

• 자식이 두 개인 Binary Search Tree(이진 탐색 트리)를 더 일반화한 형태이다. 자녀 노드의 최대 개수를 상황에 맞게 결정하여 사용할 수 있다.

BTREE 인덱스

• 트리 구조 최상위에 하나의 루트 노드(Root node) 가 존재하고 그 하위에 자식 노드가 붙어 있는 형태를 말한다. 트리 구조 가장 하위에 있는 노드를 리프 노드(Leaf node) 라 하고, 트리 구조에서 루트와 리프 노드가 아닌 노드를 브랜치 노드(Branch node), 인터널 노드(Internal node)라고 한다.
• 데이터베이스에서 인덱스와 실제 데이터가 저장된 데이터는 따로 관리되는데, 인덱스의 리프 노드는 항상 실제 데이터 레코드를 찾아가기 위한 주솟값을 가지고 있다. 인덱스는 테이블의 키 컬럼만 가지고 있으므로 나머지 컬럼을 읽으려면 데이터 파일에서 해당 레코드를 찾아야하기 때문이다.

BTREE 파라미터

1. 각 노드의 최대 자녀 수(M)

• 최대 M개의 자녀를 가질 수 있는 BTREE를 M차 BTREE라고 부른다.

2. 각 노드의 최대 KEY 수(M - 1)

• 3개의 자녀를 가질 수 있는 BTREE라면 2개의 KEY를 가져야 한다.

3. 각 노드의 최대 자녀 수([M / 2])

• 당연하게 root 노드와 leaf 노드는 제외한다.
• 이 조건을 맞추어야 하는 이유는 균형을 유지하기 위해서다. 균형이 유지되지 않는 BTREE는 특정 데이터 형식에 대해 Log(n)에 동작하는 것이 아닌 O(1)에 가깝게 동작할 수 있다. 데이터가 많아질 때 극심한 성능차이를 보일 수 있다.

4. 각 노드의 최소 KEY 수([M / 2] - 1)

• 각 노드의 최대 자녀 수에 의해 자동적으로 결정되는 값이다.

BTREE 성질

1. internal 노드의 KEY 수가 x개라면, 자녀 노드의 수는 언제나 x+1 개다.
2. 노드가 최소 하나의 key는 가지기 때문에 몇 차 BTREE인지 상관없이 internal 노드는 최소 두 개의 자녀를 가진다.

삽입 동작 방식

테이블에 레코드를 추가하는 작업 비용을 1이라고 하면, 인덱스에 키를 추가하는 작업 비용을 1.5 정도로 예측한다. 중요한 것은 비용의 대부분이 메모리와 CPU에서 처리하는 시간이 아니라 디스크로부터 인덱스 페이지를 읽고 쓰기를 해야 해서 걸리는 시간이다.

1. 추가는 항상 leaf 노드에 한다.

2. 노드가 넘치면 가운데 KEY를 기준으로 좌우 KEY 들을 분할하고 가운데 KEY는 승진한다.

삭제 동작 방식

키 값이 저장된 B-TREE 리프 노드를 찾아서 삭제 마크만 하면 된다. 삭제 마킹된 인덱스 공간은 계속 그대로 방치되거나 재활용할 수 있다. 마킹 작업 또한 디스크 쓰기가 필요하므로 I/O 작업이다. InnoDB에서는 이 작업 또한 지연처리 될 수 있다.

1. 삭제는 항상 leaf 노드에 한다.

2. 삭제 후 최소 KEY 수보다 적어졌다면 재조정한다.

1) KEY 수가 여유 있는 형제의 지원을 받는다.

• (1) 동생(왼쪽 형제)이 여유가 있다면
  • 동생의 가장 큰 Key를 부모 노드의 나와 동생 사이에 둔다.
  • 원래 그 자리에 있던 KEY는 나의 가장 왼쪽에 둔다.
• (2) 형(오른쪽 형제)이 여유가 있다면
  • 형의 가장 작은 KEY를 부모 노드의 나와 형 사이에 둔다.
    • 원래 그 자리에 있던 KEY는 나의 가장 오른쪽에 둔다.

2) 1)번이 불가능하다면 부모의 지원을 받고 형제와 합친다.

• (1) 동생이 있으면 동생과 나 사이의 KEY를 부모로부터 받는다.
  • 그 KEY와 나의 KEY를 차례대로 동생에게 합친다.
    • 나의 노드를 삭제한다.
• (2) 동생이 없으면 형과 나 사이의 KEY를 부모로부터 받는다.
  • 그 KEY와 형의 KEY를 차례대로 나에게 합친다.
  • 형의 노드를 삭제한다.

3) 2번 후 부모에 문제가 있다면 거기서 다시 재조정한다.

• (1) 부모가 root 노드가 아니라면
  • 그 위치에서부터 다시 1번부터 재조정한다.
• (2) 부모가 root 노드고 비어있다면
  • 부모 노드를 삭제한다.
  • 직전에 합쳤던 노드가 root 노드가 된다.

Internal 노드에서 데이터 삭제하려면, leaf 노드에 있는 데이터와 위치를 바꾼 후 삭제하는 데 순서를 보장하기 위해 삭제할 데이터의 선임자나 후임자를 찾아야 한다. 자세한 설명은 생략한다.

왜 DB Index에 BTREE가 사용될까?

• BTREE는 균형 트리이기 때문에 최악의 데이터 셋에 대해서도 O(logN)을 보장한다.

그런데, AVL-Tree 또는 Red-Black TREE도 O(logN)을 보장하는 데 왜 BTREE를 사용할까?

• Database는 메모리가 아닌 디스크에 있는 데이터를 블럭 단위로 읽어오는 것이다. 물론 한 번 읽어온 것은 InnoDB 버퍼풀 메모리에 저장되지만, 처음 접근할 때는 디스크에서 읽어온다.그림으로 BTREE와 AVL TREE 구조에 따른 디스크 접근 횟수를 살펴보자.
• 찾아야 할 데이터는 5, 8, 7, 9 블록이고, 루트노드만 메모리에 올라와있다고 해보자.
• 5차 BTREE 구조에선 2번의 디스크 접근, AVL 트리 구조에서는 4번의 디스크 접근이 발생한다.

*즉, BTREE처럼 연관된 데이터를 한 블럭에 모아 저장하면 디스크 접근을 최소화할 수 있다. *

참고자료

• RealMySQL 8.0
• 쉬운코드 유튜브

• 로드(부하)를 밸런싱(균형, 분산)하는 것을 말한다. 쉽게 말해, 클라이언트 요청이 많아져 서버에 부하가 많아질 때 요청들을 여러 서버에 균형있게 분산시켜 각 서버가 원활히 동작하도록 한다.

Scale Up과 Scale Out

• 사용자 요청이 급격하게 증가할 때 대처 방안으로 서버 자체의 성능을 향상 시키는 Scale Up과 비슷한 서버를 증설하여 요청을 분배하는 Scale Out 방법이 있다.

• Scale Up의 경우 하드웨어 성능을 무한정 올리기에 한계가 있고, 시스템 장애에 취약하다는 단점이 있다. (SPOF, Single Point Of Failure)
• Scale Out은 비교적 적은 비용으로 다수의 사용자 요청을 처리할 수 있으며, 무중단 서비스를 제공할 수도 있다. 하지만 세션 불일치 문제, 서버 관리 등 기술적 난이도가 존재한다.

Scale Out과 로드 밸런서를 이용하여 성능 및 가용성 이점을 얻을 수 있고, reverse proxy를 이용한다면 API 서버를 내부망에 감춰 보안이 향상된다!

SSL Offloading

• SSL Offloading이란 암호화 및 복호화 작업을 각 서버가 하는 것이 아닌 로드 밸런서에서 처리하는 것을 말한다.

• SSL OffLoading을 적용하면 로드 밸런서에 SSL 인증서가 존재하기에 SSL Handshake 과정이 효율적으로 진행된다. 또한 암호화된 요청이 아닌 평문으로 웹 서버에 전달하기 때문에 네트워크 대역폭을 더 효율적으로 사용할 수 있다. 이외에도 서버를 증설할 때 SSL 인증서를 추가하지 않아도 된다는 점, 증설된 서버는 암호화 / 복호화하는 데 CPU를 사용하지 않아도 된다는 장점이 있다.

서비스 디스커버리(Service Discovery)

• 서비스가 Scale Out되어 추가되면, 클라우드 환경에서는 IP 주소가 동적으로 할당된다. 로드 밸런서는 어떻게 동적으로 부여된 IP주소를 파악하여 로드 밸런싱을 할 수 있을까?
• AWS를 사용하는 클라우드 환경에서는 ELB(Elastic Load Balancer)를 사용해 인스턴스에 트래픽을 분배할 수 있으며, Scale Out된 인스턴스에도 쉽게 대응할 수 있다.
• Spring Boot를 사용하면, Netflix가 개발한 Eureka Server와 Eureka Client 의존성을 추가하여 Auto Scaling된 인스턴스의 IP와 상태를 동적으로 파악하고 로드밸런싱을 처리할 수 있다.

• 그림에서, Spring Cloud Gateway는 오토 스케일링된 인스턴스의 존재를 모르지만, Eureka Server가 Gateway에 목록을 전달하여 로드밸런싱 기능을 수행할 수 있다.

로드 밸런싱 알고리즘

1. Round Robin (라운드 로빈)

• 다수의 서버에게 순서대로 요청을 할당한다.
• 서버에 균등하게 요청을 분배할 수 있다는 장점이 있으나, 각 서버의 처리량이나 서버 상태, 작업 부하가 달라지는 경우 등을 고려하지 못한다는 단점이 있다.

2. Least Connection (최소 연결)

• TCP/IP 프로토콜로 생성되는 Connection 기반으로 부하를 분산한다.
• 사용자와 서버가 정상적인 연결을 맺으면 Connection을 생성하기에 가장 Connection이 적은 서버에 요청을 전달한다.
• 작업마다 처리 시간 변동폭이 클 때 효과적으로 대응할 수 있지만, 각 서버의 활성 연결 수를 계속해서 추적해야 하는 복잡성이 존재하고, 여전히 서버 응답 시간이나 상태는 고려하지 못한다는 단점이 있다.

3. Weight Ratio (가중치 비율)

• 각 서버의 처리 능력을 고려하여 서버가 가질 수 있는 처리량 또는 Connection 비율 가중치를 토대로 부하를 분산한다.
• 각 서버 성능에 따른 효율적인 처리가 가능하지만, 최적 서버 가중치를 유지하는 데 어려움이 있을 수 있다.

4. IP Hash (IP 해시)

• 위 방법들은 독립된 여러 서버마다 세션 불일치 문제가 발생할 수 있다. 사용자 세션 정보가 A 서버에 저장되어 있을 때, B 서버로 요청이 가게 되면 B서버도 A서버에 있는 세션 정보를 가지고 있을 필요(Session Clustering)가 있다. 즉, 중복 데이터 문제가 생길 수 있다.
• IP Hash 방식을 사용하면 패킷의 IP 주소를 해싱하여 결과 해시 값에 해당하는 서버가 처리하는 것을 보장하기 때문에 세션 불일치 문제를 해결할 수 있다. 하지만, 적은 수의 클라이언트와 많은 요청을 처리하는 경우 특정 서버에 부하가 몰릴 가능성이 존재한다.

5. Least Response Time(최소 응답 시간)

• 응답 시간이 가장 빠른 서버에 우선적으로 요청을 할당하는 방식이다.
• 사용자 경험이 좋아진다는 장점이 있지만, 서버 응답 시간을 파악하기 위한 모니터링 시스템을 구축해야 하므로 구현에 복잡성이 더해진다. 또한, 각 서버 상태나 처리량을 고려하지 못한다.

각 방법마다 장단점이 있기 때문에, 비즈니스 성격에 맞게 복합적인 로드 밸런싱 알고리즘을 구축할 필요가 있다.

Nginx로 로드 밸런싱하기

전체 코드: https://github.com/ji-jjang/Learning/tree/main/Practice/LoadBalancer

1. 환경 구성

2. 라운드로빈 방식

upstream api_servers {
    server 172.29.115.222:8080;
    server 34.64.229.17:5001;
    server 34.64.222.54:5002;
}

server {
    listen 80;

    location / {
        proxy_pass http://api_servers;
        proxy_set_header Host $host;
    }
}

// 출력 결과
> curl 172.29.115.222:80/hello
hello Server1

> curl 172.29.115.222:80/hello
hello Server2

> curl 172.29.115.222:80/hello
hello Server3

> curl 172.29.115.222:80/hello
hello Server1
...

• Host 헤더를 지정하지 않으면 400 Bad Request 에러가 발생한다. 그 이유는 api_servers에 proxy_pass를 하게되는데, 이는 호스트 도메인이 아니기 때문이다. 따라서 원래 Host값(172.29.115.222)을 명시해주어야 한다.

1) L4 로드 밸런서

• L4 로드 밸런서는 IP와 PORT를 기준으로 로드 밸런싱을 수행한다. 위의 예시처럼 아이피와 포트 번호를 가지고 요청을 분산시킨다.

2) L7 로드 밸런서

• L7 로드 밸런서는 URL, Header, Cookie 등의 내용으로 요청을 라우팅한다.

  Server {
    location / {
        proxy_pass http://nextjs-app:3000;
    }
  
    location ~ ^/api {
        proxy_pass http://app:8080;
    }
  
    location ~ ^/actuator {
        proxy_pass http://app:8080;
    }
  }

3. Least Connection (최소 연결 방식)

• 기본 커넥션 수가 동일하므로 라운드 로빈 방식과 동일하게 동작한다.

  upstream api_servers {
    least_conn;
    server 172.29.115.222:8080;
    server 34.64.229.17:5001;
    server 34.64.222.54:5002;
  }
  

server { listen 80;

location / {
    proxy_pass http://api_servers;
    proxy_set_header Host $host;
}

}

### 4. Weight Ratio (가중치 방식)
```java
upstream api_servers {
    server 172.29.115.222:8080 weight=3;
    server 34.64.229.17:5001 weight=2;
    server 34.64.222.54:5002 weight=1;
}

server {
    listen 80;

    location / {
        proxy_pass http://api_servers;
        proxy_set_header Host $host;
    }
}

// 출력 결과
> curl 172.29.115.222:80/hello
hello Server1

> curl 172.29.115.222:80/hello
hello Server2

> curl 172.29.115.222:80/hello
hello Server1

> curl 172.29.115.222:80/hello
hello Server3
...

5. IP Hash (IP 해시)

upstream api_servers {
    ip_hash;
    server 172.29.115.222:8080;
    server 34.64.229.17:5001;
    server 34.64.222.54:5002;
}

server {
    listen 80;

    location / {
        proxy_pass http://api_servers;

                proxy_set_header Host $host;
    }
}

// 출력 결과
> curl 172.29.115.222:80/hello
hello server3

> curl 172.29.115.222:80/hello
hello server3

...

참고 자료

• https://aws.amazon.com/ko/what-is/load-balancing/
• https://www.youtube.com/watch?v=RN6ijLAz7L8
• https://m.post.naver.com/viewer/postView.naver?volumeNo=27046347&memberNo=2521903
• https://www.geeksforgeeks.org/what-is-load-balancer-system-design/#what-will-happen-if-there-is-no-load-balancer
• https://nginxstore.com/blog/api-gateway/nginx-api-gateway-l7-load-balancer-설정/
• https://www.geeksforgeeks.org/what-is-load-balancer-system-design/#what-will-happen-if-there-is-no-load-balancer

1. 기존 Monolithic Archiecture 한계

• 모놀리식 구조란 소프트웨어를 하나의 시스템으로 구축하여 개발하는 방식을 말한다. 하나의 시스템에 모든 설정과 비즈니스 로직이 포함된다.
• 간단한 프로젝트에는 적용하기 좋으나, 프로젝트가 점점 커지고 요구사항이 많아질수록 아래 문제가 생긴다.
  • 작은 수정사항에도 전체 빌드 및 배포가 이뤄져야 해서 시간이 오래 걸림
  • 시간이 지날수록 방대한 양의 코드가 몰려 있어 유지보수 어려움
  • 일부 오류가 전체에 영향을 미치는 점 (단일 오류 포인트)
  • 특정 서비스(ex: 주문)만 스케일 아웃하고 싶지만 세세히 설정하기 어려움

2. MSA 등장 및 특징

• 이러한 단점으로 인해 MSA(MicroService Architecture) 구조가 등장하게 되었다. MSA는 소프트웨어 시스템을 여러 작은 독립적인 서비스로 분할하여 개발하고 배포하는 방식이다.

• 가장 큰 Monolithic Architecture와 가장 큰 차이점은 아래와 같은 특징을 가진다는 것이다.

1) 작고 한 가지 일에 주력

• 각 서비스는 가능한 한 한 가지 일에 주력해야 기능 추가, 변경에 대해 유연하게 대처할 수 있으며, 독립적인 배포 이점을 살릴 수 있다.
• 서비스를 작게 만들수록 서비스 간 상호 의존성이 낮아지기 때문에 유연하고 확장성 있는 구조를 가지게 된다.
• 다른 서비스에 장애가 발생하더라도 각 서비스는 독립적이기 때문에 전체 시스템에 장애가 생기지 않는다.

2) 자율성

• 서비스끼리 상호 의존성을 줄이기 위해 네트워크 통신으로 API를 호출하게 된다. 다른 서비스 변경 없이 특정 서비스만 변경하고 배포할 수 있다면, 각 서비스에게 적절한 역할과 책임을 부여한 것이다.
• 또한 각 서비스마다 부여된 책임을 다하기 위해 적절한 기술 스택을 사용할 수 있다는 장정을 가진다. 성능이 중요한 곳에선 NoSQL을 사용할 수 있고, 데이터 무결성 제약이 중요하다면 관계형 DB를 사용할 수도 있다. 언어 또한 자유롭게 선택할 수 있다.

정리하자면 MSA는 Monolithic Architecture에 비해 배포 용이성, 확장성, 유연성, 고가용성(SPOF 줄임), 기술 선택 자율성 등을 꼽을 수 있다. 하지만 MSA 구조를 항상 적용해야 하는 건 아니다. 설계의 복잡성뿐만 아니라 분산 트랜잭션, 분산 캐시 일치, 서비스 간 통신 비용 등 여러 극복해야 할 과제가 많다.

• https://www.youtube.com/watch?v=CM47-1UpgOc
  • Vingle의 예시처럼 모놀리식 구조에서는 프론트 페이지 요청 증가, API 요청 증가 등 이를 나누어서 Scale Up 하기 어려웠고, 프론트앤드 코드만 수정해도 백엔드 코드까지 배포해야 하며 단위 테스트를 실행하는데 최소 30분 이상 걸렸던 문제를 MSA 아키텍처로 해결한 사례이다.

Spring Cloud MSA 시작하기

MSA 환경 구성 요소

1. Spring Cloud Gateway

• 요청을 URL 경로에 따라 서비스에 분배하는 역할을 수행한다.

2. Spring Cloud Eureka Server

• 모니터링 서버로 Eureka Client 설정을 한 클라이언트들을 한눈에 볼 수 있다.
• Spring Cloud Gateway는 오토스케일링된 인스턴스의 존재를 모르지만, Eureka Server가 Gateway에 목록을 전달하여 로드밸런싱 기능을 수행할 수 있다.

3. Spring Cloud EureKa Client

• 독립된 서비스를 제공하는 스프링 부트 애플리케이션은 Eureka Server에 등록되어 모니터링 대상이 된다.
• 아래서 설명하겠지만 Eureka Client이면서 동시에 Config Server에서 설정 파일을 읽으므로 Config Client이기도 하다.

4. Spring Config Server

• 서비스마다 스프링 설정을 다르게 적용하기 위해 Config Server를 사용한다. 서버에 설정을 주입할 때 Valut, Private Git Repository, RDB, Redis, File 등 다양한 방법을 사용할 수 있다.

• 여기선 Git Repository를 private로 만들어 {애플리케이션이름-프로필.yml} 파일을 읽어 들인다. 저장소에 SSH 공개 키를 등록하고 Çonfig Server에서는 Private Key를 이용해 설정 파일에 접근할 수 있다.

5. Spring Config Client

• Config Server로 부터 설정 변수를 주입받는다. Config Client이면서 동시에 Spring Cloud Eureka client이기도 하다.

Spring Cloud MSA 환경 간단하게 구성하기

1. Spring Config Server

• 내부망을 사용한다면 Security 설정을 하지 않아도 되겠지만, 내부망을 사용하지 않는다면 인증된 사용자만 접근할 수 있도록 Spring Security를 사용하는 것이 권장된다.

• Config Server로 동작할 수 있도록 build.gradle에 Spring Config Server 종속성을 추가한다. implementation 'org.springframework.cloud:spring-cloud-config-server'

• Main 클래스에 @EnableConfigServer 어노테이션을 통해 Config Server로 등록한다.

  @SpringBootApplication
  @EnableConfigServer
  public class ConfigServerApplication {
  
  public static void main(String[] args) {
    SpringApplication.run(ConfigServerApplication.class, args);
  }
  }

• Config Server는 private Git Repository에 있는 설정 파일을 읽어 각 독립된 서비스를 만들 때 설정 파일을 읽어올 수 있도록 해야 한다. 따라서 private Git Repository url과 비밀 키를 application.properties에 등록해야 한다.

  spring.application.name=ConfigServer
  

server.port=9000

spring.cloud.config.server.git.uri=git@github.com:ji-jjang/msa-config.git spring.cloud.config.server.git.ignoreLocalSshSettings=true spring.cloud.config.server.git.private-key=
-----BEGIN RSA PRIVATE KEY-----\n
MIIJKQIBAAKCAgEAxERtVxWXnQ0m3inzKOWpVf+ulVQpjixCubVn+MLpIQZUYoeP\n
...생략 0OxtdUwADW7AesDVfxyszMQMGdp2tA8A8ssexRdX8NEXV7eYC0wcCiEopi65\n
-----END RSA PRIVATE KEY-----

- Config Client는 `http://ip:port/저장소이름/저장소환경`을 통해 Config Server에 접근하며, 여기서 ip와 port는 Config Server의 값을 입력하고, 저장소 이름과 저장소 환경은 Config Repository에 해당하는 내부 파일이름을 넣어야 한다.
```java
이름-환경.yml
이름-환경.properties

• localhost:9000/애플리케이션이름/환경로 접근하면 설정 파일의 key, value 값이 JSON 형태로 표시되는 것을 확인할 수 있다.
• 독립된 서비스는 각각 Config Server를 통해 private git repository에 있는 설정 파일을 읽어 들인다고 생각하면 된다.

2. Spring Config Client

• Config Client는 Config Server에 repository에 있는 설정 파일을 읽어들이기 위해 config client 종속성을 추가해야 한다.

  ext {
    set('springCloudVersion', "2023.0.3")
  }
  

dependencies { implementation 'org.springframework.cloud:spring-cloud-starter-config' } dependencyManagement { imports { mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}" } }

- Config Server와 연결하기 위해 설정 파일을 다음과 같이 추가한다. 내부망이 아니기에 인증된 사용자 아이디와 비밀번호를 사용해 연결하는 방법을 사용했다.
```properties
server.port=8081
spring.application.name=service1
spring.profiles.active=dev
spring.config.import=optional:configserver:http://admin:1234@localhost:9000

3. Spring Eureka Server

• Eureka Server는 MSA를 구성하는 각각의 서비스들을 모니터링 할 뿐만 아니라 오토 스케일링되는 여러 서비스들을 라우팅 대상이 될 수 있도록 Gateway에 알려주는 역할도 한다.
• Eureka Server로 등록하기 위해 아래 의존성을 추가한다. 모니터링 서버이므로 관리자만 접근해야 하며 내부망을 사용하지 않는다면 Security 설정을 추가로 해야 한다.

  ext {
    set('springCloudVersion', "2023.0.3")
  }
  

dependencies { implementation 'org.springframework.cloud:spring-cloud-starter-netflix-eureka-server' }

dependencyManagement { imports { mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}" } }

- Main Class에 `@EnableEurekaServer` 어노테이션을 통해 Eureka Server로 등록한다.
```java
@SpringBootApplication
@EnableEurekaServer
public class EurekaServerApplication {

    public static void main(String[] args) {
        SpringApplication.run(EurekaServerApplication.class, args);
    }
}

• server port를 지정해 주고, 유레카 서버에 만들 때 자신의 상태도 모니터링 대상이 되므로 (즉, 서버면서 클라이언트이다) 연결 설정을 해주어야 오류가 발생하지 않는다.

  server.port=8761
  eureka.client.register-with-eureka=false # Eureka 서버가 클라이언트로서 자기 자신을 다른 Eureka 서버에 등록할지 여부
  eureka.client.fetch-registry=false # Eureka 서버로부터 레지스트리 정보를 가져올지 여부
  eureka.client.service-url.defaultZone=http://admin:1234@localhost:8761/eureka

• localhost:8761에 접속하면 유레카 서버 대시보드가 나타난다. 여기에 등록된 클라이언트 정보들을 볼 수 있다.

4. Spring Eureka Client

• 기존 Config Client로 설정을 읽어 들인 서비스를 Eureka Client에 등록해야 모니터링 대상이 된다.
• 필요한 Eureka Client 종속성을 추가하고, @EnableDiscoveryClient 어노테이션을 통해 Eureka Client로 등록한다.

  ext {
    set('springCloudVersion', "2023.0.3")
  }
  

dependencies { implementation 'org.springframework.cloud:spring-cloud-starter-netflix-eureka-client' }

dependencyManagement { imports { mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}" } }

```java
@SpringBootApplication
@EnableDiscoveryClient
public class ConfigClientApplication {

  public static void main(String[] args) {
    SpringApplication.run(ConfigClientApplication.class, args);
  }
}

• 설정 파일을 통해 Eureka Client를 Eureka Server에 등록해야 한다.

  eureka.client.register-with-eureka=true
  eureka.client.fetch-registry=true
  eureka.client.service-url.defaultZone=http://admin:1234@localhost:8761/eureka

5. Gateway

• 여러 MicroService 앞에서 클라이언트의 모든 요청을 받은 후 url에 맞는 경로로 특정 마이크로서비스에 전달해 주어야 한다.
• Gateway의 경우 단순히 요청을 전달하는 I/O 처리만 하게 되므로 논블록킹 방식으로 동작하는 WebFlux와 Netty 엔진을 사용하는 것이 더 효율적이다.
• 필요한 종속성 추가 및 eureka server와 연결을 등록해 주어야 한다.

  ext {
    set('springCloudVersion', "2023.0.3")
  }
  

dependencies { implementation 'org.springframework.cloud:spring-cloud-starter-gateway' }

dependencyManagement { imports { mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}" } }

- 해당 설정을 추가하기만 하면, Netty Server를 실행하고 NonBlock 방식으로 요청을 처리한다. 이 경우 기존 RDB 설정으로 하면 제대로 동작하지 않으니 관련된 추가 기술 학습이 필요해진다.
```java
spring.application.name=SpringGateWay
server.port=8080

eureka.client.register-with-eureka=true
eureka.client.fetch-registry=true
eureka.client.service-url.defaultZone=http://admin:1234@localhost:8761/eureka

• 경로 설정은 application.properties, yml, java code 모두 가능하지만 여기선 간단히 properties 파일로 알아본다.

  server.port=8080
  

spring.cloud.gateway.routes[0].id=orderService spring.cloud.gateway.routes[0].predicates[0].name=Path spring.cloud.gateway.routes[0].predicates[0].args.pattern=/orders/** spring.cloud.gateway.routes[0].uri=http://localhost:8081

spring.cloud.gateway.routes[1].id=userService spring.cloud.gateway.routes[1].predicates[0].name=Path spring.cloud.gateway.routes[1].predicates[0].args.pattern=/users/** spring.cloud.gateway.routes[1].uri=http://localhost:8082

- 게이트 웨이 라우트를 0번부터 시작해서 하나씩 늘려가며 등록하면 된다.

- 또한, 로드밸런싱 라우팅도 진행할 수 있는데 **Gateway는 오토스케일링된 인스턴스를 알 수 없으므로** 먼저 Eureka Server와 연결한 뒤 Eureka Server로부터 인스턴스 정보들을 받아와야 한다.
```properties
ext {
    set('springCloudVersion', "2023.0.3")
}

dependencies {
    implementation 'org.springframework.cloud:spring-cloud-starter-netflix-eureka-client'
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"
    }
}

eureka.client.register-with-eureka=true
eureka.client.fetch-registry=true
eureka.client.service-url.defaultZone=http://admin:1234@localhost:8761/eureka

• 같은 인스턴스를 만들 때는 같은 application 이름이지만 다른 port를 사용하게끔 구성한다. orderService를 여러 Scale Out하여 등록했다면, 이제는 gateway route uri가 특정 ip, 포트에 매칭되는 게 아닌 어플리케이션 서버 이름으로 등록하며 Eureka Server에 등록된 여러 orderService 인스턴스에 요청이 각각 분배되는 방식으로 동작한다.

  spring.cloud.gateway.routes[0].id=orderService
  spring.cloud.gateway.routes[0].predicates[0].name=Path
  spring.cloud.gateway.routes[0].predicates[0].args.pattern=/orders/**
  spring.cloud.gateway.routes[0].uri=lb://ORDERSERVICE

  참고자료

• https://metanetglobal.com/bbs/board.php?bo_table=tech&wr_id=38#:~:text=MSA%EB%8A%94%20%EC%86%8C%ED%94%84%ED%8A%B8%EC%9B%A8%EC%96%B4%20%EC%8B%9C%EC%8A%A4%ED%85%9C%EC%9D%84,%EC%9D%84%20%EC%B2%98%EB%A6%AC%ED%95%98%EB%8A%94%20%EA%B5%AC%EC%A1%B0%EC%9D%B4%EC%A3%A0
• https://medium.com/@chaewonkong/microservices-architecture-vs-monolithic-architecture-fe24f9bd68ea
• 유미 개발자 https://www.youtube.com/watch?v=O1un1Nf810Y&list=PLJkjrxxiBSFBPk-6huuqcjiOal1KdU88R

• JDBC는 자바 코드로 데이터베이스와 직접 통신하는 저수준 API를 말한다.
• 아래 코드처럼 스프링 프레임워크 없이도, Database와 통신할 수 있다.

  // Mysql에 board 테이블은 비워져 있는 상태
  mysql> select * from board;
  Empty set (0.00 sec)
  

public static void main(String[] args) throws SQLException { BoardDAO dao = new BoardDAO();

dao.createPost("제목1", "내용1");

}

public void createPost(String title, String content) throws SQLException { String sql = "INSERT INTO board (title, content) VALUES (?, ?)";

try (Connection conn = DataSourceUtil.getConnection();
        PreparedStatement pstmt = conn.prepareStatement(sql)) {

    pstmt.setString(1, title);
    pstmt.setString(2, content);
    pstmt.executeUpdate();

} catch (SQLException e) {
    e.printStackTrace();
}

}

// Mysql에 board 테이블에 하나의 포스트가 추가된 모습 mysql> select * from board; +----+---------+---------+---------------------+ | id | title | content | created_at | +----+---------+---------+---------------------+ | 1 | 제목1 | 내용1 | 2024-08-23 15:32:50 | +----+---------+---------+---------------------+ 1 row in set (0.00 sec)

## 왜 JDBC를 알아야할까?
- JDBC를 사용하며 개발하던 시기는 거의 20년 전이라고 한다. 현재는 Mybatis 또는 Spring JPA를 통해 주로 개발한다. 그런데 왜 JDBC를 알아야할까? **Mybatis나 JPA 모두 내부적으로 JDBC를 사용하여 데이터베이스와 통신하므로 JDBC를 알면 두 기술을 더 잘 이해할 수 있게 된다.**
- JPA를 사용하면 편리한 이점을 얻는 대신 쿼리를 세밀하게 제어하기는 힘들다. 이 경우 JDBC 또는 Mybatis를 사용하여 쿼리 최적화를 할 수 있다.

## JDBC 주요 기능과 구성요소
### 1. 데이터베이스 연결과 해제
#### 1) DriverManager
- 새로운 드라이버를 등록하거나 등록된 드라이버를 해제한다. 애플리케이션이 데이터베이스와 연결할 수 있도록 `getConnection` 메서드를 제공한다. 프로퍼티 속성을 읽거나 url, user, password로 연결할 수 있다.
```java
public class DriverManager {
  ...
  @CallerSensitive
  public static Connection getConnection(String url,
    String user, String password) throws SQLException {

     java.util.Properties info = new java.util.Properties();

     if (user != null) {
        info.put("user", user);
     }
     if (password != null) {
       info.put("password", password);
     }

     return (getConnection(url, info, Reflection.getCallerClass()));
  }
}

// Client getConnection
private static final String URL = "jdbc:mysql://localhost:3306/jdbc";
private static final String USER = "root";
private static final String PASSWORD = "";

public static Connection getConnection() throws SQLException {
    return DriverManager.getConnection(URL, USER, PASSWORD);
}

2) Connection

• 데이터베이스와 연결된 객체이다.
• 데이터베이스와의 모든 통신은 Connection 객체를 이용한다.

  Statement createStatement() throws SQLException; // Statement 객체 생성
  

PreparedStatement prepareStatement(String sql) throws SQLException; // pstmt 객체 생성

void setAutoCommit(boolean autoCommit) throws SQLException; // 오토 커밋 설정

void commit() throws SQLException; // 커밋 설정 void rollback() throws SQLException; // 롤백 설정 void close() throws SQLException; // DB 연결 종료 ...

#### 3) Datasource
- Connection 객체를 재사용할 수 있도록 관리하는 커넥션 풀을 제공한다.
- Datasource 인터페이스에서 getConnection()으로 연결을 시도하면, 새롭게 연결을 시도하는 것이 아닌 미리 만들어진 곳(pool)에서 가져온다. 데이터베이스와의 연결을 설정하고 해제하는 작업은 비용이 크기 때문에 성능 개선 효과가 크다.
```java
public interface DataSource  extends CommonDataSource, Wrapper {

  Connection getConnection() throws SQLException;
  Connection getConnection(String username, String password)
    throws SQLException;
  ...
}

2. 쿼리 생성 및 실행

1) Statement

• SQL 쿼리를 데이터베이스에 전달하고 결과를 반환받는 데 사용한다.

• 단순하고 고정된 SQL 쿼리이거나 파라미터가 없는 경우 사용하며, SQL Injection에 취약하다는 단점이 있다.

  public interface Statement extends Wrapper, AutoCloseable {
  
  // SELECT 쿼리를 실행하고, 결과를 ResultSet 객체로 반환
  ResultSet executeQuery(String sql) throws SQLException;
  
  // INSERT, UPDATE, DELETE 쿼리를 실행하고, 영향을 받은 행의 수를 반환
  int executeUpdate(String sql) throws SQLException;
  
  // 결과가 ResultSet일지, 아니면 업데이트된 행의 수일지 미리 알 수 없는 상황에 사용
  boolean execute(String sql) throws SQLException;
  boolean execute(String sql, int autoGeneratedKeys) throws SQLException
  ...
  }

2) PreparedStatement

• 파라미터화된 SQL 쿼리를 실행하기 위해 사용한다.

• SQL Injection 방지하고, 반복 쿼리에 대해서 성능 개선 효과가 있다.

  public interface PreparedStatement extends Statement {
  
  // SQL 쿼리 내 파라미터 값 설정
  void setString(int parameterIndex, String x) throws SQLException;
  void setInt(int parameterIndex, int x) throws SQLException;
  ...
  }

3. 쿼리 결과 저장 및 반환

ResultSet

• 데이터베이스 쿼리의 결과를 테이블 형식으로 유지한다.

• 쿼리의 결과 데이터를 읽고 처리하는 데 사용한다.

  public interface ResultSet extends Wrapper, AutoCloseable {
  
    // 결과 집합의 다음 행으로 이동
    boolean next() throws SQLException;
  
    // 결과 집합의 특정 열 값 가져옴
    String getString(int columnIndex) throws SQLException;
    int getInt(int columnIndex) throws SQLException;
    ...
  }

DB 커넥션 풀링

• DriverManager를 사용하게 되면 특정 쿼리들을 실행할 때마다 getConnection() 메서드를 통해 데이터베이스와 새롭게 연결해야 한다.

• 데이터베이스와의 연결을 설정하고 해제하는 작업은 비용이 크기 때문에 커넥션 풀링이라는 기술이 도입되었다. 이는 미리 일정 수의 데이터베이스 연결을 생성해 두고, 애플리케이션이 필요할 때마다 이 풀(pool)에서 연결을 가져가는 방식으로 동작한다.

  public class DriverManagerUtil {
  private static final String URL = "jdbc:mysql://localhost:3306/jdbc";
  private static final String USER = "root";
  private static final String PASSWORD = "";
  
  public static Connection getConnection() throws SQLException {
    return DriverManager.getConnection(URL, USER, PASSWORD);
  }
  }

• 매번 쿼리를 실행할 때 DriverManager로 DB와 커넥션을 맺었다면, 아래와 같이 커넥션 풀을 이용할 수 있다. 다양한 커넥션 풀 구현체가 있지만, 여기에선 고성능 HikiriCP를 사용한다.

  public class DataSourceUtil {
  private static HikariDataSource dataSource;
  
  static {
    HikariConfig config = new HikariConfig();
    config.setJdbcUrl("jdbc:mysql://localhost:3306/jdbc");
    config.setUsername("root");
    config.setPassword("");
  
    config.setMaximumPoolSize(10);
    config.setMinimumIdle(2);
    config.setConnectionTimeout(30000);
  
    dataSource = new HikariDataSource(config);
  }
  
  public static Connection getConnection() throws SQLException {
    return dataSource.getConnection();
  }
  
  public static DataSource getDataSource() {
    return dataSource;
  }
  }

• 미리 커넥션을 10개 만들어두고, 매번 새롭게 연결하는 게 아닌 커넥션 풀에서 꺼내 사용할 수 있도록 설정한다.

• 간단히 게시글을 만들고, 읽고, 수정하는 쿼리를 400개를 실행해보고 커넥션 풀을 사용했을 때와 사용하지 않았을 때 성능 차이를 비교해본다.

  // 드라이버 매니저 getConnection - 매번 새로운 연결
  try (Connection conn = DriverManger.getConnection();
        PreparedStatement pstmt = conn.prepareStatement(createTableSQL)) {
    pstmt.executeUpdate();
  

// HikariCP getConnection - 커넥션 풀에서 가져옴 try (Connection conn = DataSourceUtil.getConnection(); PreparedStatement pstmt = conn.prepareStatement(createTableSQL)) { pstmt.executeUpdate();

```java
 public static void main(String[] args) throws SQLException {
    BoardDAO dao = new BoardDAO();

    long startTime = System.currentTimeMillis();

    dao.createTable();
    for(int i = 0; i < 100; i++) {

      dao.createPost("제목1", "내용1");

      dao.readPost(2);

      dao.updatePost(1, "제목2", "내용2");
    }

    long endTime = System.currentTimeMillis();

    long duration = endTime - startTime;

    System.out.println("Hikari Datasource pool 10개 실행 시간: " + duration + " milliseconds");
  }

• 커넥션 풀을 사용했을 때 대략 6배정도 더 빠른 걸 확인할 수 있었다.

JDBC와 Spring JDBC(JdbcTemplate)

• 그렇다면 Spring JDBC는 무엇일까? Spring JDBC는 Spring에서 지원하는 것으로 JdbcTemplate 핵심 클래스를 제공하여 데이터베이스 접근을 단순화하고, 예외 처리 및 자원 해제를 자동으로 해준다.
• 추가로 결과 집합을 객체로 매핑하기 위해 RowMapper 인터페이스를 제공하여 ResultSet 데이터를 간단하게 Java 객체로 변환할 수 있다.
• Spring JDBC를 사용하면 코드가 어떻게 변하는지 간단하게 Post CRUD를 JDBC로 작성하고, 이를 Spring JDBC로 변환해보자.

게시글 CRUD (JDBC)

public class BoardDAO {

  public void createPost(String title, String content) throws SQLException {
    String sql = "INSERT INTO board (title, content) VALUES (?, ?)";

    try (Connection conn = DataSourceUtil.getConnection();
        PreparedStatement pstmt = conn.prepareStatement(sql)) {

      pstmt.setString(1, title);
      pstmt.setString(2, content);
      pstmt.executeUpdate();

    } catch (SQLException e) {
      e.printStackTrace();
    }
  }

  public void readPost(int id) {
    String sql = "SELECT * FROM board WHERE id = ?";

    try (Connection conn = DataSourceUtil.getConnection();
        PreparedStatement pstmt = conn.prepareStatement(sql)) {

      pstmt.setInt(1, id);
      ResultSet rs = pstmt.executeQuery();

      if (rs.next()) {
        System.out.println("id: " + rs.getInt("id"));
        System.out.println("title: " + rs.getString("title"));
        System.out.println("content: " + rs.getString("content"));
        System.out.println("created_at: " + rs.getTimestamp("created_at"));
      } else {
        System.out.println("포스트 없음 ID: " + id);
      }

    } catch (SQLException e) {
      e.printStackTrace();
    }
  }

  public void updatePost(int id, String title, String content) {
    String sql = "UPDATE board SET title = ?, content = ? WHERE id = ?";

    try (Connection conn = DataSourceUtil.getConnection();
        PreparedStatement pstmt = conn.prepareStatement(sql)) {

      pstmt.setString(1, title);
      pstmt.setString(2, content);
      pstmt.setInt(3, id);
      int rowsAffected = pstmt.executeUpdate();

      if (rowsAffected > 0) {
        System.out.println("성공적으로 게시물 업데이트");
      } else {
        System.out.println("포스트 없음 ID: " + id);
      }

    } catch (SQLException e) {
      e.printStackTrace();
    }
  }

  public void deletePost(int id) {
    String sql = "DELETE FROM board WHERE id = ?";

    try (Connection conn = DataSourceUtil.getConnection();
        PreparedStatement pstmt = conn.prepareStatement(sql)) {

      pstmt.setInt(1, id);
      int rowsAffected = pstmt.executeUpdate();

      if (rowsAffected > 0) {
        System.out.println("성공적으로 게시물 삭제");
      } else {
        System.out.println("포스트 없음 ID: " + id);
      }

    } catch (SQLException e) {
      e.printStackTrace();
    }
  }
}

게시글 CRUD (Spring JDBC)

public class SpringJDBCBoardDAO {

  private JdbcTemplate jdbcTemplate;

  public SpringJDBCBoardDAO(DataSource dataSource) {
    this.jdbcTemplate = new JdbcTemplate(dataSource);
  }

  public void createTable() throws SQLException {
    String createTableSQL =
        "CREATE TABLE IF NOT EXISTS board ("
            + "id INT AUTO_INCREMENT PRIMARY KEY, "
            + "title VARCHAR(255) NOT NULL, "
            + "content TEXT NOT NULL, "
            + "created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
            + ")";

    jdbcTemplate.execute(createTableSQL);
  }

  public void createPost(String title, String content) throws SQLException {
    String sql = "INSERT INTO board (title, content) VALUES (?, ?)";
    jdbcTemplate.update(sql, title, content);
  }

  public void readPost(int id) {
    String sql = "SELECT * FROM board WHERE id = ?";

    ResBoard board = jdbcTemplate.queryForObject(sql, new RowMapper<ResBoard>() {
      @Override
      public ResBoard mapRow(ResultSet rs, int rowNum) throws SQLException {
        return new ResBoard(
          rs.getInt("id"),
          rs.getString("title"),
          rs.getString("content"),
          rs.getTimestamp("created_at")
        );
      }
    }, id);

    try {
      if (board != null) {
        System.out.println("ID: " + board.getId());
        System.out.println("Title: " + board.getTitle());
        System.out.println("Content: " + board.getContent());
        System.out.println("Created At: " + board.getCreatedAt());
      }
    } catch (Exception e) {
      System.out.println("포스트 없음 ID: " + id);
    }
  }

  public void updatePost(int id, String title, String content) {
    String sql = "UPDATE board SET title = ?, content = ? WHERE id = ?";
    int rowsAffected = jdbcTemplate.update(sql, title, content, id);

    if (rowsAffected > 0) {
      System.out.println("성공적으로 게시물 업데이트");
    } else {
      System.out.println("포스트 없음 ID: " + id);
    }
  }

  public void deletePost(int id) {
    String sql = "DELETE FROM board WHERE id = ?";
    int rowsAffected = jdbcTemplate.update(sql, id);

    if (rowsAffected > 0) {
      System.out.println("성공적으로 게시물 삭제");
    } else {
      System.out.println("포스트 없음 ID: " + id);
    }
  }
}

• Spring JDBC는 반복해서 커넥션을 얻는 코드, 예외 처리 코드 등 공통적으로 처리해야될 부분을 개발자 대신 처리해주어 코드가 더욱 간단해진다. 또한 RowMapper로 쿼리 결과를 자바 객체로 변환시키는 것도 용이하다는 장점이 있다.

• 배치란 주기적으로 <대량>의 반복적인 데이터 작업을 처리하기 위해 사용하는 방식

배치가 필요한 상황

• 대량의 집계 데이터 생성(ex: 월별 매출, 월별 환불액, 한 달 동안 신규 회원 수, 월간 상품별 판매량 …)
• DB에 있는 대량의 데이터를 백업 용도로 새로운 서버에 저장할 때
• 특정 조건(ex: 30일 연속 출석, 특정 답변 개수마다 등급, 매달 말 예금에 대한 이자…)을 맞춘 다수의 사용자들에게 보상 지급할 때
• 다수의 사용자에게 광고 이메일, 공지 사항을 보낼 때

어떤 이벤트가 발생하자마자 즉각적으로 처리를 해야 하는 실시간 작업이 아니라면, 데이터를 모아 일괄적으로 처리하면 아래와 같은 이점을 누릴 수 있다.

1. 모아서 처리하기 때문에 더 적은 DB 접근(네트워크 접근)이 가능
2. 서버 유휴 시간에 배치 프로세스 실행하면 서버 자원을 효율적으로 사용
   

꼭 스프링 배치를 사용해야 할까?

1. 배치 작업 중간에 에러가 발생했다면?

• 처음부터 다시 작업을 시작하는 건 굉장히 비효율적이다.
• 중복해서 처리되면 안 되는 경우(ex: 보상 지급) 실패 지점까지는 건너뛰거나 롤백해야 한다.
• 물론, 스프링 배치 프레임워크를 사용하지 않고 특정 배치 작업을 실행할 ID, 진행 지점을 파일에 기록한 뒤 실패했다면 파일을 읽어 실패했던 부분 이후부터 재개할 수 있다. 또는 진행한 부분까지 롤백하는 로직을 직접 구현할 수 있다.

2. 더 많은 요구사항

• 특정 작업을 중복해서 실행하고 싶지 않다면? 조건마다 처리할 작업을 다르게 하고 싶다면?
• 실패했을 경우 재시도할 횟수를 지정하고 싶다면?
• 실패해도 치명적이지 않은 에러는 무시하고 진행하고 싶다면?
• 트랜잭션으로 묶이는 작업이 많다면? 배치 작업마다 트랜잭션 격리 수준을 지정하고 싶다면?
• …

• 이러한 기능들을 사용자가 일일이 구현해야 한다.
• 스프링 배치 프레임워크를 이용하면 대량의 데이터를 효율적으로 처리하면서도 데이터의 무결성을 유지할 수 있는 다양한 방법을 제공한다.

스프링 배치 구조

• 배치는 읽고, 처리하고, 쓰는 작업을 반복적으로 실행한다. 여러 배치 작업(Job)이 존재할 수 있고 하나의 작업은 여러 독립적인 단계(Step)를 가질 수 있다. 각 단계는 읽고, 처리하고, 쓰는 작업을 반복한다.

1. JobLauncher

@FunctionalInterface
public interface JobLauncher {
  JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, JobParametersInvalidException;
}

public interface JobRegistry extends ListableJobLocator {
  void register(JobFactory jobFactory) throws DuplicateJobException;

  void unregister(String jobName);
}

• 스프링 배치 작업(Job)을 시작하는 인터페이스이다. 실행할 작업(Job)과 파라미터(jobParameter)를 받는다. jobParameter를 토대로 실행할 작업 식별한다. 같은 파라미터를 사용하면 동일한 작업으로 인식한다.
• JobRegistry에 있는 여러 작업 정보를 토대로 실행할 배치 작업(Job)을 실행한다.
• 일반적으로 스케줄러 또는 컨트롤러에서 JobLauncher와 JobRegistry를 주입받아 사용한다.

@Configuration
@RequiredArgsConstructor
public class AggregationScheduler {

  private final JobLauncher jobLauncher;
  private final JobRegistry jobRegistry; 

  @Scheduled(cron = "0 0 2 15 * *", zone = "Asia/Seoul")
  public void getSalesAndRefunds() throws Exception {

    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM");
    String date = dateFormat.format(new Date());

    JobParameters jobParameters = new JobParametersBuilder()
      .addString("date", date)
      .toJobParameters();

    jobLauncher.run(jobRegistry.getJob("saleAggregationJob"), jobParameters);
  }
}

2. Job

• Job은 하나의 배치 작업을 말하며, 배치 처리 기본 단위이다.
• 하나의 Job은 여러 개의 Step(독립적인 처리 단위)으로 구성된다.

  @Bean
  public Step cancelPendingOrdersStep() {

    return new StepBuilder("cancelPendingOrdersStep", jobRepository)
      .<Order, Order>chunk(10, platformTransactionManager)
      .reader(pendingOrdersReader())
      .processor(pendingOrdersProcessor())
      .writer(pendingOrdersWriter())
      .build();
  }

• 하나의 Job으로부터 파라미터 값을 이용하여 여러 JobInstance를 만들 수 있으며, JobInstance로 실제 실행을 할 수 있다. 재시도, 재시작 등의 이유로 하나의 JobInstance는 여러 개의 JobExecution(실행)을 가질 수 있다.

// job이름과 job 파라미터로 jobInstance 생성
public interface JobRepository {
  ...
  JobInstance createJobInstance(String jobName, JobParameters jobParameters);
}

// job과 JobParameter에 관해 JobExecution 반환
public interface JobLauncher {
  JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, JobParametersInvalidException;
}

3. Step

• Job 내에서 실행되는 독립적인 처리 단위로, 읽고 처리하고 쓰는 단계를 반복한다.

@Bean
public Step cancelPendingOrdersStep() {
    return new StepBuilder("cancelPendingOrdersStep", jobRepository)
        .<Order, Order>chunk(10, platformTransactionManager)
        .reader(pendingOrdersReader())
        .processor(pendingOrdersProcessor())
        .writer(pendingOrdersWriter())
        .build();
}

@Bean
public RepositoryItemReader<Order> pendingOrdersReader() {
    LocalDateTime oneMonthAgo = LocalDateTime.now().minusMonths(1);
    return new RepositoryItemReaderBuilder<Order>()
        .name("pendingOrdersReader")
        .repository(orderRepository)
        .methodName("findByStatusAndCreatedAtBefore")
        .arguments(OrderStatus.결제대기, oneMonthAgo)
        .pageSize(10)
        .sorts(Map.of("createdAt", Sort.Direction.ASC))
        .build();
}

@Bean
public ItemProcessor<Order, Order> pendingOrdersProcessor() {
    return order -> {
        if (order.getCreatedAt().isBefore(LocalDateTime.now().minusDays(7))) {
            order.ChangeOrderStatus(OrderStatus.결제만료);
        }
        return order;
    };
}

@Bean
public RepositoryItemWriter<Order> pendingOrdersWriter() {
    return new RepositoryItemWriterBuilder<Order>()
        .repository(orderRepository)
        .methodName("save")
        .build();
}

• 읽기 → 처리 → 쓰기 작업은 청크 단위로 진행되는데, 대량의 데이터를 얼만큼 끊어서 처리할지에 대한 값으로 적당한 값을 선정

한 번에 굉장히 많은 양을 읽는다면 위험성이 크다. (OOM 위험, 작업 실패지점까지 데이터를 다시 읽고 처리하는 데 오래 걸릴 수 있음) 만약, 작은 양을 읽는다면 빈번한 I/O 요청 및 네트워크 요청이 발생한다. 서버 자원에 따라 테스트를 통해 적절한 값으로 설정한다.

@FunctionalInterface
public interface ItemReader<T> {
  @Nullable
  T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException;
}

@FunctionalInterface
public interface ItemProcessor<I, O> {
  @Nullable
  O process(@NonNull I item) throws Exception;
}

@FunctionalInterface
public interface ItemWriter<T> {
  void write(@NonNull Chunk<? extends T> chunk) throws Exception;
}

• ItemReader, ItemProcessor, ItemWriter로 인터페이스를 제공하며, 다양한 구현체를 제공함. (JDBC, JPA, …)
  • https://github.com/spring-projects/spring-batch/tree/main/spring-batch-samples/src/main/java/org/springframework/batch/samples

스프링 배치 메타데이터 테이블

• 스프링 배치에서 배치 작업을 실행할 때 사용하는 테이블을 말한다. 어떤 배치 작업을 실행하는지, 언제 실행했는지 등 다양한 정보를 추적하고 관리한다.

1. BATCH_JOB_INSTANCE

• 배치 작업(Job)의 인스턴스를 관리한다.
• 주요 필드
  • JOB_INSTANCE_ID: 작업 인스턴스의 고유 식별자
  • JOB_NAME: 작업의 이름
  • JOB_KEY: 작업의 고유 키 (파라미터에 기반하여 생성됨)

SELECT JOB_INSTANCE_ID, JOB_NAME
FROM BATCH_JOB_INSTANCE
WHERE JOB_NAME = 'firstJob' AND JOB_KEY = '5bafbcc37c869ccbab63a820955f1cec';

// JOB_NAME, JOB_KEY가 일치하는 JOB_INSTANCE_ID와 JOB_NAME 가져옴

INSERT INTO BATCH_JOB_INSTANCE(JOB_INSTANCE_ID, JOB_NAME, JOB_KEY, VERSION)
    VALUES (7, 'firstJob', 'ceda08b8a0eb8ede34b1e7b2cc83a66b', 0)
;
// 배치 잡 인스턴스 추가

2. BATCH_JOB_SEQ

• 고유 식별자를 생성하고 관리한다.
• 주요 필드
  • ID: 시퀀스 값을 저장(파라미터에 따라 잡 인스턴스가 생성될 때마다 변경)
  • UNIQUEKEY: 작업 종류 식별

UPDATE BATCH_JOB_SEQ 
SET ID = LAST_INSERT_ID(ID + 1) 
LIMIT 1;
// 마지막으로 삽입된 아이디 값 + 1로 ID 갱신

3. BATCH_JOB_EXECUTION

• 각 단계 실행의 세부 정보 저장한다.
• 주요 필드
  • STEP_EXECUTION_ID: 단계 실행의 고유 식별자
  • JOB_EXECUTION_ID: 관련된 작업 실행의 ID
  • STEP_NAME: 단계 이름
  • START_TIME, END_TIME: 단계 실행의 시작 및 종료 시간
  • STATUS: 단계의 상태
  • COMMIT_COUNT, READ_COUNT, FILTER_COUNT, WRITE_COUNT: 각종 처리 정보

SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION
FROM BATCH_JOB_EXECUTION E
WHERE JOB_INSTANCE_ID = 6 
    AND JOB_EXECUTION_ID 
    IN (
        SELECT MAX(JOB_EXECUTION_ID) 
        FROM BATCH_JOB_EXECUTION E2 
        WHERE E2.JOB_INSTANCE_ID = 6
    );
// JOB_INSTANCE_ID 6이고, 가장 최근에 실행된 JOB_EXCUTION_ID에서 시작 시간, 종료 시간, 상태, 종료 메시지 등의 관련 정보 가져옴

SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION
FROM BATCH_JOB_EXECUTION
WHERE JOB_INSTANCE_ID = 6
ORDER BY JOB_EXECUTION_ID DESC;

// JOB_INSTANCE_ID 6에 대한 모든 작업 실행을 내림차순으로 검색 (가장 최근 잡 실행 목록)

INSERT INTO BATCH_JOB_EXECUTION(JOB_EXECUTION_ID, JOB_INSTANCE_ID, START_TIME, END_TIME, STATUS, EXIT_CODE, EXIT_MESSAGE, VERSION, CREATE_TIME, LAST_UPDATED)
    VALUES (7, 7, NULL, NULL, 'STARTING', 'UNKNOWN', '', 0, '2024-08-14T13:19:15.578+0900', '2024-08-14T13:19:15.578+0900')
;
// 배치 잡 실행 로우 추가

SELECT VERSION
FROM BATCH_JOB_EXECUTION
WHERE JOB_EXECUTION_ID = 7;

// JOB_EXECUTION_ID가 7인 작업 실행의 버전을 조회

UPDATE BATCH_JOB_EXECUTION
SET 
    START_TIME = '2024-08-14T13:19:15.589+0900', 
    END_TIME = NULL,  
    STATUS = 'STARTED', 
    EXIT_CODE = 'UNKNOWN', 
    EXIT_MESSAGE = '', 
    VERSION = 1, 
    CREATE_TIME = '2024-08-14T13:19:15.578+0900', 
    LAST_UPDATED = '2024-08-14T13:19:15.590+0900'
WHERE 
    JOB_EXECUTION_ID = 7 
    AND VERSION = 0;

// JOB_EXECUTION_ID가 7이고 버전이 0인 작업 실행의 상태를 업데이트(STARTING->STARTED, 0 -> 1)

4. BATCH_JOB_EXECUTION_PARMAS

• 작업 실행에 사용된 매개변수를 관리한다.

SELECT JOB_EXECUTION_ID, PARAMETER_NAME, PARAMETER_TYPE, PARAMETER_VALUE, IDENTIFYING
FROM BATCH_JOB_EXECUTION_PARAMS
WHERE JOB_EXECUTION_ID = 6;

// JOB_EXECUTION_ID가 6인 작업이 사용한 매개변수를 가져옴

5. BATCH_STEP_EXECUTION

• 각 단계 실행의 세부 정보 관리한다.
• 주요 필드
  • STEP_EXECUTION_ID: 단계 실행의 고유 식별자
  • JOB_EXECUTION_ID: 관련된 작업 실행의 ID
  • STEP_NAME: 단계 이름
  • START_TIME, END_TIME: 단계 실행의 시작 및 종료 시간
  • STATUS: 단계의 상태
  • COMMIT_COUNT, READ_COUNT, FILTER_COUNT, WRITE_COUNT: 각종 처리 메트릭

SELECT STEP_EXECUTION_ID, STEP_NAME, START_TIME, END_TIME, STATUS, COMMIT_COUNT, READ_COUNT, FILTER_COUNT, WRITE_COUNT, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, PROCESS_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED, VERSION, CREATE_TIME
FROM BATCH_STEP_EXECUTION
WHERE JOB_EXECUTION_ID = 6
ORDER BY STEP_EXECUTION_ID;

// JOB_EXECUTION_ID가 6인 작업 실행 내 각 단계의 실행 세부 정보를 검색

INSERT INTO BATCH_STEP_EXECUTION(STEP_EXECUTION_ID, VERSION, STEP_NAME, JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, COMMIT_COUNT, READ_COUNT, FILTER_COUNT, WRITE_COUNT, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT, PROCESS_SKIP_COUNT, ROLLBACK_COUNT, LAST_UPDATED, CREATE_TIME)
    VALUES (7, 0, 'firstStep', 7, NULL, NULL, 'STARTING', 0, 0, 0, 0, 'EXECUTING', '', 0, 0, 0, 0, '2024-08-14T13:19:15.601+0900', '2024-08-14T13:19:15.600+0900')
;
// 스텝 실행 인스턴스 추가

UPDATE BATCH_STEP_EXECUTION
SET 
    START_TIME = '2024-08-14T13:19:15.606+0900', 
    END_TIME = NULL, 
    STATUS = 'STARTED', 
    COMMIT_COUNT = 0, 
    READ_COUNT = 0, 
    FILTER_COUNT = 0, 
    WRITE_COUNT = 0, 
    EXIT_CODE = 'EXECUTING', 
    EXIT_MESSAGE = '', 
    VERSION = 1, 
    READ_SKIP_COUNT = 0, 
    PROCESS_SKIP_COUNT = 0, 
    WRITE_SKIP_COUNT = 0, 
    ROLLBACK_COUNT = 0, 
    LAST_UPDATED = '2024-08-14T13:19:15.606+0900'
WHERE 
    STEP_EXECUTION_ID = 7 
    AND VERSION = 0;
// STEP_EXECUTION_ID가 7이고 버전이 0인 단계 실행의 상태를 업데이트

6. BATCH_JOB_EXECUTION_CONTEXT

• 작업 실행의 컨텍스트 정보를 관리
• 주요 필드
  • JOB_EXECUTION_ID: 관련된 작업 실행의 ID
  • SHORT_CONTEXT: 짧은 컨텍스트 정보
  • SERIALIZED_CONTEXT: 직렬화된 컨텍스트 정보

SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT
FROM BATCH_JOB_EXECUTION_CONTEXT
WHERE JOB_EXECUTION_ID = 6;

// JOB_EXECUTION_ID가 6인 작업 실행의 컨텍스트 정보 가져옴

쿼리를 보고 어떻게 동작하는지 그림으로 표현해 보았다.

Job과 Step 세부 설정

1. Job

1) 순차적으로 실행하거나 조건에 따라 실행(Step Flow)

public Job footballJob(JobRepository jobRepository) {

    return new JobBuilder("footballJob", jobRepository)
             .start(playerLoad())
             .next(gameLoad())
             .next(playerSummarization())
             .build();
}

@Bean
public Job job(JobRepository jobRepository, Step stepA, Step stepB, Step stepC, Step stepD) {

    return new JobBuilder("job", jobRepository)
        .start(stepA)
        .on("*").to(stepB)
        .from(stepA).on("FAILED").to(stepC)
        .from(stepA).on("COMPLETED").to(stepD)
        .end()
        .build();
}

2) 잡 실행 전후에 특정 작업 추가(Listener)

@Bean
public JobExecutionListener jobExecutionListener() {

    return new JobExecutionListener() {
        @Override
        public void beforeJob(JobExecution jobExecution) {
                JobExecutionListener.super.beforeJob(jobExecution);
        }

        @Override
        public void afterJob(JobExecution jobExecution) {
                JobExecutionListener.super.afterJob(jobExecution);
        }
    };
}

@Bean
public Job sixthBatch() {

    return new JobBuilder("sixthBatch", jobRepository)
        .start(sixthStep())
        .listener(jobExecutionListener())
        .build();
}

2. Step

1) Tasklet과 Chunk 단위로 읽어오는 방식

(1) Tasklet

• 단순한 작업을 하나의 Tasklet으로 정의하여 수행한다.
• 전체 작업을 하나의 트랜잭션으로 관리한다.

@Bean
public Step taskletStep(StepBuilderFactory stepBuilderFactory) {
    return stepBuilderFactory.get("taskletStep")
        .tasklet((contribution, chunkContext) -> {
            // 간단한 작업 로직
            return RepeatStatus.FINISHED;
        })
        .build();
}

(2) Chunk 방식

• Chunk 방식은 데이터를 일정한 크기로 나누어 읽고, 처리하고, 저장하는 방식이다.
• Chunk 단위로 트랜잭션을 관리한다.

2) 특정 예외 건너뛰기, 다시 시도하기(Skip, Retry)

• Step의 과정 중 예외가 발생하게 되면 예외를 특정수까지 건너뛸 수 있도록 설정한다.
• Retry는 Step의 과정 중 예외가 발생하게 되면 예외를 특정수까지 반복할 수 있도록 설정한다.

@Bean
public Step skipStep() {

    return new StepBuilder("skipStep", jobRepository)
            .<BeforeEntity, AfterEntity> chunk(10, platformTransactionManager)
            .reader(beforeSixthReader())
            .processor(middleSixthProcessor())
            .writer(afterSixthWriter())
            .faultTolerant()
            .skip(Exception.class)
            .noSkip(FileNotFoundException.class)
            .noSkip(IOException.class)
            .skipLimit(10)
            .build();
}

public Step retryStep() {

        return new StepBuilder("retryStep", jobRepository)
            .<BeforeEntity, AfterEntity>chunk(10, platformTransactionManager)
            .reader(beforeReader())
            .processor(processor())
            .writer(afterWriter())
            .faultTolerant()
            .retry(Exception.class)
            .retryLimit(3)
            .noRetry(FileNotFoundException.class)
            .noRetry(IOException.class)
            .build();
}

3) WRITER 트랜잭션 제어

• Writer시 특정 예외에 대해 트랜잭션 롤백을 제외할 수 있다.

@Bean
public Step noRollbackStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {

    return new StepBuilder("noRollbackStep", jobRepository)
        .<String, String>chunk(2, transactionManager)
        .reader(itemReader())
        .writer(itemWriter())
        .faultTolerant()
        .noRollback(ValidationException.class)
        .build();
}

4) Step 전후에 특정 작업 추가(Listener)

@Bean
public StepExecutionListener stepExecutionListener() {

    return new StepExecutionListener() {
        @Override
        public void beforeStep(StepExecution stepExecution) {
            StepExecutionListener.super.beforeStep(stepExecution);
        }

        @Override
        public ExitStatus afterStep(StepExecution stepExecution) {
            return StepExecutionListener.super.afterStep(stepExecution);
        }
    };
}

@Bean
public Step stepListerStep() {

    return new StepBuilder("stepListerStep", jobRepository)
            .<BeforeEntity, AfterEntity> chunk(10, platformTransactionManager)
            .reader(beforeSixthReader())
            .processor(middleSixthProcessor())
            .writer(afterSixthWriter())
            .listener(stepExecutionListener())
            .build();
}

스프링 배치 실행 방법

1. 컨트롤러

@RestController
@RequiredArgsConstructor
public class BatchController {

  private final JobLauncher jobLauncher;
  private final JobRegistry jobRegistry;

  @GetMapping("/jobs/cancel-orders")
  public String firstApi(@RequestParam("value") String value) throws Exception {

    JobParameters jobParameters = new JobParametersBuilder()
      .addString("date", value)
      .toJobParameters();

    jobLauncher.run(jobRegistry.getJob("cancelPendingOrdersJob"), jobParameters);

    return "ok";
  }
}

2. 스케줄러

@Configuration
@RequiredArgsConstructor
public class AggregationScheduler {

  private final JobLauncher jobLauncher;
  private final JobRegistry jobRegistry;

  @Scheduled(cron = "0 0 2 15 * *", zone = "Asia/Seoul")
  public void getSalesAndRefunds() throws Exception {

    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM");
    String date = dateFormat.format(new Date());

    JobParameters jobParameters = new JobParametersBuilder()
      .addString("date", date)
      .toJobParameters();

    jobLauncher.run(jobRegistry.getJob("salesAndRefundsJob"), jobParameters);
  }
}

Programmers Lv3, 가장 긴 팰린드롬

핵심

• 입력의 크기가 2,500이라 대략 $O(N^2)$이하 알고리즘을 사용한다. 앞뒤를 뒤집어도 똑같은 문자열을 팰린드롬(palindrome)이라고 한다. 문자열 s가 주어질 때, s의 부분 문자열 중 가장 긴 팰림드롬의 길이를 반환한다.
• 팰린드롬 여부를 확인하는 함수는 아래와 같이 구할 수 있다.

  boolean isPalinDrome(String s, int l, int r) {
    while (l <= r) {
        if (s.charAt(l) == s.charAt(r)) {
            ++l;
            --r;
        } else {
            return false;
        }
    }
    return true;
  }

1. 모든 부분 문자열 구한 후 팰린드롬 구하기

• 가능한 부분 문자열의 시작 인덱스, 끝 인덱스를 구한 후 팰린드롬 여부를 검사한다. 가능한 좌표는 아래와 같이 이중 반복문으로 모두 구할 수 있다. 팰린드롬을 만들 수 있다면, 가장 긴 길이가 되도록 갱신한다.

  for (int i = 0; i < n; i++) {
    for (int j = i; j < n; j++) {
        if (isPalinDrome(s, i, j)) {
            int len = j - i + 1;
                if (ans < len) {
                    ans = len;
                }
            }
        }
    }
  }

시간복잡도

• 시간복잡도는 $O(n^3)$에 동작하지만, n이 크지 않고 팰린드롬 함수에서 매번 최대 길이를 검사하지 않고, 인덱스가 두 개씩 이동해서 그런지 생각보다 빠르게 동작한다. n이 더 컸다면 시간 초과가 났을 것이다.

전체 코드

class Solution {
    public int solution(String s) {
        int n = s.length();

        int ans = 1;
        for (int i = 0; i < n; i++) {
            for (int j = i; j < n; j++) {
                if (isPalinDrome(s, i, j)) {
                    int len = j - i + 1;
                    if (ans < len) {
                        ans = len;
                    }
                }
            }
        }
        return ans;
    }

    boolean isPalinDrome(String s, int l, int r) {
        while (l <= r) {
            if (s.charAt(l) == s.charAt(r)) {
                ++l;
                --r;
            } else {
                return false;
            }
        }
        return true;
    }
}

2. 중심으로부터 팰린드롬 길이 확장하기

• 중심 인덱스부터 팰린드롬의 길이를 구하는 방식으로 최적화할 수 있다. 하나의 중심으로부터 시작할 수 있고(홀수 팰린드롬), 나란히 붙어 있는 두 개의 중심(짝수 팰린드롬)으로 시작할 수 있다.
• 문자열의 길이가 7이라면, 아래와 같은 인덱스에서 퍼져나간다고 생각하면 된다.

  // helloeh 7
  홀수: {0, 0} -> h                   
  짝수: {0, 1} -> h e
  홀수: {1, 1} -> e
  짝수: {1, 2} -> e l
  ...
  홀수: {6} -> h
  짝수: {6, 7} -> h, outofbound

• 홀수 시작 좌표와 짝수 시작 좌표에 대한 팰린드롬 길이를 구한다. 기존 팰린드롬 로직에서 길이 로직만 추가되었다. 홀수 팰린드롬을 구할 때 시작 좌표가 같아도 길이를 +2하므로 결과값에서 -1을 한다.

  for (int i = 0; i < n; i++) {
    ans = Math.max(ans, getPalindromeLength(s, i, i) - 1);
    ans = Math.max(ans, getPalindromeLength(s, i, i + 1));
  }
  

int getPalindromeLength(String s, int l, int r) { int n = s.length();

int len = 0;
while (l >= 0 && r < n) {
    if (s.charAt(l) == s.charAt(r)) {
        len += 2;
        --l;
        ++r;
    } else {
        break;
    }
}
return len;

}

### 시간복잡도
- $O(n^2)$
![](https://velog.velcdn.com/images/ji-jjang/post/7f5790ba-8834-4259-a1de-f65761e77cf3/image.png)


### 전체 코드
```java
class Solution {
    public int solution(String s) {
        int n = s.length();

        int ans = 0;
        for (int i = 0; i < n; i++) {
            ans = Math.max(ans, getPalindromeLength(s, i, i) - 1);
            ans = Math.max(ans, getPalindromeLength(s, i, i + 1));
        }

        return ans;
    }

    int getPalindromeLength(String s, int l, int r) {
        int n = s.length();

        int len = 0;
        while (l >= 0 && r < n) {
                if (s.charAt(l) == s.charAt(r)) {
                        len += 2;
                        --l;
                        ++r;
                } else {
                        break;
                }
        }

        return len;
    }
}

3. DP

• 모든 길이가 1인 부분 문자열은 팰린드롬이고, 길이가 2일 때 두 문자가 동일한 경우에도 팰린드롬이다.
• 길이가 3 이상인 경우 양 끝 문자가 같고, 사이에 있는 부분 문자열이 팰린드롬일 때 해당 부분 문자열은 팰린드롬이다.

  dp[i][i] = true // 길이가 1인 부분 문자열은 팰린드롬
  

if (s[i] == s[i + 1]) { dp[i][i + 1] = true // 두 문자가 같다면 i와 i+1 부분 문자열은 팰린드롬
ans = 2; }

- 예를 들어 길이가 3일 때 i가 0이고, j가 2인 경우 s[i]와 s[j]가 같고, i와 j 사이에 있는 부분 문자열이 팰린드롬이어야 한다. 따라서 dp[i + 1][j - 1]이 true일 때 dp[i][j]가 true가 될 수 있다.
![](https://velog.velcdn.com/images/ji-jjang/post/7fd7398e-9660-4f95-8932-2ba13156c514/image.png)

### 시간복잡도
- $O(n^2)$
![](https://velog.velcdn.com/images/ji-jjang/post/547dfc74-1d4e-422e-a119-f36497749838/image.png)

### 전체 코드
```java
class Solution {
    public int solution(String s) {
        int n = s.length();

        boolean[][] dp = new boolean[n][n];
        int ans = 1;

        for (int i = 0; i < n; ++i) {
            dp[i][i] = true;
            if (i < n - 1 && s.charAt(i) == s.charAt(i + 1)) {
                dp[i][i + 1] = true;
                ans = 2;
            }
        }

        for (int len = 3; len <= n; ++len) {
            for (int i = 0; i < n - len + 1; ++i) {
                int j = i + len - 1;

                if (dp[i + 1][j - 1] && s.charAt(i) == s.charAt(j)) {
                    dp[i][j] = true;
                    ans = len;
                }
            }
        }

        return ans;
    }
}

• 운영체제에서 실행 중인 하나의 애플리케이션을 프로세스(process)라고 한다.
• 스레드(thread)는 하나의 실행 흐름을 말하며 모든 자바 애플리케이션은 메인 스레드(main thread)가 main() 메소드를 실행하며 시작한다.

  public static void main(String[] args) {     // main thread 실행
  ...
  }

데몬 스레드와 유저 스레드

• 자바 스레드는 크게 데몬 스레드와 유저 스레드로 나뉜다.

• 데몬 스레드(Daemon Thread)는 JVM이 종료될 때 같이 종료. 일반적으로 백그라운드에서 실행되며, 메인 스레드와 사용자 스레드가 종료되면 자동으로 종료된다. 그 이유는 주 스레드의 보조 역할을 수행하는데 주 스레드가 종료되면 데몬 스레드의 존재 이유가 없어지기 때문이다.

• 유저 스레드(User Thread)는 메인 스레드가 종료되더라도 독립적으로 실행 가능하고, 모든 사용자 스레드가 종료되지 않으면 JVM은 종료되지 않는다.

  public static void main(String[] args) {
    Thread userTread =
            new Thread(
                    () -> {
                        try {
                            Thread.sleep(3000);
                            System.out.println("UserThread finished");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    });
  
    System.out.println("MainThread finished");
  }
  // MainThread finished
  // UserThread finished

public static void main(String[] args) {
    Thread daemonThread =
        new Thread(
            () -> {
                try {
                    Thread.sleep(3000);
                    System.out.println("DaemonThread finished");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

    daemonThread.setDaemon(true);
    daemonThread.start();

    System.out.println("MainThread finished");
}
// MainThread finished

멀티 스레드

• 메인 작업 이외에 추가적인 병렬 작업의 수만큼 스레드를 생성할 수 있다.

• java.lang.Thread 클래스를 직접 객체화해서 생성(runnable interface)해도 되고, Thread를 상속해서 하위 클래스를 만드는 방법도 가능하다.

  1. Thread 상속

  public class MultiThread {
  static class MyThread extends Thread{
    @Override
    public void run() {
      try {
        Thread.sleep(3000);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
      System.out.println("extended userThread is finished");
    }
  }
  
  public static void main(String[] args){
    Runnable runnable =
        new Runnable() {
          @Override
          public void run() {
            try {
              Thread.sleep(2000);
            } catch (InterruptedException e) {
              throw new RuntimeException(e);
            }
            System.out.println("runnable userThread is finished");
          }
        };
  
    Thread thread = new Thread(runnable);
    thread.start(); // implement ruunable interface
  
    MyThread myThread = new MyThread();
    myThread.start(); // extend Thread class
  
    System.out.println("mainThread is finished");
  }
  }

  2. Runnable 익명 객체로 사용

• 위에서 본 것처럼 Runnable을 익명 객체(람다)로 사용할 수 있다.

  Thread thread = new Thread(() -> {
    try {
        Thread.sleep(3000);
        System.out.println("UserThread finished");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
  });

동시성과 병렬성

• 동시성이란 멀티 작업을 위해 하나의 코어에서 멀티 스레드가 번갈아 가며 실행하는 성질을 말하며, 병렬성은 멀티 작업을 위해 멀티 코어에서 개별 스레드를 동시에 실행하는 성질. 싱클 코어 CPU를 이용한 멀티 스레드 작업은 병렬적으로 실행되는 것으로 보이지만, 사실은 번갈아 가며 실행되는 동시성 작업이다.

우선순위(priority) 방식과 순환 할당(Round-Robin) 방식

• 자바 스레드 스케줄링은 우선순위 방식과 순환 할당 방식을 사용한다.

• 우선순위 방식은 스레드 객체에 우선 순위 번호를 부여하는 방식으로 동작한다. 자바 코드로 설정가능하지만, 스레드 스케줄링은 JVM이 운영체제 기능을 활용하는 방식으로 동작하기에 운영체제마다 동작 방식이 달라질 수 있다.

  myThread.setPriority(1); // 1 ~ 10값을 줌. 높을수록 실행 기회를 더 많이 가짐. 상수로 설정 권

  public class PriorityThread {
  
  public static void main(String[] args){
  
    Thread thread = new Thread(() -> {
      for(int i = 0; i < 50; i++) {
        System.out.println("priority1 thread running.");
      }
    });
    Thread thread2 = new Thread(() -> {
      for(int i = 0; i < 50; i++) {
        System.out.println("priority10 thread running.");
      }
    });
  
    thread.setPriority(1);
    thread2.setPriority(10);
  
    thread.start();
    thread2.start();
  
    System.out.println("mainThread is finished");
  }
  }
  

// 출력결과를 보면 thread1이 실행되다가 thread2가 실행되면 thread2를 우선적으로 처리하고 thread1이 마무리가 됌 // priority1 thread running. // priority1 thread running. // priority1 thread running. // priority1 thread running. // priority10 thread running. // priority10 thread running. // priority10 thread running. // priority10 thread running. ... // mainThread is finished // priority1 thread running. ...

- 순환 할당 방식은 JVM에 의해 관리되기에 직접 자바 코드로 제어할 수 없음

# 멀티스레드와 병행 제어
## 동기화 필요성과 synchronized 키워드
- 멀티스레드 환경에서 스레드가 사용 중인 객체를 다른 스레드가 변경할 수 없도록 하려면 스레드 작업이 끝날 때까지 객체에 잠금을 걸어 다른 스레드가 사용할 수 없도록 해야 한다.

### 1. 임계 영역
- 멀티스레드 프로그램에서 단 하나의 스레드만 실행할 수 있는 코드 영역을 임계 영역(critical season)이라고 한다.

### 2. 동기화 메소드 및 블록
- 자바에선 임계 영역을 지정하기 위해 동기화(synchonized) 메소드와 동기화 블록을 제공한다.
- 동기화 메소드를 만드는 방법은 메소드 선언에 synchronized 키워드를 붙이면 된다. 인스턴스와 정적 메소드 어디든 붙일 수 있다.
```java
public sysnchronized void method() {
    // 임계 영역 
}

public void method() {
    // 공유 영역

    synchronized(공유 객체) {
        // 임계 영역
    }

    // 공유 영역
}

• 동기화 블록 외부 코드들은 여러 스레드가 동시에 실행할 수 있지만, 동기화 블록 내부 코드는 한 번에 하나의 스레드만 실행 할 수 있다.
• 동기화 메소드로 작성된 것을 동기화 블록으로도 만들 수 있다.

  public synchronized void setMemory(int memory) {
    this.memory = memory;
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {}
    System.out.println(Thread.cuurentThread().getName() + " " + this.memory);
  }
  

public void setMemory(int memory) { synchronized (this) { this.memory = memory; try { Thread.sleep(2000); } catch (InterruptedException e) {} System.out.println(Thread.cuurentThread().getName() + " " + this.memory); } }

- 일반적으로 동기화 메서드는 메서드 전체에 락을 걸기 때문에 메서드 블록보다 오버헤드가 크다고 한다. 메서드 특정 부분이 락을 걸 필요가 없다면, 동기화 블록을 사용해 락을 거는 부분을 최소화하는 것이 좋다.

## 스레드 상태와 스레드 상태 제어
### 1. 스레드 상태
- NEW(생성 및 대기): 스레드가 생성되었지만 아직 실행되지 않은 상태. 스레드 객체를 생성하고, start() 메소드를 호출하면 스레드가 실행되는 것처럼 보이지만 사실은 실행 되기 상태로 스케줄링이 되지 않아 실행을 기다리고 있다.
- RUNNING(실행): 실행 대기 상태에 있는 스레드 중 스레드 스케줄링으로 선택된 스레드가 CPU를 점유하고 run() 메소드를 실행하여 Running 상태가 된다.
- TERMINATED(종료): 실행 상테에서 run() 메소드가 종료되면, 더 이상 실행할 코드가 없기에 스레드 실행은 멈추고 스레드는 TERMINATED 상태가 된다.
- 일시 정지 상태
    - WAITING: 다른 스레드의 작업이 완료되기를 기다리는 상태
    - TIME_WATING: 주어진 시간 동안 기다리는 상태
    - BLOCKED: 락이 풀리기를 기다리는 상태 
### 2. 스레드 상태 제어 메서드 
#### 1) 주어진 시간 동안 일시 정지(sleep)
- 실행 중인 스레드를 일정 시간 동안 멈추게 하고 싶다면 Trhead 클래스의 정적 메소드인 sleep()을 사용한다.
```java
try {
    Thread.sleep(1000);
} catch (InterruptedException e) {
    interrupt(1000); 
}

• 매개 값에는 얼마 동안 일시 정지 상태로 있을지 밀리세컨드 단위로 시간을 준다.

  2) 다른 스레드에 실행 양보(yield)

• 아래 코드처럼 스레드가 특정 작업을 반복하여 실행할 때 무의미한 작업을 하는 경우가 있다. (BusyWaiting) 이때 다른 스레드에 CPU를 양보하면 CPU를 좀 더 효율적으로 사용할 수 있다.

  public void run() {
    while (true) {
        if (work) {
            System.out.println("TrheadA 작업 내용");
        } else {
            Thread.yield();
        }
    }
  }

• yield() 메소드를 호출한 스레드는 실행 대기 상태로 돌아가고 동일한 우선순위 또는 높은 우선순위를 갖는 다른 스레드가 실행 기회를 가질 수 있도록 한다.

3) 다른 스레드 종료 기다림(join)

• 스레드는 다른 스레드와 독립적으로 실행하는 것이 기본이지만 다른 스레드가 종료될 때까지 기다렸다가 실행해야 하는 경우가 발생할 수 있다. 예를 들어 모든 계산 결과값을 받아 합계를 계산하는 스레드를 생각해 보자. 다른 스레드의 모든 계산 결과값을 기다리기 위해 Thread.join() 메서드를 사용한다.

public class JoinExample {
    public static void main(String[] args) {
        SumThread sumThread = new SumThread();
        sumThread.start();

        try {
            sumThread.join();
        } catch (InterruptedException e) {}
    }
}

4) 스레드 간 협업(wait, notify, notifyAll)

• 두 개의 스레드를 교대로 번갈아가며 실행하고 싶을 때 자신의 작업이 끝나면 상대방 스레드를 일시 정지 상태에서 풀어주고, 자신은 일시 정지 상태로 만드는 것이다.

• 공유 객체는 두 스레드가 작업할 내용을 각각 동기화 메소드로 구분한다. 한 스레드가 작업을 완료하면 notify() 메소드를 호출해서 다른 스레드를 일시정지에서 실행 대기 상태로 만들고, 자신은 wait() 메소드를 호출해서 일시 정지 상태로 만든다.

  public class WorkObject {
    public synchronized void methodA() {
        notify();
        try {
            wait();
        } catch (InterruptedExcpetion e) {}
    }
  
    public synchronized void methodB() {
        notifiy();
        try {
            wait();
        } catch (InterruptedException e) {}
    }
  }

  public class ThreadA extends Thread {
    private WorkObject workObject;
  
    // 공유 객체를 매개값으로 받아 필드에 저장
    public ThreadA(WorkObject workObject) {
        this.workObject = workObject; 
    }
  
    @Override
    public void run() {
        workObject.methodA();
    }
  }

  public class ThreadB extends Thread { ... }
  

public static void main(String[] args) { WorkObject sharedObject = new WorkObject();

ThreadA threadA = new ThreadA(sharedObject);
TrehadB threadB = new TrehadB(sharedObject);

threadA.start();
threadB.start();
// 번갈아 가며 실행

}

#### 5) 스레드 안전한 종료(stop, interrupt)
-  스레드는 자신의 run() 메소드가 모두 실행되면 자동으로 종료되지만, 필요에 따라 실행 중인 스레드를 즉시 종료할 필요가 있다.
- Thread를 즉시 종료시키기 위해 stop() 메서드를 제공하는데, 이 메소드는 스레드가 사용 중인 자원을 불안정한 상태로 남겨두기 때문에 deprecated 되었다.
- stop flag를 이용해 run() 메소드의 종료를 유도하는 방법을 사용하거나 앞에서 본 interrupt() 메서드를 이용해 스레드가 일시 정지 상태에서 InterruptedException()을 발생시켜 정상 종료할 수 있다.

> wai(), notify(), notifyAll()은 Object 클래스의 메소드이고, 그 이외는 Thread 클래스의 메소드들이다.

# 스레드 그룹
- 스레드 그룹은 관련된 스레드를 묶어서 관리할 목적으로 이용된다. JVM이 실행되면 system 스레드 그룹을 만들고, JVM 운영에 필요한 스레드들을 생성해서 system thread 그룹에 포함한다. 그리고 system의 하위 스레드 그웁으로 main을 만들고 메인 스레드를 main 스레드 그룹에 포함한다.
- 스레드는 반드시 하나의 스레드 그룹에 속하는데, 명시적으로 스레드 그룹에 포함하지 않으면 기본적으로 자신을 생성한 스레드와 같은 그룹에 속하게 된다. 대부분 작업 스레드는 main 스레드가 생성하므로 기본적으로 main 스레드 그룹에 속하게 된다.

## 1. 그룹 이름 조회
```java
TrheadGroup group = Thread.currentThread().getThreadGroup();
String groupName = group.getName();

• Thread의 정적 메소드인 getAllStackTraces()를 이용하면 프로세스 내 실행하는 모든 스레드에 대한 정보를 알 수 있다.

  Map<Trehad, StackTraceElement[]> map = Thread.getAllStackTraces();

• 키는 스레드 객체이고, 값은 스레드 상태 기록이 갖고 있는 StackTraceElement 배열이다.

2. 그룹 생성

• 명시적으로 스레드 그룹을 만들고 싶다면 다음 생성자 중 하나를 이용해서 TrheadGroup 객체를 만든다. TrheadGroup 이름을 주거나, 부모 ThreadGroup과 이름을 매개값으로 줄 수 있다.

  ThreadGroup tg = new ThreadGroup(string name);
  ThreadGroup tg = new ThreadGroup(ThreadGroup parent, string name);

• 새로운 스레드 그룹을 생성한 후 이 그룹에 스레드를 포함하려면 Thread 객체를 생성할 대 생성자 매개값으로 스레드 그룹을 지정하면 된다.

  Thread t = new Thread(ThreadGroup group, Runnable target);
  Thread t = new Thread(ThreadGroup group, Runnable target, String name);
  Thread t = new Thread(ThreadGroup group, Runnable target, String name, long stackSize);
  Thread t = new Thread(ThreadGroup group, String name);

3. 그룹 일괄 인터럽트

• 여러 스레드를 하나의 그룹에 포함하면 어떤 이점이 있을까? 스레드 그룹에서 제공하는 interrupt() 메서드를 이용하면 그룹 내에 포함된 모든 스레드들을 일괄 interrupt()할 수 있다.
• 이외에도 제공하는 여러 메서드가 있다.
  • activeCount(): 현재 그룹 및 하위 글춥에서 활동 중인 모든 스레드 수 리턴
  • activeGroupCount(): 현재 그룹에서 활동 중인 모든 하위 그룹 수 리턴
  • checkAccess(): 현재 스레드가 스레드 그룹을 변경할 권한이 있는지 검사, 권한이 없으면 SecurityException을 발생
  • destroy(): 현재 그룹 및 하위 그룹을 모두 삭제. 단 그룹 내 포함된 모든 스레드가 종료 상태이어야 함는 최대 우선순위를 리턴

스레드 풀

• 갑작스러운 병렬 작업의 폭증으로 인한 오버헤드를 피하려면 스레드풀(ThreadPool)을 사용해야 한다.

스레드 풀 필요성

• 스레드 풀은 작업 처리에 사용되는 스레드를 제한된 개수만큼 정해 놓고 작업 큐에 들어오는 작업을 하나씩 스레드가 맡아 처리한다. 작업이 끝난 스레드는 다시 작업 큐에서 새로운 작업을 가져와 처리하기에 작업 처리 요청이 폭증되어도 스레드 전체 개수가 늘어나지 않아 애플리케이션 성능이 급격히 나빠지지 않게 된다.
• 따라서 매번 요청마다 새로운 스레드를 생성하는 오버헤드를 피하기 위해 스레드 풀이 필요하다.

스레드 풀 장단점

• 장점
  • 스레드 생성과 소멸에 드는 비용을 줄일 수 있다.
  • 스레드의 수를 제한하여 시스템 자원을 효율적으로 사용할 수 있다.
  • 요청마다 스레드를 생성하지 않기에 응답 시간이 빠르다.
• 단점
  • 스레드 풀의 크기를 부적절하게 설정하면 스레드가 부족하거나 너무 많이 생성될 수 있다. 스레드가 부족하면 지연 시간이 길어지고, 스레드가 과도하게 생성되면 시스템 자원이 낭비된다.
  • 스레드 풀을 잘못 사용하면 특정 스레드가 서로 자원만을 요청하는 데드락 상태에 빠질 수 있다.

스레드 풀 사용 방법

1. 스레드 풀 생성

1) newCachedThreaedPool()

• 초기 스레드 수: 0, 코어 스레드 수: 0, 최대 스레드 수: Integer.MAX_VALUE
• 초기 스레드 수는 ExecutorService 객체가 생성될 때 기본적으로 생성되는 스레드 수를 말하며, 코어 스레드 수는 스레드 수가 증가한 후 사용되지 않는 스레드를 스레드풀에서 제거할 때 최소한 유지해야 할 스레드 수를 의미한다. 최대 스레드 수는 스레드 풀에서 관리하는 최대 스레드 개수이다.
• 1개 이상의 스레드가 추가되었을 때 60초 동안 추가된 스레드가 아무 작업을 하지 않으면 추가된 스레드를 종료하고 풀에서 제거한다.

  ExecutorService executorService = Executors.newCachedTrheadPool();

2) newFixedThreadPoll(int nThreads)

• 초기 스레드 수: 0, 코어 스레드 수: nThreads, 최대 스레드 수: nThreads
• 스레드 개수보다 작업 개수가 많으면 새 스레드를 생성시키고 작업을 처리한다. 최대 스레드 개수는 매개값으로 준 nTrheads이다. 스레드가 작업을 처리하지 않고 놀고 있더라도 스레드 개수가 줄지 않는다.

  // CPU 코어 개수만큼 스레드 생성
  ExecutorService executorService = Executors.newFixedThreadPool(
    Runtime.getRuntime().availableProcessors()
  );

3) 직접 스레드풀 생성

• ThreadPoolExecutor 객체를 생성하여 초기 스레드 개수 0개, 코어 스레드 개수 3개, 최대 스레드 개수가 100개인 스레드 풀을 생성해 보자. 그리고 코어 스레드 3개를 제외한 나머지 추가된 스레드가 120초 동안 놀고 있으면 해당 스레드를 제거하여 스레드 수를 관리한다.

  ExecutorService ThreadPool = new ThreadPoolExecutor(
    3, // 코어 스레드 개수
    100, // 최대 스레드 개수
    120L, // 놀고 있는 시간
    TimeUnit.SECONDS, /// 놀고 있는 시간 단위
    new SynchronousQueue<Runnable> // 작업 큐
  );

2. 스레드풀 종료

• 스레드풀의 스레드는 데몬 스레드가 아니기에 main 스레드가 종료되더라도 작업을 처리하기 위해 계속 실행 상태로 남아있다.
• ExecutorService는 종료와 관련해서 다음 3개 메서드를 제공한다.
  • shutdown(): 현재 처리 중인 작업뿐만 아니라 작업 큐에 대기하고 있는 모든 작업을 처리한 뒤에 스레드 풀을 종료
  • shutdownNow(): 현재 작업 처리 중인 스레드를 interrupt해서 작업 중지를 시도하고 스레드 풀을 종료시킨다. 리턴값은 작업 큐에 있는 미처리된 작업(Runnable)의 목록이다.
  • awaitTermination(long timeout, TimeUnit unit): shutdown() 메소드 호출 이후, 모든 작업 처리를 timeout 시간 내에 완료하면 true를 리턴하고, 완료하지 못하면 작업 처리 중인 스레드를 interrupt하고 false를 리턴한다.

일반적으로 남아있는 작업을 마무리하고 종료할 때는 shutdown(), 남아있는 작업과는 상관없이 강제로 종료할 때는 shutdownNow()를 호출한다.

3. 작업 생성과 처리 요청

1) 작업 생성

• 하나의 작업은 Runnable 또는 Callable 구현 클래스로 표현한다. 작업 처리 완료 후 리턴값 여부에 따라 사용하며 없다면 Runnable, 있다면 Callable을 사용한다.

  Runnable task = new Runnable() {
    @Override
    public void run() {
        // 작업
    }
  }
  

Callbale task = new Callable() { @Override public T call() throws Exception { // 작업 return T; } }

#### 2) 작업 처리 요청
- 작업 처리 요청이란 ExecutorService 작업 큐에 Runnable 또는 Callable 객체를 넣는 행위를 말한다.
- ExecutorService는 작업 처리 요청을 위해 다음 두 가지 메서드를 제공한다.
  - execute(Runnable command): Runnable을 작업 큐에 저장. 작업 처리 결과를 받지 못한다.
  - submit(Runnable task): Runnable 또는 Callble을 작업 큐에 저장. 리턴된 Future를 통해 작업 처리 결과를 얻을 수 있다.

> execute()는 작업 처리 도중 예외가 발생하면 스레드가 종료되고 해당 스레드는 스레드 풀에서 제거되는 반면, submit()은 작업 처리 도중 예외가 발생하더라도 스레드는 종료되지 않고 다음 작업을 위해 재사용된다.


### 4. 블로킹 방식 작업 완료 통보
- Future를 이용한 블로킹 박식의 작업 완료 통보에서 주의할 점은 작업을 처리하는 스레드가 작업을 완료하기 전까지는 get() 메소드가 블로킹 되므로 다른 코드를 실행할 수 없다.
- Future 객체는 작업 결과를 얻기 위한 get() 메소드 외에도 다음과 같은 메소드를 제공한다.
  - caccel(boolean mayInterruptIfRunning): 작업 처리가 진행 중일 경우 취소
  - isCancelled(): 작업이 취소되었는지 여부
  - isDone(): 작업 처리가 완료되었는지 여부

#### 1) 리턴값이 없을 때
- 결과값이 없는 작업 처리 요청은 submit(Runnable task) 메소드를 이용한다. 결과값이 없음에도 Future 객체를 리턴하는 이유는 스레드가 작업 처리를 정상적으로 완료했는지, 아니면 작업 처리 도중 예외가 발생했는지 확인하기 위해서다.
```java
Future future = executorService.submit(task);

• 작업 처리가 정상적으로 완료되었다면 Future의 get() 메서드는 null을 리턴한다.

  2) 리턴값이 있을 때

• 스레드풀의 스레드가 작업을 완료한 후 처리 결과를 얻어야 한다면 작업 객체를 Callable로 생성한다.

  Future<T> future = executorService.submit(task);

3) 작업 처리 결과를 외부 객체 저장

• 스레드가 작업 처리를 완료하고 외부 Result 객체에 작업 결과를 젖아하면 애플리케이션이 Result 객체를 사용해서 특정 작업을 진행할 수 있을 것이다. 대개 Result 객체는 공유 객체가 되어 두 개 이상의 스레드 작업을 취합할 목적으로 사용한다. 이러한 작업을 위해 ExecutorService의 submit(Runnable task, V result) 메서드를 사용할 수 있는데, V가 바로 Result가 된다.

  Result result = ...;
  Runnable task = new Task(result);
  Future<Result> future = executorService.submit(task, result);
  result = future.get();

4) 작업 완료 순으로 통보

• 작업 요청 순서대로 작업 처리가 완료된 것이 아니고, 작업의 양과 스레드 스케줄링에 따라 먼저 요청한 작업이 나중에 완료되는 경우도 발생한다.
• 스레드 풀에서 작업 처리가 완료된 것만 통보받는 방법이 있다. CompletionService를 이용하면 처리 완료된 작업을 가져오는 poll()과 take() 메소드를 사용할 수 있다.

  ExecutorService executorService = Executors.newFixedThreadPoll(
    Runtime.getRuntime().availableProcessors();
  );
  

CompletionService completionService = new ExecutorCompletionService( executorService );

completionService.submit(Callable task); completionService.submit(Runnable task, V result);

### 4. 콜백 방식 작업 완료 통보
- 블로킹 방식은 작업 처리를 요청한 후 작업이 완료될 때까지 블로킹되지만, 콜백 방식은 작업 처리를 요청한 후 결과를 기다릴 필요 없이 다른 기능을 수행할 수 있다.
- ExecutorService는 콜백을 위한 별도의 기능을 제공하진 않지만, Runnable 구현 클래스를 작성할 때 콜백 기능을 구현할 수 있다. 직접 정의할 수도 있고 java.nio.channelsCompletionHandler를 이용할 수도 있다.
```java
CompletionHandler<V, A> callback = new CompletionHandler<V, A>() {
    @Override
    public void completed(V result, A attachment) {}

    @Override
    public void failed(Throwable exc, A attachment) {}
}

• completed()는 작업을 정상 처리했을 때 호출되는 메서드이고, failed()는 작업 처리 도중 예외가 발생했을 때 호출되는 콜백 메소드이다.
• CompletionHandler의 V 타입 파라미터는 결과값 타입이고, A는 첨부값 타입이다. 첨부값은 콜백 메소드에 결과값 이외에 추가적으로 전달하는 객체라고 생각하면 된다. 첨부 값이 필요 없다면 Void를 전달한다.

Programmers Lv3, n + 1 카드게임

핵심

• n개의 카드로 이루어진 카드 뭉치와 여러 코인이 주어진다. 아래 규칙에 따라 진행된다.
  • 처음에 카드 뭉치에서 카드 n/3장을 뽑아 모두 가집니다. (n은 6의 배수입니다.) 당신은 카드와 교환 가능한 동전 coin개를 가지고 있습니다.
  • 게임은 1라운드부터 시작되며, 각 라운드가 시작할 때 카드를 두 장 뽑습니다. 카드 뭉치에 남은 카드가 없다면 게임을 종료합니다. 뽑은 카드는 카드 한 장당 동전 하나를 소모해 가지거나, 동전을 소모하지 않고 버릴 수 있습니다.
  • 카드에 적힌 수의 합이 n+1이 되도록 카드 두 장을 내고 다음 라운드로 진행할 수 있습니다. 만약 카드 두 장을 낼 수 없다면 게임을 종료합니다.
• 위 규칙을 적용하며 최대 진행할 수 있는 라운드를 구하면 된다.

DFS (시간 초과)

• 가장 직관적으로 접근할 수 있는 풀이다. 라운드마다 4가지 경우를 선택할 수 있다.
  • 왼쪽 카드를, 코인을 사용해 갖기
  • 오른쪽 카드를, 코인을 사용해 갖기
  • 둘 다 갖기
  • 둘 다 갖지 않기
• n이 최대 999개일 떄, 라운드가 최대 666번 진행한다. 한 라운드마다 4번의 경우가 있으므로 $4^{666}$ 경우의 수가 나온다. 가지치기 하더라도, 워낙 경우가 많기 때문에 시간 초과라고 판단하였다.

Greedy

• 모든 경우를 탐색하는 건 위에서 살펴본 것처럼 매우 많은 경우가 나온다. 매 순간 최적의 선택을 하는 그리디 알고리즘을 적용해 볼 수 있다. 코인을 최대한 사용하지 않으면서, 두 카드를 이용해 n+1 숫자를 만들 수 없을 때 코인을 사용하면 되지 않을까? 효율적으로 코인을 사용하기 위해 코인 하나만 사용하여 (n + 1 - 기존 가지고 있는 카드번호)인 카드를 카드 더미에서 찾아볼 수 있다. 카드 더미에서 찾을 수 없다면, 이번에는 코인 2개를 사용하여 카드 더미에서 2개를 골라 n+1 숫자를 만들 수 있는지 찾는다. n+1 숫자를 만들 수 없을 때까지 진행하면 최대 가능한 라운드를 구할 수 있다.

• 구체적인 코드 작성 과정은 다음과 같다.

  1. 처음에 n/3 장을 손에 가진다.
  2. 손에 있는 카드 중 두 장이 target(n+1)이 되는지 확인 (List hand)
  3. 라운드마다 2개의 카드를 카드 더미에 카드 추가 (List cardStack)
  4. target이 되지 않는다면
     • 4-1: (target - 손에 있는 카드 숫자)가 카드 더미에 있는지 확인한다. (코인으로 카드 한 장)
     • 4-1: 4-1에서 카드가 없다면 카드 더미에서 코인 2개를 사용해 target을 만들 수 있는지 확인한다. (코인으로 카드 두 장)

• 손 또는 카드더미에서 카드 두 장을 골라 target을 만드는지 검사하는 로직은 아래와 같이 작성할 수 있다. 주의할 점은 가장 큰 인덱스부터 제거해야 한다. 작은 거 부터 제거하면, 큰 인덱스를 제거할 때 전체 배열의 크기가 작아진 상태이므로 OutOfBound가 발생할 수 있기 때문이다.

  boolean search(List<Integer> list, int target) {
    for (int i = 0; i < list.size(); ++i) {
        for (int j = i + 1; j < list.size(); ++j) {
            if (list.get(i) + list.get(j) == target) {
                list.remove(j);
                list.remove(i);
                return true;
            }
        }
    }
    return false;
  }

• 놓치기 쉬운 테스트 케이스가 존재한다. 현재 로직에선 target을 만들 수 있다면, 계속해서 라운드를 증가시킨다. 특정 케이스에서 최대 가능한 라운드보다 더 커질 수 있다. 따라서 정답을 구할 때 최대 라운드에 도달했다면 더 이상 카드를 탐색하지 않는다.

  while (true) {
  if (ans == n / 3 + 1) break; // 최대 라운드 도달 시 종료
  
  if (isSearched) ++ans; // 찾았다면 다음 라운드로, 못 찾았다면 종료
  else break;
  }

개선

시간복잡도

• $O(n^3)$

코드

import java.util.*;

class Solution {
    public int solution(int coin, int[] cards) {
        int n = cards.length;

        int target = n + 1;
        List<Integer> cardStack = new ArrayList<>();
        List<Integer> hand = new ArrayList<>();
        for (int i = 0; i < n / 3; ++i) hand.add(cards[i]);

        int cursor = n / 3; 
        int ans = 1;
        while (true) {
            if (ans == n / 3 + 1) break;
            if (cursor < n) cardStack.add(cards[cursor++]);
            if (cursor < n) cardStack.add(cards[cursor++]);

            boolean isSearched = search(hand, target);

            if (!isSearched && coin > 0) {
                for (int i = 0; i < hand.size(); i++) {
                    int find = target - hand.get(i);
                    if (cardStack.contains(find)) {
                        cardStack.remove(Integer.valueOf(find));
                        coin--;
                        isSearched = true;
                        break;
                    }
                }
            }  

            if (!isSearched && coin >= 2) {
                isSearched = search(cardStack, target);
                if (isSearched) {
                    coin -= 2;
                }
            }

            if (isSearched) ++ans;
            else break;
        }

        return ans;
    }

    boolean search(List<Integer> list, int target) {
        for (int i = 0; i < list.size(); ++i) {
            for (int j = i + 1; j < list.size(); ++j) {
                if (list.get(i) + list.get(j) == target) {
                    list.remove(j);
                    list.remove(i);
                    return true;
                }
            }
        }
        return false;
    }
}

Programmers Lv3, 주사위 고르기

핵심

• A와 B가 n개의 주사위 중 n/2 주사위를 골라 주사위를 모두 굴린 후 나온 수들을 비교하여 점수가 큰 쪽이 이기는 주사위 게임을 한다. 주사위마다 쓰인 수의 구성이 모두 다를 때 A가 자신이 승리할 확률이 가장 높은 주사위를 가져가는 경우를 구해야 한다.
• 지문을 읽어보면 직관적으로 완전 탐색 문제임을 알 수 있다. A, B가 가져갈 주사위 순서를 구한 뒤 나올 수 있는 점수의 합을 구하여 A가 이길 확률이 가장 높은 주사위 순서를 반환하면 된다.

풀이

1. 주사위 고르는 모든 경우를 구한다. (A, B 각각)
2. 고른 주사위 각각의 순서에 대해 점수 합계를 구한 뒤 A가 이기는 횟수를 구한다. 전체 경우의 수는 같고, 무승부와 패배는 고려할 필요 없다.
3. A가 가장 크게 이길 때 주사위 순서쌍을 오름차순으로 반환한다.

1. 주사위 고르는 모든 경우

• dfs를 이용해 모든 순열을 구할 수 있다. 처음엔 아래와 같이 작성했지만, 시간 초과가 발생했다. 이 부분은 아래에서 설명할 예정인데, 어디에서 개선할 수 있는지 잠깐 고민해 보자.

boolean[] isVisited = new boolean[n];
List<int[]> dices = new ArrayList<>();

void dfs(int depth, int[] cur, int n, boolean[] isVisited, List<int[]> dices) {
    if (depth == n / 2) {
        dices.add(cur.clone());
        return;
    }

    for (int i = 0; i < n; i++) {
        if (!isVisited[i]) {
            isVisited[i] = true;
            cur[depth] = i;
            dfs(depth + 1, cur, n, isVisited, dices);
            isVisited[i] = false;
        }
    }
}

• 이렇게 하면 주사위가 {1,2,3,4} 4개가 있을 때 아래와 같은 순열을 만들 수 있다.

  0 1 
  0 2 
  0 3 
  1 2 
  1 3 
  2 3 

• A주사위가 0번, 1번 주사위를 선택했을 때 B는 2번 3번 주사위를 선택해야 하므로 전체 {0, 1, 2, 3}을 Set에 담고, A가 가지고 있지 않는 번호는 B에 추가하여 A주사위와 B 주사위를 만들어주었다.

  for (var d : dices) {
    Set<Integer> s = new HashSet<>();
    for (int i = 0; i < d.length; ++i) {
        s.add(d[i]);
    }
    List<int[]> aDice = new ArrayList<>();
    List<int[]> bDice = new ArrayList<>();
  
    for (int i = 0; i < n; ++i) {
        if (s.contains(i))
            aDice.add(dice[i]);
        else
            bDice.add(dice[i]);
    }
  }

  2. 고른 주사위에 대해 A가 이기는 횟수 구하기

• 위의 과정을 통해 A 주사위와 B주사위에 각각 아래와 같은 값이 들어가 있다.

  A (0, 1번 선택)
  1 2 3 4 5 6 
  3 3 3 3 4 4 
  

B (2, 3번 선택) 1 3 3 4 4 4 1 1 4 4 5 5

- **A, B가 만들 수 있는 점수 합계, 개수를 구한 뒤 A가 B보다 큰 점수 합계에 대해 A 개수 * B 개수를 하면 A가 이기는 횟수를 효율적으로 구할 수 있다**. 이를 위해 해쉬맵을 사용한다.
```java
int calculateWin(List<int[]> aDice,  List<int[]> bDice) {

    Map<Integer, Integer> aSum = new HashMap<>();
    Map<Integer, Integer> bSum = new HashMap<>();

    dfs2(0, 0, aDice, aSum);
    dfs2(0, 0, bDice, bSum);

    int aWinCnt = 0;
    for (var a : aSum.entrySet()) {
        for (var b : bSum.entrySet()) {
            if (a.getKey() > b.getKey()) {
                aWinCnt += (a.getValue() * b.getValue());
            }
        }
    }

    return aWinCnt;
}

• A, B가 n개의 주사위 중 n/2개를 골라 만들 수 있는 점수 합계와 개수를 구하기 위해 DFS를 사용할 수 있다.

  void dfs2(int depth, int sum, List<int[]> dice, Map<Integer, Integer> sums) {
    if (depth == dice.size()) {
        sums.put(sum, sums.getOrDefault(sum, 0) + 1);
        return ;
    }
    for (var score : dice.get(depth)) {
        dfs2(depth + 1, sum + score, dice, sums);
    }
  }

3. A가 이전보다 더 크게 이긴 경우 갱신하기

• A가 이전보다 더 크게 이겼을 때 주사위 순열까지 함께 갱신한다. 주사위 순열은 오름차순으로 만들어졌으므로 별도로 정렬할 필요는 없다.

  if (aWinCnt > max) {
    max = aWinCnt;
    for (int i = 0; i < d.length; ++i) {
        ans[i] = d[i] + 1;
    }
  }

개선

• 위 로직으로 코드를 실행하면 아래와 같이 특정 케이스에서 시간 초과가 난다.

  void dfs(int depth, int[] cur, int n, boolean[] isVisited, List<int[]> dices) {
    if (depth == n / 2) {
        dices.add(cur.clone());
        return;
    }
  
    for (int i = 0; i < n; i++) {
        if (!isVisited[i]) {
            isVisited[i] = true;
            cur[depth] = i;
            dfs(depth + 1, cur, n, isVisited, dices);
            isVisited[i] = false;
        }
    }
  }

• 모든 순열을 탐색할 때 매번 0부터 탐색한다. isVisited 배열로 인해 중복 순열이 만들어지지 않지만, 매번 처음부터 탐색하므로 탐색 범위가 넓다.

• 순열을 만들 때 이전 선택된 것에서 +1 증가한 수를 만들면 되므로 전체를 볼 필요 없이 이전 선택된 것 + 1로 탐색 범위를 좁힐 수 있다. 탐색 범위가 $n!$에서 $2^n$으로 줄어든다고 볼 수 있다.

  int st = depth == 0 ? 0 : cur[depth - 1] + 1; 
  for (int i = 0; i < n; i++) {
  }

시간복잡도

• $O(2^n * 6^n)$

코드

import java.util.*;

class Solution {
    public int[] solution(int[][] dice) {
        int n = dice.length;

        boolean[] isVisited = new boolean[n];
        List<int[]> dices = new ArrayList<>();

        dfs(0, new int[n / 2], n, isVisited, dices);

        int[] ans = new int[n / 2];
        int max = -1;
        for (var d : dices) {
            Set<Integer> s = new HashSet<>();
            for (int i = 0; i < d.length; ++i) {
                s.add(d[i]);
            }
            List<int[]> aDice = new ArrayList<>();
            List<int[]> bDice = new ArrayList<>();

            for (int i = 0; i < n; ++i) {
                if (s.contains(i))
                    aDice.add(dice[i]);
                else
                    bDice.add(dice[i]);
            }

            int aWinCnt = calculateWin(aDice, bDice);

            if (aWinCnt > max) {
                max = aWinCnt;
                for (int i = 0; i < d.length; ++i) {
                    ans[i] = d[i] + 1;
                }
            }
        }

        return ans;
     }

     void dfs(int depth, int[] cur, int n, boolean[] isVisited, List<int[]> dices) {
        if (depth == n / 2) {
            dices.add(cur.clone());
            return;
        }

        int st = depth == 0 ? 0 : cur[depth - 1] + 1; 
        for (int i = st; i < n; i++) {
            if (!isVisited[i]) {
                isVisited[i] = true;
                cur[depth] = i;
                dfs(depth + 1, cur, n, isVisited, dices);
                isVisited[i] = false;
            }
        }
    }

    int calculateWin(List<int[]> aDice,  List<int[]> bDice) {

        Map<Integer, Integer> aSum = new HashMap<>();
        Map<Integer, Integer> bSum = new HashMap<>();

        dfs2(0, 0, aDice, aSum);
        dfs2(0, 0, bDice, bSum);

        int aWinCnt = 0;
        for (var a : aSum.entrySet()) {
            for (var b : bSum.entrySet()) {
                if (a.getKey() > b.getKey()) {
                    aWinCnt += (a.getValue() * b.getValue());
                }
            }
        }

        return aWinCnt;
    }

    void dfs2(int depth, int sum, List<int[]> dice, Map<Integer, Integer> sums) {
        if (depth == dice.size()) {
            sums.put(sum, sums.getOrDefault(sum, 0) + 1);
            return ;
        }
        for (var score : dice.get(depth)) {
            dfs2(depth + 1, sum + score, dice, sums);
        }
    }
}

엘리스 클라우드 트랙 과정

• 엘리스 백엔드 클라우드 트랙은 6개월간 진행되며, 크게 3개의 프로젝트를 진행한다. (개인 프로젝트, 팀 프로젝트, 마지막 팀 프로젝트로 구성) Java 기초 과정에서부터 시작해서 Spring framework까지 자바 백엔드 관련 기술을 전반적으로 다룬다. 첫 스터디에선 HTML, CSS, JavaScript으로 실제 사이트를 클론 코딩하며 기초적인 프론트 기술까지도 배우니 웹 개발 전반에 대해서 빠르게 배울 수 있다는 장점이 있다.

좋았던 점

1. 자동화된 채점 사이트

• 개념을 정리한 짧은 영상과 자료를 보며 핵심을 빠르게 파악할 수 있다. 프로그래머스 알고리즘 문제를 풀듯이 코드를 치며 복습을 할 수 있는 게 가장 큰 강점이다.
• 앨리스 트랙을 진행하기 전에는 DB가 많이 어렵고 생소했는데, 문제를 반복해서 풀면서 DB에 익숙해지는 좋은 계기가 되었다. 이때 한 주간 실행한 코드 횟수가 천 번이 넘었던 걸로 기억한다.

2. 양질의 코드

• 1번에서 문제를 풀 때, 제공되는 베이스 코드가 좋다. 나의 경우 문제를 풀고 나서, 전체 코드를 개인 깃헙에 따라 쓰며 구조를 파악하는 연습을 했었는데 전체 코드가 마치 작은 프로젝트와도 같은 규모라 많은 도움이 된다.
• 이와 같은 연습을 하면 첫 프로젝트에서도 간단한 게시판 CRUD를 넘는, 조금 더 어렵거나 디테일한 기능을 구현하고 싶은 도전 정신이 생긴다. 첫 번째 프로젝트에서 공간 예약 서비스를 2주간 개발을 했는데, 프로젝트를 수료하고 이를 다시 개발하게 되었다. 화면단을 생각하지 않는 아주 이기적인 API를 볼 수 있었고, 요상한 DB 설계를 보면서 '엘리스 트랙에서 많이 성장했구나'고 느낄 수 있었다.

3. 3번의 프로젝트, 지속적인 피드백

• 프로젝트를 하면서 정말 많은 걸 배운다. 프로젝트를 진행할 때 담당 코치님이 배정되는 데, 구현에 어려움이 있거나 개선점을 지속적으로 피드백해 주신다. 고민해서 코드를 작성하고, 부족한 점을 피드백 받고, 다음 미팅 때까지 수정하는 식으로 진행되는데 엘리스 프로젝트에 몰입할 수 있는 환경이라면 정말 시간 가는 줄 모르고 개발하는 재미를 느끼게 된다.
• 코치님들과 미팅 시간 외에도 궁금한 점이 있으면, 디스코드로 물어보고 답변을 받을 수 있다.

• 간단한 프로젝트여도 고민한 경험이 쌓이면서 개발 프로세스를 익히게 된다. 어느 순간 화면 설계(와이어 프레임)부터 하며 ERD를 고민하는 자신을 보게 된다.
• 열심히 하면 상도 받는다 ㅎㅎㅎ

아쉬웠던 점

1. 오프라인 한계

• 엘리스 랩이 성수, 강남, 부산에 위치하고, 여기에 나가서 학습하면 같이 학습하는 좋은 동료를 얻을 수 있었는데, 사정상 나가지 못했다. 앨리스 트랙에서 여러 스터디, 프로젝트를 하며 많은 사람을 만났지만, 인연이 이어졌던 건 마지막 프로젝트 사람 중 일부인 게 아쉽다.

• 그래도 마지막 프로젝트 팀원분들과는 좋은 인연으로 남아 현재까지 TIL 스터디를 진행하고 있다.

• 오프라인으로 사람들을 만나고 같이 스터디, 프로젝트 하는 게 학습 효과도 좋고 결과도 더 좋아 보였다! 엘리스 트랙을 앞둔 사람들이라면 꼭 오프라인 참석해서 좋은 사람들과 인연 쌓고, 같이 공부하는 걸 추천한다.

2. 무작위 팀원

• 마지막 프로젝트에서는 부분적으로 팀을 꾸릴 수 있었는데, 이게 프로젝트 초반부터 되었으면 더 좋았을 듯싶다. 프로젝트가 새로 진행될 때마다 역할도 새로 바뀌는 경우가 많았기 때문에 고도화 작업을 하기 어려웠다. 예를 들어, 첫 번째 프로젝트에서 Spring Security 인증 인가를 맡아서 Oauth2.0을 달고, Access Token과 Refresh Token 로직을 추가했다면, 두 번째 프로젝트에서 다른 사람이 첫 번째 사람과 비슷하게 기능을 추가하는 것에서 멈춘다는 것이다. 새로운 IP에서 로그인할 경우 2차 인증이나 캡챠 서비스나 로그인 실패 시 계정 잠금 또는 totp 인증 ... 새로운 기능을 적용할 시간이 부족하기 때문에 다들 비슷한 기능을 가진 프로젝트로 끝이 난다. 물론, 반대로 여러 도메인을 만져볼 수 있다는 장점이 있다.

소정의 비용을 받고 작성된 글입니다. https://elice.training/

Programmers Lv3, 카드 짝 맞추기

핵심

• 게임 진행 중 카드의 짝을 맞춰 몇 장 제거된 상태에서 카드 앞면의 그림을 알고 있다면, 남은 카드를 모두 제거하는 데 필요한 키 조작 횟수의 최솟값을 구해야 한다. 같은 그림을 제거하지 않으면, 다시 원상 복귀되므로 같은 그림을 먼저 제거하는 방식으로 구해야 최솟값을 구할 수 있다. 1~3번 그림 중 어떤 그림을 먼저 제거하느냐에 따라 움직여야 하는 횟수가 달라지며, 1번을 먼저 지운다고 했을 때 아래 그림처럼 (0,0)을 먼저 지우는지 또는 (3, 2)를 먼저 지우는지에 따라 움직이는 횟수가 다르다.

• 즉, 최소 횟수 움직임을 구하기 위해 DFS로 가능한 순열을 만든다. 아래 그림에선 아래와 같은 전체 순열을 만들어야 한다. 가능한 카드 쌍은 최대 3개이며(3!), 카드 쌍마다 순서를 변경할 수 있으므로($2^3$) 총 48가지가 나온다.

  0 0, 3 2, 1 0, 2 3, 0 3, 3 0
  0 0, 3 2, 1 0, 2 3, 3 0, 0 3
  0 0, 3 2, 2 3, 1 0, 0 3, 3 0
  0 0, 3 2, 2 3, 1 0, 3 0, 0 3
  0 0, 3 2, 0 3, 3 0, 1 0, 2 3
  0 0, 3 2, 0 3, 3 0, 2 3, 1 0
  0 0, 3 2, 3 0, 0 3, 1 0, 2 3
  0 0, 3 2, 3 0, 0 3, 2 3, 1 0 
  

...

- 위의 순열은 아래와 같은 코드로 만들 수 있다. 
```java
void dfs(List<int[]> cur, List<int[]> cards, boolean[] isVisited, List<List<int[]>> perms) {
    if (cur.size() == cards.size()) {
        perms.add(new ArrayList<>(cur));
        return;
    }

    for (int i = 0; i < cards.size(); i += 2) {
        if (!isVisited[i]) {
            isVisited[i] = true;
            cur.add(cards.get(i));
            cur.add(cards.get(i + 1));
            dfs(cur, cards, isVisited, perms);
            cur.remove(cur.size() - 1);
            cur.remove(cur.size() - 1);

            cur.add(cards.get(i + 1));
            cur.add(cards.get(i));
            dfs(cur, cards, isVisited, perms);
            cur.remove(cur.size() - 1);
            cur.remove(cur.size() - 1);

            isVisited[i] = false;
        }
    }
}

• 물론 DFS로 순열을 만들기 전에 같은 그림에 해당하는 좌표를 모으는 작업을 해주어야 한다.

  List<int[]> cards = new ArrayList<>();
  Map<Integer, List<int[]>> cardPos = new HashMap<>();
  for (int i = 0; i < 4; i++) {
    for (int j = 0; j < 4; j++) {
        if (board[i][j] > 0) {
            cardPos.putIfAbsent(board[i][j], new ArrayList<>());
            cardPos.get(board[i][j]).add(new int[]{i, j});
        }
    }
  }
  

for (var pos : cardPos.values()) { cards.add(pos.get(0)); cards.add(pos.get(1)); }

- 가능한 경로를 모두 계산했다면, 각 경로에 대해 최소 움직임 횟수를 구해야 한다. 이는 BFS로 구할 수 있으며, 기존 BFS와 다르게 상하좌우 한 칸씩 이동하는 로직에 더해 **한 번에 끝으로 이동하는 로직을 추가**해야 한다.
```java
int getMinimumMove(int[][] board, int r, int c, List<int[]> perm) {
    int[][] tmp = new int[4][4];
    for (int i = 0; i < 4; i++) {
        System.arraycopy(board[i], 0, tmp[i], 0, 4);
    }

    int move = 0;
    int cr = r;
    int cc = c;
    for (int i = 0; i < perm.size(); i += 2) { // 각 경로에 도착하기 까지
        int[] firstCard = perm.get(i);
        int[] secondCard = perm.get(i + 1);

        int firstMove = bfs(tmp, cr, cc, firstCard[0], firstCard[1]); // 현재 경로에서 첫 번째 카드 위치까지
        int secondMove = bfs(tmp, firstCard[0], firstCard[1], secondCard[0], secondCard[1]); // 첫 번째 카드 위치에서 두 번째 카드 위치까지

        move += firstMove + secondMove + 2; // 엔터 두번 추가
        tmp[firstCard[0]][firstCard[1]] = 0;
        tmp[secondCard[0]][secondCard[1]] = 0;

        cr = secondCard[0];
        cc = secondCard[1];
    }

    return move;
}

int bfs(int[][] board, int sy, int sx, int ey, int ex) {
    if (sy == ey && sx == ex) {
        return 0;
    }

    Queue<int[]> q = new LinkedList<>();
    boolean[][] isVisited = new boolean[4][4];
    q.add(new int[]{sy, sx, 0});
    isVisited[sy][sx] = true;

    while (!q.isEmpty()) {
        int[] cur = q.poll();
        int y = cur[0];
        int x = cur[1];
        int cnt = cur[2];

        for (int i = 0; i < 4; i++) {
            int ny = y + dy[i];
            int nx = x + dx[i];
            if (!isOOB(ny, nx) && !isVisited[ny][nx]) {
                if (ny == ey && nx == ex) {
                    return cnt + 1;
                }
                isVisited[ny][nx] = true;
                q.add(new int[]{ny, nx, cnt + 1});                 
            }

            ny = y;
            nx = x;            
            // 처음 위치에서 한 방향으로 끝에 도달하는 경우
            while (!isOOB(ny + dy[i], nx + dx[i])) {
                ny += dy[i];
                nx += dx[i];
                if (board[ny][nx] != 0) {
                        break;
                }
            }
            if (!isVisited[ny][nx]) {
                if (ny == ey && nx == ex) {
                    return cnt + 1;
                }
                isVisited[ny][nx] = true;
                q.add(new int[]{ny, nx, cnt + 1});
            }
        }
    }

    return Integer.MAX_VALUE;
}

개선

시간복잡도

• $O(V + E)$ (나머지 상수시간)

코드

import java.util.*;

class Solution {
    int[] dy = {-1, 0, 1, 0};
    int[] dx = {0, 1, 0, -1};

    public int solution(int[][] board, int r, int c) {
        List<int[]> cards = new ArrayList<>();
        Map<Integer, List<int[]>> cardPos = new HashMap<>();

        for (int i = 0; i < 4; i++) {
            for (int j = 0; j < 4; j++) {
                if (board[i][j] > 0) {
                    cardPos.putIfAbsent(board[i][j], new ArrayList<>());
                    cardPos.get(board[i][j]).add(new int[]{i, j});
                }
            }
        }

        for (var pos : cardPos.values()) {
            cards.add(pos.get(0));
            cards.add(pos.get(1));
        }

        List<List<int[]>> perms = new ArrayList<>();
        boolean[] visited = new boolean[cards.size()];
        dfs(new ArrayList<>(), cards, visited, perms);

        int ans = Integer.MAX_VALUE;

        for (var perm : perms) {
            ans = Math.min(ans, getMinimumMove(board, r, c, perm));
        }

        return ans;
    }

   void dfs(List<int[]> cur, List<int[]> cards, boolean[] isVisited, List<List<int[]>> perms) {
        if (cur.size() == cards.size()) {
            perms.add(new ArrayList<>(cur));
            return;
        }

        for (int i = 0; i < cards.size(); i += 2) {
            if (!isVisited[i]) {
                isVisited[i] = true;
                cur.add(cards.get(i));
                cur.add(cards.get(i + 1));
                dfs(cur, cards, isVisited, perms);
                cur.remove(cur.size() - 1);
                cur.remove(cur.size() - 1);

                cur.add(cards.get(i + 1));
                cur.add(cards.get(i));
                dfs(cur, cards, isVisited, perms);
                cur.remove(cur.size() - 1);
                cur.remove(cur.size() - 1);

                isVisited[i] = false;
            }
        }
    }

    int getMinimumMove(int[][] board, int r, int c, List<int[]> perm) {
        int[][] tmp = new int[4][4];
        for (int i = 0; i < 4; i++) {
            System.arraycopy(board[i], 0, tmp[i], 0, 4);
        }

        int move = 0;
        int cr = r;
        int cc = c;

        for (int i = 0; i < perm.size(); i += 2) {
            int[] firstCard = perm.get(i);
            int[] secondCard = perm.get(i + 1);

            int firstMove = bfs(tmp, cr, cc, firstCard[0], firstCard[1]);
            int secondMove = bfs(tmp, firstCard[0], firstCard[1], secondCard[0], secondCard[1]);

            move += firstMove + secondMove + 2; 
            tmp[firstCard[0]][firstCard[1]] = 0;
            tmp[secondCard[0]][secondCard[1]] = 0;

            cr = secondCard[0];
            cc = secondCard[1];
        }

        return move;
    }

    int bfs(int[][] board, int sy, int sx, int ey, int ex) {
        if (sy == ey && sx == ex) {
            return 0;
        }

        Queue<int[]> q = new LinkedList<>();
        boolean[][] isVisited = new boolean[4][4];
        q.add(new int[]{sy, sx, 0});
        isVisited[sy][sx] = true;

        while (!q.isEmpty()) {
            int[] cur = q.poll();
            int y = cur[0];
            int x = cur[1];
            int cnt = cur[2];

            for (int i = 0; i < 4; i++) {
                int ny = y + dy[i];
                int nx = x + dx[i];
                if (!isOOB(ny, nx) && !isVisited[ny][nx]) {
                    if (ny == ey && nx == ex) {
                        return cnt + 1;
                    }
                    isVisited[ny][nx] = true;
                    q.add(new int[]{ny, nx, cnt + 1});                 
                }

                ny = y;
                nx = x;
                while (!isOOB(ny + dy[i], nx + dx[i])) {
                    ny += dy[i];
                    nx += dx[i];
                    if (board[ny][nx] != 0) {
                        break;
                    }
                }
                if (!isVisited[ny][nx]) {
                    if (ny == ey && nx == ex) {
                        return cnt + 1;
                    }
                    isVisited[ny][nx] = true;
                    q.add(new int[]{ny, nx, cnt + 1});
                }
            }
        }

        return Integer.MAX_VALUE;
    }

    boolean isOOB(int r, int c) {
        return r < 0 || r >= 4 || c < 0 || c >= 4;
    }
}

Programmers Lv3, 매칭 점수

핵심

• html 문자열 파싱 문제이다. 웹 페이지들의 매칭 점수를 구한 뒤 가장 큰 매칭 점수를 가진 웹 페이지 인덱스를 출력한다. 매칭 점수는 기본 점수와 링크 점수의 합으로 계산하며, 기본 점수는 해당 웹 페이지의 텍스트 중 검색어가 등장하는 횟수이다. 링크 점수는 해당 웹페이지로 링크가 걸린 다른 웹페이지의 기본 점수 / 외부 링크 수의 총합으로 구할 수 있다.

• 기본 점수는 메시지 body 부분을 잘라낸 후 body에서 word가 몇 개 있는지 세어 확인할 수 있다.

• 링크 점수는 각 페이지를 순회하며 해당 페이지에서 외부로 링크된 모든 페이지를 확인하면서, 외부 링크된 각 페이지에 현재 페이지의 기본 점수를 외부 링크 수로 나눈 값을 더하여 구할 수 있다.

  for (int i = 0; i < pages.length; ++i) {
    String page = pages[i];
    String baseUrl = getBaseUrl(page);
    for (var link : externalLinks[i]) {
        if (ps.containsKey(link)) {
            int linkIndex = ps.get(link);
            linkScores[linkIndex] += (double) basicScores[i] / externalLinks[i].size();
        }
    }
  }

• 외부 링크된 각 페이지에 현재 페이지의 링크 점수를 계산하기 위해 먼저 해당 웹 페이지 url과 인덱스 번호를 저장한다. 특정 인덱스에 대한 외부 링크 번호를 저장하는 것도 동일한 방식으로 파싱할 수 있다.

  String getBaseUrl(String page) {
    String find = "<meta property=\"og:url\" content=\"";
    int st = page.indexOf(find) + find.length();
    int en = page.indexOf("\"", st);
    return page.substring(st, en);
  }
  

List getExternalLinks(String page) { List links = new ArrayList<>(); int pos = page.indexOf("<a href=""); while (pos != -1) { int start = pos + "<a href="".length(); int end = page.indexOf(""", start); links.add(page.substring(start, end)); pos = page.indexOf("<a href="", end); } return links; }

- **기본 점수를 구하기 위해 각 페이지 Body부분에 word가 몇 번 반복되었는지 계산**한다.
```java
int getBasicScore(String word, String page) {
    String lowerPage = page.toLowerCase();
    String body = lowerPage.substring(lowerPage.indexOf("<body>") + 7, lowerPage.indexOf("</body>"));
    int score = 0;
    String[] words = body.split("[^a-zA-Z]");
    for (var w : words) {
            if (w.equals(word)) {
                    score++;
            }
    }
    return score;
}

• 기본 점수와 링크 점수를 합한 뒤 매칭 점수가 더 크다면 인덱스를 갱신하는 방식으로 답을 구한다.

  double[] totalScores = new double[pages.length];
  for (int i = 0; i < pages.length; ++i) {
        totalScores[i] = basicScores[i] + linkScores[i];
  }
  

int ans = 0; for (int i = 1; i < pages.length; ++i) { if (totalScores[i] > totalScores[ans]) { ans = i; } }

return ans;

# 개선

# 시간복잡도
- $O(n * L + n * L + n * k)$ 
- n = 페이지 수, L = 페이지 길이, K = 외부 링크 수
![](https://velog.velcdn.com/images/ji-jjang/post/5964910e-ba78-427e-9889-aaab55e89172/image.png)

# 코드
```java
import java.util.*;

class Solution {

    public int solution(String word, String[] pages) {
        word = word.toLowerCase();
        Map<String, Integer> ps = new HashMap<>(); // url, index
        List<String>[] externalLinks = new ArrayList[pages.length];
        int[] basicScores = new int[pages.length];
        double[] linkScores = new double[pages.length];

        for (int i = 0; i < pages.length; ++i) {
            externalLinks[i] = new ArrayList<>();
            String baseUrl = getBaseUrl(pages[i]);
            ps.put(baseUrl, i);
            basicScores[i] = getBasicScore(word, pages[i]);
            externalLinks[i] = getExternalLinks(pages[i]);
        }

        for (int i = 0; i < pages.length; ++i) {
            String page = pages[i];
            String baseUrl = getBaseUrl(page);
            for (var link : externalLinks[i]) {
                if (ps.containsKey(link)) {
                    int linkIndex = ps.get(link);
                    linkScores[linkIndex] += (double) basicScores[i] / externalLinks[i].size();
                }
            }
        }

        double[] totalScores = new double[pages.length];
        for (int i = 0; i < pages.length; ++i) {
            totalScores[i] = basicScores[i] + linkScores[i];
        }

        int ans = 0;
        for (int i = 1; i < pages.length; ++i) {
            if (totalScores[i] > totalScores[ans]) {
                ans = i;
            }
        }

        return ans;
    }

    String getBaseUrl(String page) {
        String find = "<meta property=\"og:url\" content=\"";
        int st = page.indexOf(find) + find.length();
        int en = page.indexOf("\"", st);
        return page.substring(st, en);
    }

    int getBasicScore(String word, String page) {
        String lowerPage = page.toLowerCase();
        String body = lowerPage.substring(lowerPage.indexOf("<body>") + 7, lowerPage.indexOf("</body>"));
        int score = 0;
        String[] words = body.split("[^a-zA-Z]");
        for (var w : words) {
            if (w.equals(word)) {
                score++;
            }
        }
        return score;
    }

    List<String> getExternalLinks(String page) {
        List<String> links = new ArrayList<>();
        int pos = page.indexOf("<a href=\"");
        while (pos != -1) {
            int start = pos + "<a href=\"".length();
            int end = page.indexOf("\"", start);
            links.add(page.substring(start, end));
            pos = page.indexOf("<a href=\"", end);
        }
        return links;
    }
}