🧠 Redis Pub/Sub의 구조는?

Redis Pub/Sub은 발행자-구독자 패턴을 따르며, 중간에 Redis가 브로커 역할을 수행한다.

메시지는 Redis 서버를 거쳐 해당 채널을 구독 중인 모든 클라이언트에게 fan-out 된다.

📌 이 구조는 트위터의 일반 유저 알림 시스템에 적용된적이 있다고 한다 (인플루언서는 별도 처리)

🔄 Channel-based Fan-out 시스템

Redis는 채널 기반 fan-out 시스템으로 동작한다.

예를 들어:

하나의 채널에 대해 여러 클라이언트가 구독하고 있으면, 메시지는 모두에게 전파된다. → 이것이 바로 1:N 구조다.

🧩 내부 자료구조

Redis는 Pub/Sub을 다음과 같이 관리한다:

dict<string, list<client>> pubsub_channels;

pubsub_channels = { "chatting-room-1": [clientA, clientB], "chatting-room-2": [clientC] }

ClientD가 "chatting-room-2"를 구독하면?

pubsub_channels = { "chatting-room-1": [clientA, clientB], "chatting-room-2": [clientC, clientD] }

❗ 중복 구독은 어떻게 될까?

ㄱㅊ Redis는 Set처럼 동작해서, 같은 채널을 여러 번 구독해도 한 번만 등록된다.

코드로 보면 다음과 같다:

int pubsubSubscribeChannel(client *c, robj *channel) {
    if (dictAdd(c->pubsub_channels, channel, NULL) == DICT_OK) {
        ...
        listAddNodeTail(clients, c);
    }
}

⏱️ 주요 연산과 시간 복잡도

그럼 다시 자료구조로 돌아와서 시간복잡도를 알아보자

redis는 싱글 스레드(워커) 기반이기 때문에 시간복잡도가 중요함

채널 추가 제거 탐색은 HashMap이기 때문에 시간복잡도가 좋음 (리해싱이나 해시 충돌은 나중에)

구독자 추가, 제거도 리스트기 때문에 간편함 하지만 구독자 탐색이 시간복잡도가 N임

그렇다면?? 하나의 채널에 구독자가 많으면 많을수록 안좋다,

🧮 dict의 해시 함수는?

Redis의 해시 테이블은 성능에 민감하기 때문에 다음과 같은 조건을 만족해야 한다:

그래서 Redis는 SHA256, MD5 대신 murmurhash2 기반의 dictGenHashFunction을 사용한다.

murmurhash2 특징: murmurhash2는 Google 출신 개발자인 Austin Appleby가 만든 비암호화(non-cryptographic) 해시 함수

• 매우 빠름: CPU 캐시 친화적인 구조, 단순한 연산 (곱셈, XOR, shift) 기반
• 낮은 충돌률: key가 유사하더라도 해시 결과가 고르게 분산됨

Redis 해시 테이블의 충돌 처리 & 리해싱 구조 (with Java 비교)

Redis의 내부 해시 테이블은 어떻게 충돌을 처리할까?
Java의 HashMap과 비교하며 Redis의 설계를 들여다보자.

☠️ 해시 충돌이란?

해시 충돌은 서로 다른 키가 동일한 해시 슬롯으로 매핑되는 상황이다.
Java(8이상) 와 Redis는 이를 어떻게 처리할까?

간단한 해시인덱스 - 연결리스트 기반의 해시 체이닝을 기반 그렇다면? 엔트리 개수가 많아질수록 한 해시인덱스의의 연결리스트안의 엔트리가 많이 추가되므로 해시 충돌 확률도 올라감 어떻게 해쉬 충돌을 해결했을까? → 해시 리사이징과 점진적 리해싱으로 해결

🔁 Redis의 해시 체이닝 예시

hash % 4 로 간단하게 슬롯을 나눈다고 가정

❓ 왜 banana가 apple 앞에 왔을까?

Redis는 최근 삽입된 데이터를 리스트의 앞쪽에 배치한다. 최근 데이터가 자주 조회될 가능성이 높기 때문이다.

dictEntry *dictAddRaw(dict *d, void *key, dictEntry **existing)
{
    long index;
    dictEntry *entry;
    int htidx;

    if (dictIsRehashing(d)) _dictRehashStep(d);

    /* Get the index of the new element, or -1 if
     * the element already exists. */
    if ((index = _dictKeyIndex(d, key, dictHashKey(d,key), existing)) == -1)
        return NULL;

    /* Allocate the memory and store the new entry.
     * Insert the element in top, with the assumption that in a database
     * system it is more likely that recently added entries are accessed
     * more frequently. */
    htidx = dictIsRehashing(d) ? 1 : 0;
    size_t metasize = dictMetadataSize(d);
    entry = zmalloc(sizeof(*entry) + metasize);
    if (metasize > 0) {
        memset(dictMetadata(entry), 0, metasize);
    }
    entry->next = d->ht_table[htidx][index];
    d->ht_table[htidx][index] = entry;
    d->ht_used[htidx]++;

    /* Set the hash entry fields. */
    dictSetKey(d, entry, key);
    return entry;
}

여기서

entry->next = d->ht_table[htidx][index];
d->ht_table[htidx][index] = entry;

이부분을 보면 맨 앞으로 삽입한것을 볼 수 있다. 최근에 추가한 놈을 찾을 확률이 높기 때문에 앞에다 배치 이런식으로 redis는 해시 충돌을 링크드 리스트 체이닝 방식으로 해결한다.

근데 자바는 충돌 횟수가 8개 즉 동일 슬롯에 8개 이상 충돌 시 링크드 리스트를 red black tree로 변경한다. 왜? 조회성능을 올리려고 O(N) → O(log N)

하지만 redis는 충돌이 많아져도 트리 변환을 안한다,

왜와이?

• Redis는 고성능 in-memory 시스템 → 단순한 자료구조로 캐시 최적화
  • 트리화하면 메모리 오버헤드(연결리스트는 next 포인터만 있으면 되는데 트리는 왼쪽자식 오른쪽 자식 부모 포인터)가 큼
  • redis는 메모리 사용량이 성능과 비용에 직결되기 때문에 구조가 단순하면서 메모리 오버헤드 최소화하면서 cpu 캐시 친화성(연결리스트는 지역성이 좋음)
  • 메모리 오버헤드는 GC에도 병목이 생김 redis는 싱글 스레드라서 GC에 치명적 (stop the world)
• 해시 분산이 좋고, 리사이징이 빨라서 충돌이 오래 지속되지 않음
  • Redis는 MurmurHash2 같은 고품질 해시 사용
  • load factor 1만 넘어도 바로 리사이징 & rehash
  • 충돌된 슬롯에 key가 오래 남아있을 일이 별로 없음
• 점진적 리해싱으로 충돌은 빠르게 사라짐
  • Java는 리사이징이 한 번에 일어나서 큰 충돌이 문제지만,
  • Redis는 요청마다 조금씩 rehash하므로 성능 영향 적음
• Redis 철학
  • 트리 변환 같은 복잡한 구조 도입보다는 속도와 단순성에 집중

이제 4번 순서로 가보자 여기서 Load Factor = used / size 이다

• used: 테이블에 저장된 key 수 (모든 슬롯에 있는 dictEntry 개수 총합) (예: "apple", "banana")
• size: 테이블의 슬롯 수 (배열 길이) 여기서는 4라고 치자 사실 말이안됨, 해시맵을 만들때 초기 슬롯수를 4로 하는게 말이 안되지만 이해를 위해…

4번째 grape를 넣는 순간 load factor가 1.0이 되면서 리사이징 조건을 충족하면서 리해싱을 시작함

struct dict {
    dictType *type;

    dictEntry **ht_table[2];
    unsigned long ht_used[2];

    long rehashidx; /* rehashing not in progress if rehashidx == -1 */

    /* Keep small vars at end for optimal (minimal) struct padding */
    int16_t pauserehash; /* If >0 rehashing is paused (<0 indicates coding error) */
    signed char ht_size_exp[2]; /* exponent of size. (size = 1<<exp) */
};

처음에 dict를 만들때부터 해시테이블은 두개 만들어놨음

그래서 두번쨰 해시테이블에 조금씩 점직적으로 요청 처리시마다 조금씩 옮김 새로운 테이블은 사이즈를 늘려서 만듬

    long rehashidx; /* rehashing not in progress if rehashidx == -1 */

이부분이 리해싱하고 잇는 인덱스라고 보면 됨

이제 리해싱 시작: 점진적 이동

새 테이블 생성:

• ht[1].size = 8
• rehashidx = 0

int dictRehash(dict *d, int n) {
    int empty_visits = n*10; /* Max number of empty buckets to visit. */
    if (dict_can_resize == DICT_RESIZE_FORBID || !dictIsRehashing(d)) return 0;
    if (dict_can_resize == DICT_RESIZE_AVOID &&
        (DICTHT_SIZE(d->ht_size_exp[1]) / DICTHT_SIZE(d->ht_size_exp[0]) < dict_force_resize_ratio))
    {
        return 0;
    }

    while(n-- && d->ht_used[0] != 0) { // -> n다쓸때까지, 해시테이블[0]에 아무것도 없을 때 까지
        dictEntry *de, *nextde;

        /* Note that rehashidx can't overflow as we are sure there are more
         * elements because ht[0].used != 0 */
        assert(DICTHT_SIZE(d->ht_size_exp[0]) > (unsigned long)d->rehashidx);
        while(d->ht_table[0][d->rehashidx] == NULL) {
            d->rehashidx++;
            if (--empty_visits == 0) return 1;
        }
        de = d->ht_table[0][d->rehashidx];
        /* Move all the keys in this bucket from the old to the new hash HT */
        while(de) {
            uint64_t h;

            nextde = de->next;
            /* Get the index in the new hash table */
            h = dictHashKey(d, de->key) & DICTHT_SIZE_MASK(d->ht_size_exp[1]);
            de->next = d->ht_table[1][h];
            d->ht_table[1][h] = de;
            d->ht_used[0]--;
            d->ht_used[1]++;
            de = nextde;
        }
        d->ht_table[0][d->rehashidx] = NULL;
        d->rehashidx++;
    }

    /* Check if we already rehashed the whole table... */
    if (d->ht_used[0] == 0) {
        zfree(d->ht_table[0]);
        /* Copy the new ht onto the old one */
        d->ht_table[0] = d->ht_table[1];
        d->ht_used[0] = d->ht_used[1];
        d->ht_size_exp[0] = d->ht_size_exp[1];
        _dictReset(d, 1);
        d->rehashidx = -1;
        return 0;
    }

    /* More to rehash... */
    return 1;
}

            de->next = d->ht_table[1][h];
            d->ht_table[1][h] = de;
            d->ht_used[0]--;
            d->ht_used[1]++;

• 예: rehashidx = 0
  • ht[0].table[0] → banana
  • banana의 해시값 % 8 → 예: 4
  • ht[1].table[4] = banana
  • ht[0].table[0] = null
  • rehashidx++

그럼 다음 요청에선 apple 리해싱함

int _dictExpand(dict *d, unsigned long size, int* malloc_failed)
{
    if (malloc_failed) *malloc_failed = 0;

    /* the size is invalid if it is smaller than the number of
     * elements already inside the hash table */
    if (dictIsRehashing(d) || d->ht_used[0] > size)
        return DICT_ERR;

    /* the new hash table */
    dictEntry **new_ht_table;
    unsigned long new_ht_used;
    signed char new_ht_size_exp = _dictNextExp(size);

    /* Detect overflows */
    size_t newsize = 1ul<<new_ht_size_exp;
    if (newsize < size || newsize * sizeof(dictEntry*) < newsize)
        return DICT_ERR;

    /* Rehashing to the same table size is not useful. */
    if (new_ht_size_exp == d->ht_size_exp[0]) return DICT_ERR;

    /* Allocate the new hash table and initialize all pointers to NULL */
    if (malloc_failed) {
        new_ht_table = ztrycalloc(newsize*sizeof(dictEntry*));
        *malloc_failed = new_ht_table == NULL;
        if (*malloc_failed)
            return DICT_ERR;
    } else
        new_ht_table = zcalloc(newsize*sizeof(dictEntry*));

    new_ht_used = 0;

    /* Is this the first initialization? If so it's not really a rehashing
     * we just set the first hash table so that it can accept keys. */
    if (d->ht_table[0] == NULL) {
        d->ht_size_exp[0] = new_ht_size_exp;
        d->ht_used[0] = new_ht_used;
        d->ht_table[0] = new_ht_table;
        return DICT_OK;
    }

    /* Prepare a second hash table for incremental rehashing */
    d->ht_size_exp[1] = new_ht_size_exp;
    d->ht_used[1] = new_ht_used;
    d->ht_table[1] = new_ht_table;
    d->rehashidx = 0;
    return DICT_OK;
}

리해싱은 슬롯 단위로, 순차적(0 → 1 → 2 → …)으로 진행됨 (다음 요청에는 1번 슬롯이 옮겨지겠지?)

    d->ht_size_exp[1] = new_ht_size_exp;
    d->ht_used[1] = new_ht_used;
    d->ht_table[1] = new_ht_table;
    d->rehashidx = 0; -> 0번 슬롯부터 시작
    return DICT_OK;

삽입은 항상 ht[1]에만 들어감

조회는 ht[0]과 ht[1] 둘 다 탐색함

    /* Check if we already rehashed the whole table... */
    if (d->ht_used[0] == 0) {
        zfree(d->ht_table[0]);
        /* Copy the new ht onto the old one */
        d->ht_table[0] = d->ht_table[1];
        d->ht_used[0] = d->ht_used[1];
        d->ht_size_exp[0] = d->ht_size_exp[1];
        _dictReset(d, 1);
        d->rehashidx = -1;
        return 0;
    }

rehashidx == ht[0].size (예: 4) 되면:

• ht[1]을 ht[0]으로 승격
• ht[1] 제거
• rehashidx = -1 복귀

그럼 그 사이에 추가되는 값 추가는 어떻게 될까?

dictEntry *dictAddRaw(dict *d, void *key, dictEntry **existing)
{
    long index;
    dictEntry *entry;
    int htidx;

    if (dictIsRehashing(d)) _dictRehashStep(d);

    /* Get the index of the new element, or -1 if
     * the element already exists. */
    if ((index = _dictKeyIndex(d, key, dictHashKey(d,key), existing)) == -1)
        return NULL;

    /* Allocate the memory and store the new entry.
     * Insert the element in top, with the assumption that in a database
     * system it is more likely that recently added entries are accessed
     * more frequently. */
    htidx = dictIsRehashing(d) ? 1 : 0;
    size_t metasize = dictMetadataSize(d);
    entry = zmalloc(sizeof(*entry) + metasize);
    if (metasize > 0) {
        memset(dictMetadata(entry), 0, metasize);
    }
    entry->next = d->ht_table[htidx][index];
    d->ht_table[htidx][index] = entry;
    d->ht_used[htidx]++;

    /* Set the hash entry fields. */
    dictSetKey(d, entry, key);
    return entry;
}

htidx = dictIsRehashing(d) ? 1 : 0; → 리해싱 중이라면 1, 아니면 0

redis pub/sub은 어떻게 그 많은 처리를 할 수 있을까?

redis는 기본적으로 단일 스레드 구조다 모든 명령은 하나의 메인 스레드에서 실행된다 → pub/sub의 메시지 발행, 전송도 하나의 스레드에서 처리한다.

핵심은 논블로킹 I/O, 이벤트루프 이다

자세히 설명해보면 리눅스에서 FD → 모든 것은 파일이다

소켓 또한 파일이다

redis는 모든 클라이언트 소켓을 non-blocking으로 설정한다 따라서 어떤 클라이언트가 느리더라도, 그 클라이언트 때문에 Redis 전체가 멈추지 않는다

또한 enpoll 기반의 이벤트 루프로 대기중인 수천개의 소켓(파일)을 효율적으로 모니터링 한다

간단하게 설명하면 select 시스템 콜은, for문 처럼 모든 소켓을 순회한다면 (O(n))

enpoll은

• 감시 대상 등록은 Tree (Red-Black Tree)
• 빠른 검색 / 등록 (epoll_ctl)
• 준비된 이벤트는 Ready List (LinkedList)
• 이벤트 발생한 FD만 반환 (epoll_wait)
• 반응형 구조
• 이벤트가 발생한 FD만 감지 (select는 전수 검사)
• 수천~수만 개 FD 감시 가능

간단하게 소켓중에 준비되거나 보낼게 있는 놈들만 조사할게! 이다

이벤트 감지를 select처럼 for문으로 순회하는게 아닌 LinkedList에서 이벤트 감지 하기 때문에 O(n) FD등록은 red black tree에 하기 때문에 O(log n)

또한 메시지 전파는 매우 빠르다 왜? 사실 아까 말했듯이 구독자 리스트 조회할때, O(n)이긴 하다, 하지만 구독자의 숫자는 적게 설정하기 때문에 n이 작다

또한 실제 메시지를 바로 TCP로 전송하는게 아니라 클라이언트 버퍼에 넣어둔다, 그 후 write 시스템 콜이 가능할때(이벤트 루프에서 클라이언트가 writeable 상태로 감지되면), OS가 알려주면 버퍼에서 꺼내서 전송한다

메인 스레드는 list 순회만 하면 끝난다

[PUBLISH foo "hello"] ↓ [Redis 단일 스레드] ↓ (pubsub_channels["foo"] → clientA, clientB, clientC) ↓ 각 client의 reply 버퍼에 메시지 push ↓ event loop가 write 가능할 때 TCP로 전송

클라이언트 출력 버퍼?

아까 말했듯이

Redis는 메시지를 "바로 소켓으로 보내는 게 아니라, 먼저 출력 버퍼에 넣고, 소켓에 write() 시도함" 그리고 클라이언트가 읽지 않으면, 그 write는 지연되고, 버퍼가 점점 차게 됨.

예시)

• 클라이언트가 SUBSCRIBE room:123
• 누군가 PUBLISH room:123 "hello" 함
• Redis는 room:123 구독자 리스트 순회
• 각 클라이언트에게 addReply() → 출력 버퍼(client->buf)에 저장
• Redis는 event loop에서 소켓에 write() 시도
• 클라이언트가 read() 하지 않으면 → 버퍼 안 비워짐 → 점점 쌓임

Redis 서버 입장에서 “클라이언트에게 보낼 데이터를 임시로 쌓아두는 공간”인데. 클라이언트가 느리거나 받지 않으면, 이 공간에 계속 메시지가 쌓임

Redis는 메시지를 전송할 준비가 되면,클라이언트 소켓에 write() 하기 전에 → buf에 먼저 넣어둠 → output buffer

문제가 되는 상황은

• 클라이언트가 느려서 메시지를 늦게 받음
• Redis는 Pub/Sub 메시지를 계속 전송
• 클라이언트가 느려서 버퍼가 줄지 않음
• output buffer가 한계치를 초과함
• Redis는 해당 클라이언트 소켓을 강제 종료(KICK) 시킴

typedef struct client {
    ...
    list *reply;     // 💬 일반적인 응답 메시지 큐 (linked list)
    char buf[PROTO_REPLY_CHUNK_BYTES]; // 🔄 작고 빠른 응답 캐시 (16KB 정도)
    size_t sentlen;  // 현재까지 write된 양
    ...
}

• buf[] – 소형 버퍼
  • 매우 짧은 응답(예: +OK, :1, $5\r\nhello)은 여기 저장
  • 빠른 전송용, memcpy 최적화
• reply list – 연결 리스트 기반 큐
  • buf로 다 못 담는 경우 → reply list에 분할 저장
  • 일반적으로 Pub/Sub, AOF, 큰 리스트 응답 등이 여기에 쌓임

동작 흐름

1. Redis가 응답을 생성
2. 작으면 → buf[]에 저장
3. 크거나 버퍼 가득 차면 → reply list에 저장
4. 클라이언트 소켓이 writable 상태가 되면 → sendReplyToClient() 호출
5. write()로 전송, 전송된 만큼 buf/list에서 지움

근데만약에 버퍼가 계속 쌓이면?

ex) 느린 클라이언트에게 메시지를 꼐속 보내는 경우 Pub/Sub

client → reply가 너무 커지면, 메모리 누수, 서버 과부하

redis는 이를 감지하고 자동 차단

# redis.conf
client-output-buffer-limit pubsub 32mb 8mb 60

스프링 서버가 웹소켓을 관리하는 동작

클라이언트 ↔ 스프링

스레드 하나당, 클라이언트와 연결하나일까? 놉 자바는 selector 기반의 이벤트 루프가 들어있음 webflux가 아닌데도 어떻게 이벤트 루프가 된다는거지? 스프링 자체가 이벤트 루프를 갖고 있찌 않지만, 서블릿 컨테이너 (tomcat)는 내부적으로 이벤트루프를 갖고있기 때문

웹소켓 연결 흐름 (Spring + Tomcat):

• 클라이언트가 웹소켓으로 handshake → 서블릿 스레드로 처리됨
• 연결이 성립되면 → Tomcat의 WebSocketProcessor가 별도의 비동기 I/O로 WebSocketSession 관리
• 이후 메시지가 오면 → Tomcat의 NIO SelectorThread (Poller Thread)가 감지 → 메시지 처리만 서블릿 스레드 풀에서 소비됨 (default는 200개)

연결 상태 유지

• Tomcat의 Poller Thread가 감시 (NIO 기반)

메시지 수신 시 처리

• ExecutorService로 꺼내서 처리 (서블릿 스레드)

연결 수 제한

• 크게 없음 (FD와 메모리로만 제한됨)

스레드 수 제한

• 메시지가 몰릴 때만 서블릿 스레드가 부족해짐

스프링 ↔ redis

[Client] → WebSocket → [Tomcat 스레드 처리] → RedisTemplate.convertAndSend() ← 동기

                                                                             ↓

                                                                       Redis (PUBLISH)

                                                                          ↓

[RedisMessageListenerContainer 스레드] ← 비동기 → RedisSubscriber.sendMessage()

이 RedisMessageListenerContainer는 백그라운드 스레드에서 돌아가며

Redis Pub/Sub 메시지를 비동기적으로 수신하고,

우리가 등록한 RedisSubscriber.sendMessage() 메서드를 호출 즉, 수신은 완전히 비동기로 동작하고, 서버는 따로 요청 기다릴 필요 없음

서버가 Redis에게 "나 여기에 구독할게" 하고 등록만 해놓고, 이후에는 직접 요청하지 않아도, Redis에서 메시지가 오면 → 자동으로 호출되는 방식

즉,

서버는 수동으로 받을 준비를 하지 않음

Redis가 알아서 메시지 줄 때 → 리스너 스레드가 대신 받고 처리

주의사항

1. 결국 하나의 소켓도 하나의 FD다, 그러면 스프링 입장에선 정말 몇십만명이 소켓 연결이 가능할까? 아니다, 리눅스에서 FD의 soft limit은 1024이고 hard limit은 65535명의 소켓만 연결이 가능한 것이다 그래서 실제로 FD의 limit를 보고 적절한 설정을 해줘야 한다.
2. redis에서 오는 메시지를 스프링에서 받을때 스레드 설정을 잘 해줘야 한다, 너무 많은 스레드는 필요 없긴하지만 단일 스레드로 너무 많은 부하를 받는다면, 그래서 느려진다면 redis의 클라이언트 출력 버퍼의 한계에 도달하고 해당 스프링 서버와 redis의 연결이 kick 유실된다

이전 에는 구독이 끊겼을 경우에 대해서 해결책을 알아봤다 (https://velog.io/@wellbeing-dough/Redis-pubsub%EC%9D%84-%EC%9D%B4%EC%9A%A9%ED%95%9C-%EC%B1%84%ED%8C%85-%EB%8B%A4%EC%8B%9C-%EB%8F%8C%EC%95%84%EB%B3%B4%EA%B8%B0)

이번엔 구독이 안끊겼는데, 중간 메시지가 유실되는 case에 대해서 해결책을 알아보자

이걸 왜 이제서...?

근데 왜 이걸 이제서야 해결할까? 채팅 시스템 구축은 3개월 전에 끝냈는데? 이유는 이렇다

1. 구독이 끊기는 문제는 너무 치명적임. 유저가 직접 새로고침을 눌러서 재 구독을 하지 않는 한, 메시지 자체가 전송이 안됨 하지만, 구독이 끊기지 않았는데, 한두개의 메시지가 전송되지 않는다는 것이 그렇게 치명적일까? 보통의 유저는 채팅을 한번 더 보낼 것이다. 이게 과연 유저에게 우리 서비스에 이탈 사유가 될까?
2. 해결 방법은 생각했었다. 하지만 빠르게 MVP를 내야 하는 상황에서 이 해결 방법을 적용하는 시간 대비 효율이 나올까?

어느정도 정신없이 여러 프로젝트를 끝내고 기술 부채를 해결할 수 있는 시간이 생겼다. 또한 중간 메시지 유실이 얼마나 발생하는지 확실하게 트래킹 해볼 수 있다. 이제 해결해보자

문제

구독이 끊기지 않았는데, 중간에 메시지가 하나가 유실됨.

이게 정상적인 흐름이라고 가정해 보자

하지만 "산본역에서" 라는 메시지가 유실된 것이다

구독이 끊기지 않았기 때문에 6, 7번 메시지가 나간 상황이다.

문제 해결

출처: https://www.cspsprotocol.com/tcp-sequence-number/

TCP는 데이터 전송 시 시퀀스 넘버(Sequence Number)를 사용한다. 보내는 데이터 바이트마다 번호를 붙이고, 수신 측에서는 이 번호가 순차적으로 잘 도착했는지 확인한다.

만약 중간에 누락되거나 순서가 꼬이면?

ACK를 멈추거나

재전송 요청하거나

타임아웃 후 자동 재전송이 일어난다

즉, "데이터 유실 → 감지 → 보완" 흐름을 아주 단단하게 갖춘 프로토콜이다.

여기서 해결책을 얻었다.

1. 위 사진 처럼 메시지에는 채팅방마다 중복되지 않는 고유한 번호를 넣고, 그 번호는 auto increment처럼 천천히 올라간다.
2. 클라이언트는 메시지 고유 번호가 누락됬을 경우, 채팅 내역 재 조회, 재 구독 요청을 한다.

그림으로 보면

이게 정상적인 상황인데 만약에 "산본역에서" 가 누락됬다고 쳐보자

그렇다.

숫자의 증가 원자성을 어떻게 보장할 수 있을까?

유저 두명이 정말 동시에 메시지를 보내면 숫자가 1씩 증가하지 않고 같은 시퀀스 숫자가 두번 나올 수 있다. 어플리케이션 코드 내에서 AtomicInteger로 하나씩 늘릴 수 있지만 스케일 아웃이 되는 순간 그 또한 보장할 수 없다

출처: https://redis.io/docs/latest/commands/incr/

해당 redis 명령어로 원자성을 보장할 수 있다. 왜? redis는 싱글 스레드니까. 그리고 심지어 시간복잡도도 1이고 ACL categories에 무서운 단어도 없다.

하지만 이미, redis에 너무 많은 의존을 하는 채팅 시스템이다.... 물론 redis에 시간 복잡도가 log n 이상인 명령어를 쓰지 않아서 cpu bound는 없지만, 메모리적인 측면에서 좀 생각을 해봐야 한다.

우리가 redis 에서 메모리로 올라가 있는게

1. 채널 목록 + 구독 목록
2. 채팅방 세션
3. 채팅 시퀀스 번호

여기에 저장된 데이터을 최대한 간결하게 하고 아직 간단하게 추론하여 보수적으로 계산 해 보았을 때, 천명정도가 수용 된다. 물론 redis memory 80퍼 찼을 때, 알림도 있지만 어쨌든 redis에 대한 의존도가 너무 높다

간단한 그림

Redis pub/sub의 채팅 유실 문제 알아보기

Redis pub/sub은 기본적으로 채팅이 유실될 수 있는 문제가 있다. 케이스를 하나씩 살펴보자

1. 구독자 연결 상태 문제 구독자가 연결이 끊겨지거나 네트워크로 인해 메시지를 받을 수 없는 경우 메시지가 유실된다 이게 가장 큰 이슈인데, 우리 서비스 기준, 레디스 입장에서 구독자는 ec2의 어플리케이션 프로세스 이다. 첫 구독이 시작되는 순간, ec2와 redis는 tcp기반 소켓으로 열리는데, 만약에 서버가 배포된다면? 소켓 연결이 끊기면서, 구독 리스트가 전부 사라지는 것이다.

2. Redis의 비영구 저장 Redis Pub/Sub은 메시지를 휘발성으로 다루며, 메시지의 상태를 영구적으로 저장하지 않는다. 메시지 큐(예: Kafka)와 달리 Pub/Sub에서는 메시지가 수신자에게 전달된 후 바로 삭제

3. 구독자와 발행자의 동기화 문제 Redis Pub/Sub은 구독자가 특정 채널을 구독한 후에만 메시지를 수신할 수 있음. 구독하기 전에 발행된 메시지는 유실

복합적으로 따져봤을때 가장 중요한 핵심은

구독이 끊기거나 유저의 네트워크 문제로 메시지를 못받으면 끊긴 동안에 메시지가 그대로 유실됨 그래서 다시 구독을 하거나 네트워크가 다시 연결되도 메시지는 못받아봄

그럼 구독은 언제 끊길까?

인프라를 다시 확인해보자

유저의 네트워크 문제는 어쩔수 없다 치자

서버가 redis의 채널에 구독을한다? 그럼 서버와 redis는 tcp로 연결되어있다. 만약에 배포를 해서 서버가 잠깐 내려간다? 그럼 tcp연결이 끊기는거고 구독은 자연스레 해지가 된다. 이는 배포할 때마다 모든 채널에 구독이 해제가 되어있는 것을 보고 확인을 했다.

그래서 최종적인 문제 정의

1. 유저의 채팅중에 구독은 절대 끊기면 안됨
2. 유저가 구독이 끊긴 후에 유실된 채팅을 어떻게 할건지 생각해야됨
3. 배포할때, 모든 서버에 구독이 끊기는 문제를 해결해야됨

해결 과정

1. 유저가 채팅방에 들어오면 구독을 하고 나가면 구독을 끊자

웹소켓에 연결할때 클라이언트는 CONNECTED ACK가 떨어지면 바로 SUBSCRIBE로 구독을 하자 그리고 DISCONNECT가 오면 구독을 끊자 사실 UNSUBSCRIBE가 있긴하다 정석적으로는 유저가 채팅방을 나가면 UNSUBSCRIBE를 보내고 DISCONNECT를 보내는게 맞다 하지만 유저가 어플을 강제 종료 하거나 그랬을 때, 저 두개를 보낼 수 잇을까? 클라이언트 개발자분이 심히 염려된다 하시고 나도 어느정도 동의하기 때문에 DISCONECT만 빠르게 던져주면 서버에서 구독까지 해지해주자 나중에 UNSUBSCRIBE와 DISCONNECT의 책임과 역할이 더 커지면 분리하자

그럼 유저가 어플리케이션을 나간다면, 메시지를 어떻게 받지? 데이터베이스에 채팅 내역을 저장하니까 상관없다.

2. 채팅 기록은 데이터베이스에 저장하자

사실 채팅 기록을 전부 어플이나 웹에 저장하면 그 데이터가 얼마나 커질까? 유저가 어플을 삭제하면 그 기록도 전부 삭제되는거 아닌가? 우리서비스는 웹 앱 다 서비스하는데 웹에서 채팅내역과 앱에서 채팅 내역이 다르면? 절대안된다 일단 유저가 채팅할때 기록은 데이터베이스에 저장은 해야된다

그러면 하나는 해결된다. 유저가 채팅방을 보고 있을땐(여기서 보고있다는 뜻은 눈이 아닌 채팅 기능에 들어와 있는 상태) 구독을 해야하지만 채팅을 안보고 있을 땐, 데이터베이스에 저장하고 유저가 채팅방에 들어오면 채팅 내역 조회 api로 채팅을 안보고잇을 때, 왔던 채팅들을 볼 수 있어진다

채팅 내역은 join을 걸 필요가 없고 빠른 조회를 위해 mongoDB같은 NoSQL에 저장하는게 맞지만, 우리팀은 당장 인프라를 추가할 금액적 리소스가 없기 때문에 일단 RDBMS에 넣어놓자... 인덱스까지 걸었는데 성능 안나오고 유저가 많아져 돈이 많아지면 그때 마이그레이션 하자

3. 서버를 배포할 때 활성화된 채팅방을 전부 조회하고, 구독해볼까?

@PostContructor로 간단하게 구현할 수 있다. 하지만 채팅방은 활성화 되어있지만, 유저가 채팅방에 들어와있지 않은 상태라면 쓸모없는 구독이 생기고 이는 리소스 낭비가 아닐까?

4. 클라이언트에서 웹소켓이 끊기면 자동으로 웹소켓을 바로 재연결하자

우리의 인프라는 블루그린 무중단 배포다. 그렇더라도 서버가 재배포되면 웹소켓도 끊길 수 밖에 없다. 그럼 바로 웹소켓 재연결을 하면 배포된 서버에 연결이 될 것이도 연결과 동시에 해결 과정 1번에 있듯이 구독도 할 것이다.

이게 최선인가?

정말 이것만으로 유저의 채팅 유실 문제를 해결했다 생각할 수 있을까....

1. 만약에 알 수 없는 이유로 채널에 구독이 해지되면?
2. 만약에 알 수 없는 이유로 구독이 되어있지만 메시지가 유실된다면? 우리는 소개팅 서비스인데, 소개팅 전에 채팅하는데 누군가가 답장을 안하고 있다면... 만나기도 전에 서로의 신뢰가 떨어질 수 있는 심각한 문제가 될 수 있다

이정도면 그냥 kafka를 도입할 걸 그랬나? 금액적, 시간적 리소스가 장난이 아니라서 적절한 선이 필요하다

5. ping pong을 커스텀 하자

ping pong은 연결 상태 확인, Keep Alive 유지를 위해 필요하다 일정시간 동안 트래픽이 없다면 유휴 연결로 간주해 연결을 끊기 때문이다. 이 ping pong을 커스텀 해 보자, 우리는 클라이언트에서 ping을 보내고 서버에서 pong을 반환하기로 했다

• 클라이언트는 20초마다 ping을 보내서 채팅방 접속 상태 알림
• 서버는 ping이 오면 Redis에 세션 정보 저장 (TTL 40초)
• 서버가 40초동안 ping을 받지 않았으면 Redis의 timeout으로 자동 세션 만료

이렇게 유저의 채팅 온라인 세션 관리를 할 것이다. 근데 여기서 하나 더 추가해보자

• 유저가 메시지를 보내면 수신자의 세션을 확인함, 그때 세션에 메시지를 저장
• 클라이언트가 ping을 보낼 때, 유저가 마지막으로 수신한 메시지를 서버에 전송
• 서버는 ping을 받을 때, ping에 첨부된 메시지와 세션에 저장된 메시지가 다르면 pong이 아닌 “message mismatch”을 반환
• 클라이언트는 “message mismatch”를 받으면 채팅 내역 다시 조회, 소켓 재연결, 재 구독 요청

이렇게하면 메시지 유실은 많이 개선할 수 있을 것 같다. 그림으로 간단하게 보면

한마디로 user1의 "내일 몇시?"가 유실됬는데, user2의 ping에서 마지막으로 받은 메시지가 "안녕" 이다 user1이 "내일 몇시?"를 보낼 때, 어플리케이션 서버에서 세션정보에

key: "user1:UUID1423,roomId:UUID123" value: 내일 몇시?

이렇게 저장이 되었을 거고 user2가 ping을 보낼 때, user2가 마지막으로 받은 메시지인 "안녕"을 보내겠지?

그럼 서버는 세션에 저장된 마지막 메시지와 클라이언트가 준 마지막 메시지가 다르면 pong이 아닌 message mismatch를 보낸다 그러면 클라이언트는 채팅 내역 다시 조회, 소켓 재연결, 재 구독 요청을 한다 이러면 메시지가 유실될 일은 크게 줄어든다

물론 이해를 돕기위해 텍스트 기반으로 설명했지만, 실제로는 텍스트가 아닌 유니크한 id를 기반으로 동작한다

해치웠나?

nope

구독자 연결 상태 문제에 대해서 해결을 했지만

아직 문제가 남아있다.

구독은 안끊겼는데, 중간 메시지가 유실된 경우는 어떻게 해결할 수 있을까?

2편에서 알아보자 https://velog.io/@wellbeing-dough/Redis-pubsub%EC%9D%84-%EC%9D%B4%EC%9A%A9%ED%95%9C-%EC%B1%84%ED%8C%85-%EB%8B%A4%EC%8B%9C-%EB%8F%8C%EC%95%84%EB%B3%B4%EA%B8%B0-2

예전에 서버에서 oome가 터졌었다 예전에 한번 터졌던 적이 있어서

             -XX:+HeapDumpOnOutOfMemoryError \
             -XX:HeapDumpPath=/tmp \

jar파일 만들 때 이런 옵션으로 oome가 터지면 자동으로 /tmp에 힙 덤프 파일을 만들어 놨다 물론 덤프 파일 만들고 자동으로 서버 다시 동작하게도 해놨다 어려운건 아니고 elastic beanstalk에서 알아서...

힙 덤프 파일을 분석해봤는데

아 sksamuel.scrimages는 이미지 webp로 압축하려고 했던건데 여기서 oome가 터졌다는 것을 단번에 알 수 있었다

일단 급한대로 스왑 메모리 설정하고 힙 메모리 늘려놓고

            JAVA_OPTS="-Xms512m -Xmx1024m"

우리 서버가 t3.small 이라서 메모리가 1기가다 여기다가 스왑 메모리 1기가 넣어놓고 자바 프로세스 메모리 최대 1기가 넣어놨다

또한 이미지 webp압축은 일단 일시정지 했었다

이제부터 webp압축을 다시 돌려놓고 근본적인 문제를 해결 해 보자

문제

1. 힙 덤프 분석을 자세히 봐보자

Pixel이라고 되어있다. 여기서 예상해볼 수 있는 것은 이미지의 메모리 때문이 아니라 이미지의 해상도 ex)1024x1024일 것으로 예상할 수 있다

그렇다면 스펙이 같은 서버를 빠르게 구축해보고 한번 간단하게 테스트를 해보자

여기 옛날에 했던 프로젝트의 로고가 마침 4267x4267인 것을 볼 수 있다 하지만 용량은 113KB이다

업로드를 해 봤다 100메가에서 웃도는 메모리가 이미지 하나를 업로드 하자 850메가로 치솟고 2분 후 또 업롣드를 하자 850까지 치솟는 것을 볼 수 있다. Major GC도 두번 일어난다 이러니 메모리가 터지지... 그렇다면 내 컴퓨터의 바탕화면을 전체 캡쳐해보자

용량은 7.2로 상당히 무겁지만, 해상도가 2546x1440이다 이걸 업로드 해보면

350에서 450이 되는 것을 볼 수 있다.

로고를 올린 후 바탕화면을 올려서 살짝 불공정할 수 있지만 가설은 확실하게 검증되었다

확실하게 다시 알아보자

2. 원인 제대로 파악

해상도가 높을수록 메모리 사용량이 더 크다

이미지를 처리하기 위해 메모리로 로드하면, 각 픽셀의 색상 정보를 메모리에 할당해야 한다. 일반적으로 한 픽셀당 3~4바이트(RGB 혹은 RGBA)를 차지한다 예를 들어, 100x100 해상도의 이미지는 10,000개의 픽셀이므로 메모리에서 약 3040KB가 필요하지만, 4000x4000 해상도의 이미지는 1,600만 픽셀로, 메모리에서 약 4864MB가 필요하게 된다 즉, 해상도가 높을수록 픽셀 수가 증가하고 메모리 사용량이 비례하여 증가한다

3. 이미지를 압축하기 전, resize를 진행 해야 한다

1. Thumbnailator Thumbnailator는 Java용 오픈소스 썸네일 생성 라이브러리로, 이미지 리사이징을 쉽게 구현할 수 있다 간결한 API로 코드 양을 줄이며, 빠르게 고품질의 썸네일을 생성할 수 있고 성능 최적화가 잘 되어 있어 높은 품질과 성능이 좋지만 고급 이미지 조작(회전, 필터링 등)에 대해 제한적이다

2. OpenCV (Java bindings) OpenCV는 컴퓨터 비전 작업을 위한 강력한 오픈소스 라이브러리로, Java 바인딩을 통해 Java에서도 사용할 수 있다 이미지 리사이징 기능은 물론 다양한 컴퓨터 비전 기능을 제공한다. 빠른 성능과 고품질 이미지 처리가 가능하고 GPU가속을 사용할 수 있어 대용량 이미지나 고성능이 필요한 작업에 좋지만, 외부 라이브러리 설치와 Java 프로젝트에 적용하려면 추가 설정이 필요하다. 단순 리사이징 용도로는 너무 과하고 무겁다

3. Image.getScaledInstance 특징: Image.getScaledInstance는 Java에서 기본적으로 제공하는 Image클래스의 메서드 이다 기본 Image 클래스를 사용하여 빠른 구현을 가능하게 하지만 크기 조절 시 이미지가 흐릿해질 수 있고, 성능이 떨어진다

4. imgscalr imgscalr imgscalr는 경량의 Java 이미지 리사이징 라이브러리로, 성능과 품질을 모두 고려한 이미지 스케일링이 가능하다 내부적으로 Graphics2D를 사용하여 동작하지만 보다 품질이 좋고 빠른 리사이징을 제공하며, 설정이 간단하다 이미지 품질을 위한 다양한 모드(ULTRA_QUALITY, FAST 등)를 제공하여, 성능과 품질의 균형을 쉽게 맞출 수 있다 경량 라이브러리이므로 추가적인 의존성 문제를 최소화하면서 고품질 리사이징을 수행할 수 있습니다.

5. Graphics2D Graphics2D는 Java AWT 패키지에서 제공하는 고급 그래픽 변환 및 렌더링이 가능하다 장점으로는 이미지 품질을 세밀하게 조절 할 수 있고, 크기 조절 외에도 회전, 뒤집기, 기타 2D 그래픽 처리가 가능하다 단점으로는 사용법이 비교적 복잡하고 더 많은 코드 작성이 필요하다 대규모 이미지 처리 시 성능에 영향을 준다

다 직접 리사이징 해서 눈으로 직접 봐보고 결정 해 보자 확실히 Graphics2D, imgscalr가 정말 빨랐다 하지만 화질이 너무 안좋아진다. Image.getScaledInstance도 엄청 빨랐고 화질이 처음엔 깨졌지만 그나마

getScaledInstance(imageSize.getWidth(), imageSize.getHeight(), Image.SCALE_SMOOTH);

SCALE_SMOOTH 로 스무스하게 했을때 좀 괜찮아졌다

Thumbnailator를 쓰기로 했다

해결

    private File resizeImage(File imageFile) throws IOException {
        BufferedImage resizedImage = Thumbnails.of(imageFile)
                .size(1440, 1440)
                .asBufferedImage();
        File tempFile = new File(imageFile.getParent(), "resized_" + System.currentTimeMillis() + ".png");
        ImageIO.write(resizedImage, "png", tempFile);
        return tempFile;
    }

    private File convertToWebP(File imageFile) throws IOException {
        ImmutableImage image = ImmutableImage.loader().fromFile(imageFile);
        String fileName = generateUniqueFileName("webp");
        File outputFile = new File(imageFile.getParent(), fileName);
        image.output(WebpWriter.DEFAULT, outputFile);
        return outputFile;
    }

사이즈를 여러개로 조합해본 결과 성능도 잡고 프론트 엔드 동료와 합의점도 찾은 1440x1440이 적절하다는 것을 알았고 1440x1440으로 했다

결과

이미지 결과는 하나만 봐보자 (2024 롤드컵 페이커의 슈퍼플레이)

압축 전

압축 후

솔찍히 진짜 별 차이 없는데 4.1MB를 147KB로 압축 시켰다 굳굳

메모리도 봐볼까? 물론 Major GC가 생기긴 한다 하지만 300~400 메모리 왔다갔다 한다. 기존에 사진 두장에 1기가 턱밑까지 뚫은거 생각하면 양반이다

회고

이미지 리사이징, 압축 과정이 서버에 상당히 큰 자원 소모를 하게 하는 것을 알았다. 가장 좋은 방법은 s3에 이미지 올리면 람다 함수 실행시켜서 거기서 리사이징 압축 하는 방법일 것 같다. 이왕 하는거 s3의 presigned url 써서 프론트가 직접 s3에 올리는 것도 좋아 보이지만, 일단 그럴려면 프론트 동료의 리소스도 필요하고 나중에 나랑 프론트 동료가 둘다 널널할 때 조심스럽게 얘기를 꺼내봐야 할 것 같다

그리고 앞으로 라이브러리 생각없이 막 갔다 쓰지말고 api 하나 개발할때마다 상용 서버에서도 테스트 해보면서 리소스 확인도 해야겠다 아직 dev 환경을 배포하기엔 돈이 너무 없다...

참고: https://babble-dev.tistory.com/39 https://developers.google.com/speed/webp/faq?hl=ko#can_a_webp_image_grow_larger_than_its_source_image https://medium.com/@angal2310/image-resize-webp-4dffafd68683

소개팅 주선 서비스에서 추천 데이터가 있다. 다른 데이터에 비해 추천 데이터는 엄청나게 많이 생긴다. 보통 한 유저당 일주일에 평균적으로 510개의 추천 데이터가 생기기 때문에 추천 데이터가 많이 쌓여있는 상태다 그리고 앞으로 유저가 많아질수록 추천 데이터 증가 폭은 510배 늘어나므로 앞으로도 계속 많이 쌓일 예정이다 쿼리 튜닝을 한다면 지금 이 부분이 가장 1순위라고 생각해서 튜닝을 진행 해 봤다

문제 해결

기존

SELECT
    r.id AS recommendationId,
    up.career_detail AS careerDetail,
    up.age AS age,
    up.user_representative_images AS userRepresentativeImages
FROM
    recommendation r
INNER JOIN
    users u ON r.partner_user_id = u.id
INNER JOIN
    user_profile up ON u.user_profile_id = up.id
WHERE
    r.type = 'NORMAL'
    AND r.to_user_id = UUID_TO_BIN('0192d3d9-0ccd-79a8-8dac-cc948a0b71eb')
    AND r.deleted_at IS NULL;
ORDER BY r.id ASC

기존 쿼리이다 join이 2번 들어가는 상황이다 일주일에 한번씩 추천 데이터가 초기화되기 때문에 유저가 한번에 볼 수 있는 추천 데이터는 최대 10개 뿐이라 현상황에서는 정렬은 따로 넣어주지 않았다 근데 왜 id로 정렬하냐면 ORDER BY를 안 써도 정렬이 되니까 ORDER BY를 쓰면 두 번 정렬이 된다고 생각할 수 있는데, 그렇지 않다고 한다. MySQL서버는 정렬을 인덱스로 처리할 수 있는 경우, 부가적으로 불필요한 정렬 작업은 수행하지 않는다. 그러므로 ORDER BY를 쓰는 습관을 가지는게 좋다. 혹시나 실행 계획이 변경 되었을 때, ORDER BY가 명시되지 않았다면 원하는 결과가 나오지 않을 수도 있다.

지금 보면 join두번과 WHERE절에서 성능 오버헤드가 발생하지 않을까 예상이 되긴 하지만 실행계획을 봐보자

먼저 EXPLAIN ANALIZE를 사용해서 봐보자

-> Nested loop inner join  (cost=101289 rows=0.0199) (actual time=34.6..266 rows=13 loops=1)
    -> Inner hash join (r.PARTNER_USER_ID = u.ID)  (cost=101289 rows=0.0199) (actual time=34.5..266 rows=13 loops=1)
        -> Filter: ((r.TO_USER_ID = <cache>(uuid_to_bin('0192d3d9-0ccd-79a8-8dac-cc948a0b71eb'))) and (r.`type` = 'NORMAL') and (r.DELETED_AT is null))  (cost=50639 rows=99.6) (actual time=34.4..266 rows=13 loops=1)
            -> Table scan on r  (cost=50639 rows=996200) (actual time=0.0775..234 rows=1e+6 loops=1)
        -> Hash
            -> Covering index scan on u using FK_USERS_USER_PROFILE_ID  (cost=0.45 rows=2) (actual time=0.0915..0.0931 rows=2 loops=1)
    -> Single-row index lookup on up using PRIMARY (ID=u.USER_PROFILE_ID)  (cost=0.00301 rows=1) (actual time=0.00432..0.00434 rows=1 loops=13)

들여쓰기는 호출 순서를 의미하며, 실제 실행 순서는

들여쓰기가 같은 레벨에서는 상단의 위치한 라인이 먼저 실행 들여쓰기가 다른 레벨에서는 가장 안쪽에 위치한 라인이 먼저 실행

1. recommendation 테이블 조건 필터링 (Table scan on r)

• 작업: recommendation 테이블에서 조건을 충족하는 행을 찾기 위해 전체 테이블 스캔을 수행
• 조건: type = 'NORMAL', to_user_id = UUID_TO_BIN('0192d3d9-0ccd-79a8-8dac-cc948a0b71eb'), deleted_at IS NULL
• 추가 정보: recommendation 테이블에서 이 조건을 모두 만족하는 행을 찾아냄
• 실행 세부 정보:
  • 비용: cost=50639, 예상 행 수: rows=996200.
  • 실제 시간: actual time=0.0775..234 (이 단계에서 100만 개의 행을 스캔하며, 최종적으로 13개의 행이 조건에 맞는 행으로 선택)
• 이 단계에서 반환된 13개의 행이 다음 조인 단계로 전달.

2. 해시 테이블 생성 및 users 테이블 조회 (Covering index scan on u using FK_USERS_USER_PROFILE_ID)

• 작업: users 테이블에서 recommendation.partner_user_id와 일치하는 users.id를 찾아 해시 테이블을 생성
• 조건: r.partner_user_id = u.id를 만족하는 행을 FK_USERS_USER_PROFILE_ID 인덱스를 사용하여 조회
• 실행 세부 정보:
  • 비용: cost=0.45, 예상 행 수는 rows=2.
  • 실제 시간: actual time=0.0915..0.0931 (최종적으로 2개의 행이 검색).
• 해시 조인을 위해 users 테이블에서 조회한 데이터를 해시 테이블에 저장

3. 해시 조인을 사용한 recommendation과 users 결합 (Inner hash join)

• 작업: recommendation의 필터링된 13개의 행을 해시 조인을 통해 users 테이블의 결과와 결합.
• 조건: recommendation.partner_user_id = users.id
• 실행 세부 정보:
  • 비용: cost=101289, 예상 행 수는 rows=0.0199.
  • 실제 시간: actual time=34.5..266, 13개의 행을 반환.

4. user_profile 테이블에서 단일 행 조회 (Single-row index lookup on up using PRIMARY)

• 작업: 조인된 결과 13개 행 각각에 대해 user_profile.id = users.user_profile_id 조건을 사용하여 user_profile 테이블에서 단일 행 조회.
• 조건: ID = u.USER_PROFILE_ID
• 실행 세부 정보:
  • 비용: cost=0.00301, 예상 행 수는 rows=1.
  • 실제 시간: actual time=0.00432..0.00434, 13번 반복하여 user_profile에서 13개의 최종 결과 행을 반환

또한 Inner Hash Join이 적용된 것을 볼 수 있따. MySQL 8.0부터 도입된 Batched Key Access (BKA) 와 Block Nested Loop (BNL) 방식의 최적화 덕분에, 쿼리에 따라 Hash Join을 선택할 수 있다

BKA랑 BNL에 대해서

MRR과 배치 키 액세스(mrr & batch_key_access)(off) MRR (multi-range read), DS-MRR(disk sweep multi-range read)라고도 함.

기존에는 네스티드 루프 조인 방식을 사용함 드라이빙 테이블의 레코드를 한 건 읽어서 드리븐 테이블의 일치하는 레코드를 찾아서 조인을 수행하는 방식 네스티드 루프 조인 방식에서는 조인 처리를 MySQL 엔진이 담당하고, 실제 레코드를 검색하고 읽는 부분은 스토리지 엔진이 담당, 이런 방식에서는 스토리지 엔진에서는 최적화를 수행할 수 없음

이런 단점을 보완하기 위해 조인 버퍼에 조인 대상을 버퍼링한다. 조인 버퍼에 레코드가 가득 차면 MySQL 엔진은 버퍼링된 레코드를 스토리지 엔진으로 한 번에 요청한다. 이렇게 해서 디스크 읽기를 최소화할 수 있다. 이 방식을 MRR이라고 한다.

MRR을 응용해서 실행되는 조인 방식을 BKA(Batched Key Access) 조인이라고 한다. 부가적인 정렬 작업이 필요해서 성능이 저하되기도 한다. 결론: 버퍼링된 걸 원기옥처럼 모아서 스토리지 엔진에 전달하기 때문에 디스크 접근 횟수는 줄어든다.

블록 네스티드 루프 조인(on) 조인의 연결 조건이 되는 칼럼에 모두 인덱스가 있는 경우 사용되는 조인 방식이다

네스티드 루프 조인과 차이점은 조인 버퍼 사용 여부와 조인에서 드라이빙 테이블과 드리븐 테이블이 어떤 순서로 조인되느냐다. 블록 네스티드 루프 조인에서는 조인 버퍼가 사용된다. 실행 계획에서 Extra 칼럼에 Using join buffer가 표시되면 조인 버퍼를 사용한다는 것을 의미한다.

조인은 드라이빙 테이블에서 일치하는 레코드의 건수만큼 드리븐 테이블을 검색하면서 처리된다. 그래서 드리븐 테이블을 검색할 때 인덱스를 사용할 수 없는 쿼리는 느려진다.

옵티마이저는 드라이빙 테이블에서 읽은 레코드를 메모리에 캐시(조인 버퍼)한 후 드리븐 테이블과 이 메모리 캐시를 조인하는 형태로 처리한다. 조인 버퍼가 사용되는 쿼리에서는 조인의 순서가 거꾸로인 것처럼 실행된다. A가 드라이빙, B가 드리븐 테이블일 때, A에서 검색된 레코드를 조인 버퍼에 담아두고, B의 레코드를 먼저 읽고 조인 버퍼에서 일치하는 레코드를 찾는 방식으로 처리된다.

결론: 드라이빙 테이블을 먼저 읽어 메모리에 올려둔 후, 드리븐 테이블을 읽으면서 캐싱된 드라이빙 테이블값과 연계지어 조인을 처리 / 네스티드 루프 조인은 조인 버퍼를 사용하지 않지만, 얘는 조인버퍼를 사용함

그렇다면 왜 옵티마이저는 recommendation과 user는 해시조인을 선택하고 user_profile에는 nested Loop Join을 선택했을까?

users 테이블이 크지 않지만, 한 번 해시 테이블을 생성하면 레코드가 많은 recommendation 테이블의 모든 일치하는 레코드를 빠르게 조회할 수 있으므로 Hash Join 선택

user_profile 테이블은 PRIMARY 키 (ID)를 사용한 단일 행 조회가 가능 이 상황에서 해시 테이블을 만드는 비용보다 단일 행을 직접 조회하는 비용이 더 적다

조인 순서에 의한 영향: 옵티마이저는 recommendation과 users의 조인을 먼저 실행하고, 그 결과와 user_profile을 조인한다 이전 조인 결과가 적은 양의 데이터라면, 남은 테이블에 대해 Nested Loop Join을 선택해 빠르게 단일 행을 조회하는 것이 효율적

일단 지금은 join을 할 때 FK로 join을 하기 때문에 예상대로 클러스터링 인덱스로 성능 오버헤드가 크지 않을 것으로 예상된다

이정도의 시간이 걸린다

추천 테이블의 type이 ALL로 표시되어 있는데, 이는 해당 테이블에 대해 풀 테이블 스캔이 발생하고 있다는 것을 의미한다 추천 테이블에 적절한 인덱스를 추가해보자

인덱스 적용 후

CREATE INDEX idx_recommendation_type_to_user_id_deleted_at ON recommendation (type, to_user_id, deleted_at);

이렇게 다중 칼럼 인덱스를 생성 해 줬다

실행 계획은 이렇다.

확실히 빨라진게 보인다 솔찍히 320ms -> 37ms 는 지금 당장 큰 의미는 없지만 추후에 추천 데이터는 일주일마다 엄청나게 늘어나기 때문에 미리미리 했다

추후에 정렬기준이 생긴다면 거기에도 인덱스를 걸어줘야 할 것 같다 그리고 인덱스를 걸었으니, 나중에 쿼리나 DB스펙이 변경된 경우 인덱스도 잘 관리해 줘야겠다

다중 칼럼 인덱스는 지정된 컬럼 순서에 따라 왼쪽부터 차례로 접근해야 최적의 성능을 제공한다 따라서, 인덱스는 to_user_id, type, deleted_at의 순서로 조건을 사용할 때 가장 효과적이다

소개팅 주선 서비스에서 프로필을 입력하는 기능(이하 프로필 설문)이 있다. 다른 도메인보다 프로필을 입력하는 질문 수가 많을 수 밖에 없다

페이스북은 사용자가 자기소개를 작성하고 사진과 신상정보를 추가하도록 하면서 사용자가 참여하도록 하고 심리적인 보상을 제공한다 페이스북에서 자기소개를 만드느라 몇 시간을 보내 본 사람이라면 여기에 대단히 만족스러운 몰입 상태를 느끼게 하는 어떤 것이 존재한다는 점을 알 것이다

출처: 션 엘리스, 그로스 해킹: 스타트업을 위한 실전 마케팅 전략 2020.

설문을 조금 딱딱하지 않게, 뭔가 신상 캐는 느낌이 나지 않게 하기 위해 구글폼 보다는 우리 서비스에 임베드 시킬 수 있으면서 부드러운 ux/ui를 제공하는 타입폼을 사용하기로 했다

하지만 타입폼에서 생각보다 많은 유저 이탈이 있어서 생각해 봤는데 유저 입장에서 프로필 질문이 50개 정도 있는데 한번에 다 입력하기가 쉽지가 않다. 예를들어 오전에 출근시간에 30번 질문까지 답변하고 오후 퇴근시간에 30번부터 질문을 입력하는 상황이 있다면 이게 1번 질문부터 다시 시작하는게 이유였다

이걸 해결 할 방법으로 여러가지가 있지만, 다른 파트에서 병목이 생겨 내가 시간이 좀 남기도 하고 타입폼은 유료 서비스라 돈도 무시할 수 없고 해서 타입폼을 자체 구현 해보겠다 했다

문제 상황

1. 질문내용은 추후에 엄청나게 바뀔 가능성이 있다. 굳이 개발자한테 "이 질문 추가해주세요" "이 질문 삭제해주세요" "이 질문 문구 수정해주세요" 할 필요 없이 어드민 페이지에서 이런 좋은 관리자 ui에서 수정할 수 있게 해야 한다

2. 위 사진처럼 엄청나게 많은 분기가 있는데 정확한 분기 처리와 유지보수가 용이하게 짜야 한다

3. 유저가 답변을 하다가 나중에 다시 들어오더라도 직전에 답변하던 질문으로 돌아올 수 있어야 한다

해결

create table male_profile_survey_question
(
    ID              binary(16)     not null primary key,
    STEP_NUMBER     int            not null,
    COLUMN_TYPE     varchar(50)    not null,
    TYPE            varchar(50)    not null,
    TITLE           text           not null,
    SUB_TITLE       text           not null
);

create table male_profile_survey_answer
(
    ID                    binary(16)     not null primary key,
    NEXT_QUESTION_ID      binary(16)     null,
    QUESTION_ID           binary(16)     not null,
    LABEL                 int            not null,
    CONTENT               text
);

이렇게 질문과 답변을 데이터베이스 화 해봤다

우리 서비스는 남성 여성에 따라 설문 내용이 아예 달라져서 두개를 따로 했다 하지만 남성 여성의 설문에 대한 테이블 스펙은 동일하다

question은 질문내용, answer은 해당 질문의 답변할 수 있는 답변 내용들 이다

answer 테이블에에 next_question_id 칼럼을 넣어서 이 answer의 다음 question이 뭔지 알 수 있게 하였다

또한 유저 테이블에 question_id를 넣어서 이 유저가 해야 할 다음 question이 뭔지 알게 하였다

현재 나의 설문 단계 조회 구현은 이렇게 하였다 지금 현재는 좀 더티코드긴한데

        if (userProfile.getGender() == Gender.MALE) {
            MaleProfileSurveyQuestion question = maleProfileSurveyQuestionReader.readById(userProfile.getMaleProfileSurveyQuestionId());
            List<MaleProfileSurveyAnswer> answers = maleProfileSurveyAnswerReader.readAllByQuestionId(question.getId());
            List<SurveyAnswerHttpResponse> answerContents = answers.stream()
                    .map(answer -> new SurveyAnswerHttpResponse(answer.getId(), answer.getContent()))
                    .collect(Collectors.toList());
            return new GetProfileSurveyHttpResponse(question, answerContents);
        }

이렇게 해당 userProfile에서 현재 설문지 question_id를 가져와서 질문을 읽어온다 그리고 그 질문에 해당되는 답변들 리스트를 가져와서 반환해준다

그럼 답변은 어떻게 해야 할까?

    public void answerProfileSurvey(UpdateProfileSurveyAnswerRequest request, UUID userId) {
        UserProfile userProfile = userProfileReader.readByUserId(userId);
        if (request.isListType()) {
            List<String> contents = request.getListContentByType();
            MaleProfileSurveyAnswer answer = maleProfileSurveyAnswerReader.readById(request.getAnswerId());
            userProfileWriter.handleUserProfileListSurvey(request.getColumnType(), contents, userProfile, answer.getNextQuestionId());
            return;
        }
        String content = request.getSingleContentByType();
        MaleProfileSurveyAnswer answer = maleProfileSurveyAnswerReader.readById(request.getAnswerId());
        userProfileWriter.handleUserProfileSurvey(request.getColumnType(), content, userProfile, answer.getNextQuestionId());
    }

코드가 더럽긴하다

일단 답변이 List(다중 선택), String(단일 선택)인지 파악한다\

그 답변을 저장하고

그리고 클라이언트가 주는 answer_id(나의 설문 단계 조회에서 알 수 있다) -> 유저가 답변 한 답변 id가 온다 그럼 그걸로 답변을 조회하고

    public void surveyName(UserName name, UUID nextQuestionId) {
        this.name = name;
        updateProfileSurveyQuestionId(nextQuestionId);
    }
    public void updateProfileSurveyQuestionId(UUID nextQuestionId) {
        if (this.gender == Gender.MALE) {
            this.maleProfileSurveyQuestionId = nextQuestionId;
            return;
        }
        this.femaleProfileSurveyQuestionId = nextQuestionId;
    }

해당 답변의 다음 question_id를 유저 프로필에 저장해준다

그럼 유저는 다음 질문으로 넘어간다

    public void handleUserProfileSurvey(ProfileColumnType columnType, String content, UserProfile userProfile,
                                         UUID nextQuestionId) {
        switch (columnType) {
            case NAME -> userProfile.surveyName(new UserName(content), nextQuestionId);
            case AGE -> userProfile.surveyAge(Age.fromCompactDateFormat(content), nextQuestionId);
            case PHONE_NUMBER -> userProfile.surveyPhoneNumber(new PhoneNumber(content), nextQuestionId);
            ... etc
            default -> throw new ProfileSurveyAnswerTypeMismatchException(
                    ErrorCode.PROFILE_SURVEY_ANSWER_TYPE_MISMATCH_ERROR,
                    ErrorCode.PROFILE_SURVEY_ANSWER_TYPE_MISMATCH_ERROR.getStatusMessage()
            );
        }
        userProfileRepository.save(userProfile);
    }

물론 설문 내용을 디비에 저장하는건 유지보수가 쉽지 않다.... 사실 타입폼으로 한다 해도 없던 설문 내용이 추가된다하면 결국엔 user_profile 테이블에 칼럼을 추가해야되고 코드가 추가되는건 어쩔 수 없긴 하다

하지만 여기서 큰 문제가 발생한다

새로운 문제 상황

오키 그렇다면 유저가 뒤로가기는 어떻게 하지?

이게 진짜 큰 문제인게

29번째 설문에서 뒤로가기를 눌러도 13번부터 28번까지 어디로 가야할지 모르는게 문제이다 answer은 다음 질문을 갖고잇는거지 question이 이전 질문을 갖고있을 순 없다 (설문의 분기가 존재해서...)

유저가 뒤로가기를 한다 해도 user profile에는 설문해야할 question_id가 있기 때문에 뒤로 못간다

정말 많은 고민과 더티 빡구현 코드를 하나하나 작성해야 하나 고민이 들었다 더티 빡구현 코드는 이제 유저 칼럼 하나하나 확인하면서 이전 답변이 뭐였는지 확인하는건데 이러면 너무 유지보수가 안된다

그러던 중 너무 좋은 생각이 났다

user_profile에 칼럼을 하나 파는거다 그리고 그 테이블과 연결하는 엔티티의 필드엔 Stack자료구조로 하는거다 그러면 stack에는 유저가 설문을 제출할 때 마다 제출한 question_id를 stack에 계속 push 하는거다 만약에 유저가 뒤로 가고싶다? 그럼 stack에서 pop하면 된다

    @Convert(converter = StringStackConverter.class)
    private Stack<String> profileSurveyQuestionStack;

엔티티에 이렇게 추가해준다 사실 DB와 JPA에는 자료구조를 처리할 수 있는 좋은 방법이 딱히 없다 하나 있긴 한데 @ElementCollection, @CollectionTable 을 사용하여 컬렉션의 요소를 별도 테이블에 매핑할 수 있긴 한데 테이블을 분리하기 싫어서

이렇게 했다

@Converter
public class StringStackConverter implements AttributeConverter<Stack<String>, String> {

    private static final String SPLIT_CHAR = ",";

    @Override
    public String convertToDatabaseColumn(Stack<String> stack) {
        return (stack != null && !stack.isEmpty()) ? String.join(SPLIT_CHAR, stack) : null;
    }

    @Override
    public Stack<String> convertToEntityAttribute(String string) {
        Stack<String> stack = new Stack<>();
        if (string != null && !string.isEmpty()) {
            List<String> items = Arrays.asList(string.split(SPLIT_CHAR));
            stack.addAll(items);
        }
        return stack;
    }
}

JPA AttributeConverter를 이용해 Stack 타입의 데이터를 데이터베이스에 String 형태로 저장하고, 다시 엔티티에 불러올 때는 Stack으로 변환하는 기능을 구현했다 StringStackConverter는 Stack 데이터를 콤마(,)로 구분된 String으로 직렬화하고, 이 직렬화된 문자열을 다시 Stack으로 복원할 수 있게 해주게 했다

Stack 데이터를 콤마(,)로 구분된 String으로 변환하여 데이터베이스에 저장한다 Stack이 null이거나 비어있다면 null을 반환하고, 데이터가 있다면 String.join()을 사용해 스택의 모든 항목을 하나의 문자열로 만들어준다

데이터베이스에서 불러온 String을 다시 Stack으로 변환하여 엔티티의 속성으로 반환 string이 null이거나 비어있지 않은 경우, split(SPLIT_CHAR)로 문자열을 분리해 항목 리스트를 만들고, 이를 Stack에 넣는다

엔티티의 profileSurveyQuestionStack 필드는 StringStackConverter를 이용해 데이터베이스에 String 형태로 저장 이 필드가 @Convert 어노테이션을 통해 StringStackConverter와 연결되므로, 데이터베이스와의 저장 및 불러오기 시 자동으로 변환

저장 예: Stack stack = ["Q1", "Q2", "Q3"] → 데이터베이스에 "Q1,Q2,Q3" 형태로 저장 불러오기 예: "Q1,Q2,Q3" → Stack stack = ["Q1", "Q2", "Q3"]

    public void backwardProfileSurvey() {

        if (profileSurveyQuestionStack.isEmpty()) {
            throw new EmptyProfileSurveyStackException(
                    ErrorCode.EMPTY_PROFILE_SURVEY_STACK_ERROR,
                    ErrorCode.EMPTY_PROFILE_SURVEY_STACK_ERROR.getStatusMessage()
            );
        }
        this.femaleProfileSurveyQuestionId = UUID.fromString(this.profileSurveyQuestionStack.pop());
    }

뒤로 가기 구현은 이렇게 했다

물론 설문을 답변할 때도

    public void updateProfileSurveyQuestionId(UUID nextQuestionId) {
        if (this.gender == Gender.MALE) {
            this.profileSurveyQuestionStack.push(this.maleProfileSurveyQuestionId.toString());
            this.maleProfileSurveyQuestionId = nextQuestionId;
            return;
        }
        this.profileSurveyQuestionStack.push(this.femaleProfileSurveyQuestionId.toString());
        this.femaleProfileSurveyQuestionId = nextQuestionId;
    }

이런식으로 다음 질문을 넣기 전에 이전 질문을 stack에 push해서 트래킹이 가능하게 했다

또한 설문을 끝까지 전부 제출하고 나면 stack은 필요없다 심지어 질문이 50개니까 제출 후엔 칼럼내부에 너무 많은 값을 차지 하고 있고 엔티티 조회할때마다 스택 내부에 너무 많은 값이 있어서 DB의 테이블 메모리와, 힙 메모리가 아깝다 때문에 설문을 제출하면

    public void submitSurvey() {
        this.registrationStep = RegistrationStep.findNextRegistrationStep(this.registrationStep);
        if (this.profileSurveyQuestionStack != null) {
            this.profileSurveyQuestionStack.clear();
    }

stack 내부를 비우도록 했다

회고

개발자한테 "이 질문 추가해주세요" "이 질문 삭제해주세요" "이 질문 문구 수정해주세요" 할 필요 없이 어드민 페이지에서 수정할 수 있게 하려고 했는데

이 질문 추가해주세요 는 못할 것 같다. 만약에 원래 자신의 키를 받지 않았는데 "내가 키가 몇인지 질문 추가해주세요" 할 필요없이 관리자 페이지에서 할 순 없을 것 같다. 왜냐하면 유저 테이블에 키 칼럼을 추가해야되고 코드를 수정해야되기 때문이다... 그래도 이건 같은 상황이면 타입폼에서도 서버 코드를 건들여야하기 때문에 일단 하지 않도록 하자

또 만들고 보니 설문중 이탈한 유저가 어디에서 이탈했는지 알기 쉬워서 어떤 질문 내용이 유저한테 부담스러운지 알 수 있게 되었다 나이스 ㅋㅋ

타입폼 안쓰게 되서 아낀 돈으로 마케팅에 투자하고 유저 많이 왔으면 좋겠다

사실상 이거 조금만 더 고도화 시키면 진짜 BtoB 프로젝트 하나짜리인데 우리 프로젝트 내부에선 기능 개발 하나인게 아쉽다... 다 만들고 나서 'task: 타입폼 자체구현' 에 완료 띡 눌렀을 때의 현타... 하지만 정말 너무 좋은 경험을 했고 Stack 구상은 지금 다시 생각해도 도파민 오버플로우... 시간이 늦었다 내일은 일찍 일어나서 더티코드 리펙토링좀 하자

• 우리 서비스는 소개팅 서비스로 유저와 유저가 상호작용 하기 때문에 url이나 message body에 상대방의 userId가 들어가는 경우가 종종 있다 예를들어 "/v1/어쩌고/저쩌고/{user-id}" 를 사용해서 추천, 매칭이 이미 진행된 상대방의 프로필을 볼 수 있다
• 물론 서버 내부에서 허가되지 않는 프로필 조회는 막긴 하지만(정말 너가 얘랑 매칭된게 맞는지 등등) 세션이나 jwt로 userId를 숨겨서 주는데 저렇게 심지의 타인의 userId가 노출 된다면 이는 의미가 퇴색된다고 생각했다
• 번호를 보고 서비스의 규모를 추정할 수 있어진다, 소개팅 서비스 특성상 유저 풀에 대해서 예민할 수 있다

실제로 잘못된 userId에 접근하여 슬랙 에러 알림으로 알림이 여러번 온 적 있었다 우리는 어떤 id를 요청했을때 에러가 발생하는지 로그를 띄웠는데 1, 2, 3, 4, 5, 6 이렇게 차근차근 위로 올리는 유저가 몇 있었다 ㄷㄷ

이를 해결하기 위해 userId를 암호화해서 보내거나 앞단에서 SPA(상태 관리 라이브러리(예: Redux, Vuex)를 사용해 URL에 민감한 정보를 포함시키지 않고도 페이지를 렌더링) 등 있겠지만 UUID를 입하기로 했다

(그리고 토스 페이먼츠 PG연동할때 필요하기도 해서 도입했다)

도입시 문제점

사실 UUID를 도입하는 것 자체는 너무 쉽다 그렇다고 마냥 도입할 수 없는게 여러 문제점이 있다

1. InnoDB의 클러스터링 인덱스를 효율적으로 사용하지 못한다

클러스터링 인덱스란 저장되는 순서가 PK값을 기준으로 정렬이 되는 것이다. 하지만 uuid는 랜덤 값이기 때문에, 클러스터형 인덱스를 조정 한다

InnoDB 의 테이블에서 순서를 보장하기 위해 기존의 레코드를 페이지 넘어서까지 이동 시켜서 정렬을 한다 레코드 양이 적으면 영향은 적을 수 있지만 레코드 양이 매우 많은 경우 PK 컬럼 데이터의 레코드의 이동은 매우 큰 성능적인 영향을 끼칠수도 있다

2. 세컨더리 인덱스의 크기가 늘어난다

세컨더리 인덱스는 프라이머리 키를 주소처럼 사용한다 uuid는 128비트이기 때문에 일반적으로 사용하는 정수형 pk보다 데이터가 크다

과연 진짜일까?

실제로 값을 넣어서 테스트 해보니 레코드가 100만개는 있어야 어느정도 성능적인 이점이 있다 하지만 어디서 슬로우 쿼리가 발생할 수 있을지 장담할 수 없고 이왕 하는거 미리미리 제대로 해보자

해결

UUID는 16 옥텟 (128비트)의 수이다. 표준 형식에서 UUID는 32개의 십육진수로 표현되며 총 36개 문자(32개 문자와 4개의 하이픈)로 된 8-4-4-4-12라는 5개의 그룹을 하이픈으로 구분한다.

Timestamp - Timestamp - Timestamp & Version - Variant & Clock sequence - Node id

사진 출처: https://ko.wikipedia.org/wiki/%EB%B2%94%EC%9A%A9_%EA%B3%A0%EC%9C%A0_%EC%8B%9D%EB%B3%84%EC%9E%90

5개의 버전이 존재하며 각 버전마다 UUID를 만드는 방식이 다르다

버전 1 (MAC 주소) : 현재시간과 MAC주소를 기반으로 생성 버전 2 (DCE 보안) : 현재시간과 MAC주소와 DCE Security version으로 생성 버전 3 (MD5 해시) : MD5해시를 기반으로 이름과 네임스페이스에 대한 조합으로 생성 버전 4 (랜덤) : 랜덤한 값으로 생성 버전 5 (SHA-1 해시) : 이름과, SHA-1 해싱으로 생성

버전3은 MD5알고리즘이 보안에 취약하고 네임스페이스에 의존적이기 때문에 버전5도 네임스페이스에 의존적 버전2는 Mac주소를 사용하기 때문에 보안에 취약하다

4는 가장 많이 쓰이지만 랜덤이기 때문에 "도입시 문제점"에 말한 문제가 발생한다 그럼 1버전을 사용해야 하나? 하지만 1버전은 Mac주소를 사용하기 때문에 보안에 취약하다

버전7도 있다 버전7은 시간 기반 랜덤 UUID로, 타임스탬프와 랜덤 데이터를 결합하여 생성된다. 이는 Version 1과 Version 4의 장점을 결합한 형태로 볼 수 있다고 한다

버전7은 UUID는 타임스탬프를 앞부분에 포함하고, 나머지 부분을 랜덤 데이터로 채운다 타임스탬프가 앞에 있기 때문에 시간 순서대로 UUID가 생성된다 그렇기 때문에 데이터베이스에서 시간에 따라 순차적인 UUID 값을 생성할 수 있다 이는 B-tree 인덱스에서 4버전 보다 성능이 좋아진다

그래서 버전 7을 도입하기로 했다

    implementation 'com.fasterxml.uuid:java-uuid-generator:4.1.0'

        UUID uuidV7 = Generators.timeBasedEpochGenerator().generate();

이렇게 생성할 수 있다, 그러면 이거를 매번 데이터를 삽입할 때마다 저런 코드를 작성해야 할까? 중복된 일을 제거해보자 우리는 JPA를 사용중인 ㅅ

    @Id
    @GeneratedValue(generator = "UUID")
    @GenericGenerator(
            name = "UUID",
            strategy = "your.directory.UUID7Generator" // 커스텀 UUID 7 생성기 사용
    )
    @Column(name = "id", updatable = false, nullable = false, columnDefinition = "BINARY(16)")
    private UUID id;

public class UUID7Generator implements IdentifierGenerator {

    @Override
    public Serializable generate(SharedSessionContractImplementor session, Object obj) throws HibernateException {
        // UUID Version 7 생성
        return Generators.timeBasedEpochGenerator().generate();
    }
}

코드는 Hibernate를 사용하여 UUID 버전 7을 데이터베이스에 저장할 때 커스텀 생성기를 만들었다 여기서는 UUID7Generator라는 클래스가 사용자 정의 UUID Version 7 생성기고, 엔티티 클래스의 id 필드를 위한 UUID를 생성한다

Generators.timeBasedEpochGenerator(): com.fasterxml.uuid 라이브러리의 메서드로, UUID 버전 7을 생성하는 데 사용한다 POSIX Epoch Time(1970년 1월 1일부터 경과한 시간)을 기반으로 하여 고유한 UUID를 생성한다. 이를 통해 시간이 흐름에 따라 순차적으로 증가하는 UUID가 생성된다

@GeneratedValue(generator = "UUID"): generator 속성으로 "UUID"라는 이름의 커스텀 생성기를 사용하도록 지정

@GenericGenerator: name: "UUID"로 설정된 생성기의 이름 @GeneratedValue(generator = "UUID")에서 이 이름을 참조하여 UUID를 생성하는 로직을 연결, strategy: "your.directory.UUID7Generator"는 UUID를 생성하는 커스텀 클래스(UUID7Generator)를 지정한다 이 클래스를 사용하여 Hibernate가 엔티티의 PK를 생성한다

결과, 테스트

처음에 문제 정의를 할 때 랜덤한 uuid4는 데이터가 삽입될 때마다 B-트리에 페이지단위를 넘어서는 인덱스 재정렬이 일어나기 때문에 시퀀스인 uuid7를 공부하고 적용한 것이다 과연 진짜일까?

uuid4, uuid7을 비교해보자

100만건 데이터 삽입 시간

uuid7

uuid4

정리하자면 uuid version 4는 14분 11초가 걸렸고, uuid version7은 13분 39초가 걸렸다 그리고 uuid 7버전은 시간 단위로 잘 정렬이 되는걸 볼 수 있지만 uuid4는 그렇지 않다는 것도 알 수 있다.

조회도 적용하고 싶지만 노트북 쿨링팬 돌아가는 소리가 들리면서 컴퓨터가 너무 힘들어 한다 여기까지만 해야겠다

참고: https://www.baeldung.com/java-generating-time-based-uuids https://ko.wikipedia.org/wiki/%EB%B2%94%EC%9A%A9_%EA%B3%A0%EC%9C%A0_%EC%8B%9D%EB%B3%84%EC%9E%90

채팅 기능을 구현하려 한다

우리 서버는 오토스케일링으로 최대 2대까지 ec2가 늘어날 수 있다, 로드밸런서로 부하 분산을 한다 잦은 일은 아니지만 실제로 유저들이 많이 몰리는 시간대에 2대까지 늘어난다

A와 B가 채팅을 하는 상황이라고 생각해보자 A클라이언트는 1번서버에 소켓 연결을 하고 B클라이언트는 2번 서버에 소켓 연결을 하면 A가 보낸 메시지가 1번 서버에만 도착하고 2번 서버에 있는 B에게 전달되지 않는다

2. 해결 방안

1. 스티키 세션 사용

로드 밸런서는 클라이언트의 브라우저에 고유한 식별자가 포함된 쿠키를 설정하고 이후 클라이언트가 해당 쿠키를 포함해 요청을 보내면, 로드 밸런서는 쿠키의 값을 기반으로 클라이언트가 처음 연결된 서버로 트래픽을 라우팅 하는 방식으로 구현하면 된다 NGINX에서 Sticky Cookie 기능을 통해 이를 구현할 수 있다

하지만 서버 간 부하 불균형: 스티키 세션을 사용하면 특정 서버에 특정 사용자의 모든 요청이 집중되므로, 트래픽이 고르게 분산되지 않을 수 있음 예를 들어, 일부 사용자가 과도한 트래픽을 발생시키면 해당 사용자가 연결된 서버에 과부하가 발생할 수 있다 이는 특히 서버의 수가 적을 때 더 심각해질 수 있다고 한다

스케일링의 비효율성: 만약 특정 서버에 많은 세션이 스티키 되어 있다면, 서버의 부하가 커질 때 새로운 서버를 추가해도 기존의 세션은 계속 기존 서버에 남아있기 때문에, 추가된 서버가 충분히 활용되지 못할 수 있다

서버의 재배포로 인해 서버가 재시작되면, 해당 서버에 저장된 모든 세션 데이터가 사라진다 그래서 redis나 RDB같이 외부 세션 저장소를 사용해야 한다.

2. 메시지 브로커 사용

pub/sub 방식으로 서로 직접 통신하는 것이 아닌 메시지 브로커를 통해 메시지를 전달하는 구조이다. 메시지 브로커는 Publisher가 보낸 메시지를 Subscriber로 전달해주는 미들웨어(중간 다리) 역할을 한다.

• Kafka 는 대량의 데이터를 저장하면서 높은 처리량이 필요에 적합한 메시지 브로커이다. 신뢰성 있는 메시지를 전송할 수 있다.
• Redis의 Pub/Sub은 간단하게 구현할 수 있지만 다른 메시지 브로커와 다르게 메시지 지속성이 없다. 즉, 메시지를 전송한 후 해당 메시지는 삭제되며 Redis에 저장되지 않는다. 또한, 메시지 전송 신뢰성을 보장하지 않기 때문에 단점을 보완할 별도의 추가 구현이 필요할 수 있다.

결론

일단 세션 방식은 결국엔 스케일링의 비효율성도 있고 배포같이 서버가 재시작 되었을 때를 대비해 redis나 RDB에 외부 세션 저장소를 사용해야 한다는 단점이 있다 나도 구현해보고 테스트 해봐야 알겠지만 일단 redis는 우리 인프라에 이미 elasticache로 구축이 되어있는 상황이다. 그러면 nginx설정이나 스케일링의 비효율성, 서버 간 부하 불균형을 겪을 바엔 redis pub/sub을 사용하는게 맞다는 생각이 들었다.

또한 redis에 채팅 내역이 저장되지 않는다는 문제점이 있다 이 문제에 대해서 생각해본 결론은

1. 어쩌피 채팅 내역은 MongoDB같은 곳이나 RDB에 저장해야되지 않나? 카카오톡도 푸시 알림이 와서 채팅방에 들어가면 1초동안 멈춰있다가 새로 온 채팅이 한꺼번에 보인다. 이게 카프카에서 웹소켓이 끊겨서 데이터를 저장하고 한꺼번에 보내는걸까? 아니면 MongoDB에서 조회를 하는걸까? 그렇다면 카프카에 partition에 채팅 내역을 저장되더라도 DB에 채팅 내역을 저장 안해도되나? 클라이언트에서 소켓 연결 disconnect가 되면 다시 재연결하고 구독도 다시하면 안될까?

2. 카프카를 사용한다면, Consumer의 수와 관계없이, 전송당 하나의 메시지만 발행될 것이다. 나중에 기획이 추가 될 때 다대다 미팅, 모임 관련된 기능도 추가를 고려하면 아직은 redis가 맞을 것 같다는 생각이 들었다. 하지만 이 부분도 물론 각 서버 인스턴스마다 다른 group ID를 부여하여 사용할 수도 있다. 다대다 미팅 기능이 생겼을 때는 이미 유저가 많을 상황일 것 같은데 그땐 카프카로 전환하려나?

여러 자문과 구글링으로 알아봤지만 명확한 판단이 서질 않았다. 우리 서비스의 채팅 기능은 만남 전날 몇시간동안 약속 리마인드 용도 그 이상도 이하도 아니다. 채팅기능은 서비스의 메인 기능이 아니기 떄문에 대량의 데이터, 높은 처리량과는 맞지 않는다. 물론 나중에 유저가 엄청나게 많이 생기면 그땐 필요하다. 그럼 그때 행복하게 카프카를 도입하기로 하자

해결, 구현

웹소켓의 특징이나 이론적인 내용은 생략하자(검색하면 너무 쉽게 나오니까...ㅎ)

STOMP라고 이번에 처음 알았다 웹소켓만을 사용해서 채팅 서버를 구현하다면, 메시지 포맷 형식이나 메시지 통신 과정 등을 관리해야 하는 번거로움이 있다. 따라서 이러한 관리를 대신해줄 수 있는 STOMP 프로토콜을 서브 프로토콜로 사용할 수 있다.

STOMP(Simple Text Oriented Messaging Protocol)는 메시지의 형식이나 내용 등을 정의하여 메시징 전송을 효율적으로 도와주는 프로토콜이다. 기본적으로 Pub/Sub 구조로 되어있어 STOMP 규칙에 맞춰 메시징 처리를 재정의하여 사용하면 된다

채팅을 웹소켓과 STOMP 방식으로만 구현할 경우, 두 클라이언트가 같은 서버에 pub/sub을 해야 메세지를 주고 받을 수 있다 그래서 구독 대상(Topic) 을 여러 서버에서 접근이 가능하게 하기 위해 위에 해결방안에서 말했듯이 redis를 메시지 브로커로 쓰자

구현 코드는 나중에 보고 이해부터 하려면

이런 구조와 작동 방식이다

디비스펙은 여기에 올리지 못하니 간단하게 설명하면 채팅방 채팅내역이 일대다 방식으로 설계했다

1. A와 B는 1대1 채팅을 시작한다 채팅방을 만들면서 채팅방 레코드, 채팅방 pk ex)UUID1234 채널을 구독 한다
2. 서버는 클라이언트에게 채팅방 uuid와 함께 내가 참여중인 채팅방 리스트를 반환해준다
3. A가 채팅방을 접속하면 웹 소켓을 연결하고 해당 채팅방 UUID1234 기반으로 만들었던 채널 구독한다
4. B가 채팅방을 접속하면 웹 소켓을 연결하고 해당 채팁방 UUID1234 기반으로 만들었던 채널 구독한다
5. A가 채팅방에 메시지를 보내면 redis에 메시지를 UUID1234 토픽에 pub하고 메시지 내역을 저장한다
6. redis는 메시지를 받고 그 메시지를 같은 토픽(UUID1234) 구독된 사람에게 pub한다 누구? B, 왜와이? 4번에서 구독했잖아
7. 그 메시지를 받은 서버는 다시 클라이언트는 웹소켓으로 메시지를 전송한다

간단하게 이런 과정이다 물론 A와 B가 동시에 채팅방에 접속했다는 해피케이스고 A가 채팅방에 접속해있다고 B가 접속했을거라는 보장은 없다 이건 푸시알림같은걸 사용해서 알려줄 수 있다(채팅방에 접속해있으면 푸시알림을 보내면 안되겠지? 이건 일단 구현하고 생각하자)

이제 코드로 가보자

    // web socket
    implementation 'org.springframework.boot:spring-boot-starter-websocket'
    implementation 'org.webjars:stomp-websocket:2.3.4'

@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    private final StompInterceptor stompInterceptor;
    private final StompExceptionHandler stompExceptionHandler;

    /**
     * stompInterceptor가 token 을 체크할 수 있도록 인터셉터 설정 소켓은 Webconfig
     * WebSocket 통신에서는 HTTP 요청과 다르게 동작하기 때문에, WebMvcConfigurer에 설정된 인터셉터가 적용되지 않습니다
     */
    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(stompInterceptor);
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry){
        // stomp 접속 주소 url => /ws-stomp
        registry
                .setErrorHandler(stompExceptionHandler)
                .addEndpoint("/ws-stomp")
                .setAllowedOrigins("*");
    }
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry){
        // 메시지를 구독하는 요청 url => 즉 메시지 받을 때
        registry.enableSimpleBroker("/sub");

        // 메시지를 발행하는 요청 url => 즉 메시지 보낼 때
        registry.setApplicationDestinationPrefixes("/pub");
    }

}

• @EnableWebSocketMessageBroker: 이 어노테이션은 WebSocket 메시징을 활성화하고 STOMP 프로토콜을 사용한 메시지 브로커를 구성할 수 있게 해준다 즉, WebSocket 서버를 설정하고 메시지 브로커를 통해 클라이언트 간 통신을 가능하게 만든다\

• WebSocketMessageBrokerConfigurer: 이 인터페이스는 WebSocket 메시지 브로커의 구성을 정의할 수 있는 메소드를 제공한다 Spring에서 WebSocket을 설정하기 위해 이 인터페이스를 구현함

• configureClientInboundChannel(): 이 메소드는 클라이언트로부터 들어오는 메시지를 처리하는 채널을 설정 여기서는 stompInterceptor라는 인터셉터를 설정하여 들어오는 WebSocket 메시지에 대한 추가적인 처리, 즉 토큰 검증과 같은 작업을 수행할 수 있도록 설정한다.

• stompInterceptor: 우리는 이 인터셉터는 STOMP 메시지에서 클라이언트의 토큰을 검증하게 했다 추후에 코드로 설명 WebSocket 통신에서는 HTTP와 달리 헤더 정보가 제한되기 때문에, WebSocket 메시지의 페이로드에 포함된 토큰을 읽어 인증해야함

• registerStompEndpoints(): 이 메소드는 STOMP 프로토콜을 사용하기 위한 엔드포인트를 설정하는 곳

• addEndpoint("/ws-stomp"): 클라이언트는 이 엔드포인트(/ws-stomp)를 통해 WebSocket 서버에 연결할 수 있음

• setAllowedOrigins("*"): 이 메소드는 CORS 설정으로, 모든 도메인에서 WebSocket에 접속할 수 있게 허용하는 설정. 보안상의 이유로 실제 운영에서는 * 대신 허용된 도메인만 명시할 예정

• setErrorHandler(stompExceptionHandler): 메시지 처리 중 예외가 발생할 때 이를 처리하는 핸들러 이 핸들러는 WebSocket 연결 도중 발생하는 오류를 처리하고, 클라이언트에 적절한 에러 메시지를 전달하는 역할을 함

• configureMessageBroker(): 이 메소드는 실제로 메시지를 송신하고 수신하는 브로커를 설정하는 곳

• enableSimpleBroker("/sub"): 이 부분은 간단한 메시지 브로커를 활성화 /sub로 시작하는 URL을 클라이언트가 구독 가능 즉, 클라이언트가 /sub로 시작하는 경로에 연결하면 서버에서 발행된 메시지를 실시간으로 받을 수 있다 ex) 클라이언트는 /sub~~ 을 구독하면, 서버가 /sub/chat 경로로 메시지를 발행할 때 이를 받을 수 있다

• setApplicationDestinationPrefixes("/pub"): 클라이언트가 메시지를 서버로 발행할 때 사용하는 URL 프리픽스를 설정. /pub로 시작하는 URL은 클라이언트가 서버에 메시지를 보낼 때 사용된다. ex) 클라이언트는 /pub~~ 경로로 메시지를 전송하면, 이 메시지가 서버에서 처리됩니다.

@Configuration
public class RedisConfig {

    @Value("${spring.redis.host}")
    private String redisHost;

    @Value("${spring.redis.port}")
    private String redisPort;

    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
        redisStandaloneConfiguration.setHostName(redisHost);
        redisStandaloneConfiguration.setPort(Integer.parseInt(redisPort));
        return new LettuceConnectionFactory(redisStandaloneConfiguration);
    }

    @Bean
    public RedisTemplate<Long, String> redisTemplate() {
        RedisTemplate<Long, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory());
        redisTemplate.setKeySerializer(new GenericToStringSerializer<>(Long.class));
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        return redisTemplate;
    }

    @Bean
    public StringRedisTemplate stringRedisTemplate(){
        StringRedisTemplate stringRedisTemplate= new StringRedisTemplate();
        stringRedisTemplate.setKeySerializer(new StringRedisSerializer());
        stringRedisTemplate.setValueSerializer(new StringRedisSerializer());
        stringRedisTemplate.setConnectionFactory(redisConnectionFactory());
        return stringRedisTemplate;
    }

    // redis 에 발행(publish)된 메시지 처리를 위한 리스너 설정
    @Bean
    public RedisMessageListenerContainer redisMessageListener(
            RedisConnectionFactory connectionFactory
    ) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }


    // 실제 메시지를 처리하는 subscriber 설정 추가
    @Bean
    public MessageListenerAdapter listenerAdapter(RedisSubscriber subscriber) { // (2)
        return new MessageListenerAdapter(subscriber, "sendMessage");
    }
}

• RedisMessageListenerContainer: Redis의 발행/구독(pub/sub) 기능을 사용하는 메시지 리스너 Redis에서 발행된 메시지를 처리할 수 있는 리스너 컨테이너

• setConnectionFactory(): 리스너 컨테이너가 Redis와 통신할 때 사용할 연결 팩토리를 지정 Redis에 연결된 후, 메시지가 발행되면 해당 메시지를 구독하고 처리하는 작업을 수행

• MessageListenerAdapter: Redis에서 발행된 메시지를 처리할 때 사용하는 어댑터 이 어댑터는 RedisSubscriber 클래스의 메소드와 연결되어, Redis에서 메시지를 구독할 때 그 메시지를 처리할 방법을 정의

• subscriber: RedisSubscriber는 실제로 메시지를 처리하는 클래스이고 직접 구현한 클래스이다 메시지를 받으면 sendMessage() 메소드를 호출하여 처리한다

• sendMessage: 여기서 명시한 "sendMessage"는 RedisSubscriber 클래스 내의 메소드를 가리킨다 추후 코드에 잇음

결론은 메시지가 발행되면 RedisMessageListenerContainer가 이를 구독하고, MessageListenerAdapter를 통해 RedisSubscriber의 sendMessage() 메소드로 메시지를 전달한다 sendMessage() 메소드에서는 메시지를 처리하고, 클라이언트에게 해당 메시지를 전달한다

@RequiredArgsConstructor
@Service
@Slf4j
public class RedisSubscriber {

    private final ObjectMapper objectMapper;
    private final SimpMessageSendingOperations messagingTemplate;

    /**
     * Redis에서 메시지가 발행(publish)되면
     * 대기하고 있던 Redis Subscriber가 해당 메시지를 받아 처리한다.
     */
    public void sendMessage(String jsonMessage) {
        try {

            PublishChattingMessage chattingMessage = objectMapper.readValue(jsonMessage, PublishChattingMessage.class);
            SendChattingMessageWSRequest chattingWsRequest = chattingMessage.toWSRequest();

            log.warn("Redis Subcriber chatMSG : {}", chattingMessage.getMessage());

            // 채팅방을 구독한 클라이언트에게 메시지 발송
            messagingTemplate.convertAndSend(
                    "/sub/chatting/room/chatroom-" + chattingMessage.getRoomId(), chattingWsRequest
            );

        } catch (Exception e) {
            log.error("Exception {}", e);
        }
    }
}

• messagingTemplate.convertAndSend(): 이 메소드는 WebSocket을 통해 메시지를 클라이언트로 전송한다 WebSocket을 구독 중인 클라이언트들이 메시지를 받을 수 있도록 설정된 채널로 메시지를 메시지를 전달함 구독 경로는 /sub/chatting/room/chatroom-{roomId}, 특정 채팅방을 구독 중인 클라이언트들에게 해당 메시지를 전달한다.
• 왜 직렬화/역직렬화를 해서 주고받을까? 직렬화와 역직렬화를 사용하는 이유는 시스템 간에 데이터를 주고받을 때 효율적이고 표준화된 형식으로 데이터를 전달하기 위함이다 특히 분산 시스템이나 실시간 시스템(예: Redis Pub/Sub)에서 다양한 애플리케이션이 데이터를 주고받을 때, 직렬화는 필수적인 과정이다 ex) 자바로 작성된 애플리케이션에서 Redis에 메시지를 발행한 후, 다른 프로그래밍 언어(Python, JavaScript 등)로 작성된 애플리케이션에서 이 데이터를 구독할 수 있음 또한 Redis와 같은 시스템은 바이트 배열 또는 문자열을 기반으로 데이터를 처리한다 자바에서 사용되는 객체들은 매우 복잡한 상태를 가지고 있기 때문에, Redis와 같은 시스템이 직접 자바 객체를 처리가 힘들다 이 때문에 자바 객체를 직렬화해서 Redis에 전송하고, 다시 역직렬화하여 자바 객체로 변환해야 한다

@RequiredArgsConstructor
@Component
public class StompInterceptor implements ChannelInterceptor {

    private final AuthService authService;
    private final UserChattingSessionManager userChattingSessionManager;
    private final ChattingValidator chattingValidator;
    private final ChattingRoomReader chattingRoomReader;
    private final UserReader userReader;

    private static final String SUB_PREFIX = "/sub/chatting/room/chatroom-";

    /**
     * websocket을 통해 들어온 요청이 처리 되기전 실행된다.
     */
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
        //웹 소켓 연결할 때, 세션 생성 -> 그냥 jwt 검증만
        if (StompCommand.CONNECT.equals(accessor.getCommand())) {
            UUID userId = authenticate(authService.getAuthorizationToken(
                    accessor.getFirstNativeHeader("Authorization")));

            return message;
        }

        // 구독할때 -> jwt검증하고 이 유저가 해당 채팅방을 구독할 수 있는건지 검증, 채널 구독과 세션 관리는 추후 서비스 로직에서 관
        if (StompCommand.SUBSCRIBE.equals(accessor.getCommand())) {
            String destination = accessor.getDestination();

            UUID roomId = UUID.fromString((destination.substring(SUB_PREFIX.length() + 1)));
            UUID userId = authenticate(authService.getAuthorizationToken(
                    accessor.getFirstNativeHeader("Authorization")));
            ChattingRoom chattingRoom = chattingRoomReader.readById(roomId);
            User user = userReader.readById(userId);
            chattingValidator.validateChattingRoomAndUserExist(user, chattingRoom);
        }

        // DISCONNECT 요청 처리 세션 삭제
        if (StompCommand.DISCONNECT.equals(accessor.getCommand())) {
            UUID userId = authenticate(authService.getAuthorizationToken(
                    accessor.getFirstNativeHeader("Authorization")));
            // Redis에서 상태 제거
            userChattingSessionManager.removeOnlineStatus(userId);
        }

        if (StompCommand.SEND.equals(accessor.getCommand())) {
            UUID userId = authenticate(authService.getAuthorizationToken(accessor.getFirstNativeHeader("Authorization")));
//        accessor.setNativeHeader("userId", userId.toString());
            Objects.requireNonNull(accessor.getSessionAttributes()).put("userId", userId);
            return message;
        }
        return message;
    }

    private UUID authenticate(String token) {
        authService.isValidToken(token);  // 토큰 유효성 검사
        return extractUserIdFromJwtToken(token);
    }

    private UUID extractUserIdFromJwtToken(final String token) {
        // JWT 토큰을 파싱하여 Claims 객체를 얻음
        HashMap<String, Object> parseJwtTokenMap = authService.parseJwtToken(token);
        Claims claims = (Claims) parseJwtTokenMap.get("claims");

        // 토큰 타입이 "NORMAL"인지 확인
        validateIsTokenTypeNormal(claims);

        // 토큰에서 사용자 ID를 추출
        return getUserIdFromToken(claims);
    }

    private void validateIsTokenTypeNormal(Claims claims) {
        if (!authService.checkTokenType(claims, UserType.NORMAL)) {
            throw new AuthenticationException(ErrorCode.INVALID_USER_TYPE, ErrorCode.INVALID_USER_TYPE.getStatusMessage());
        }
    }

    private UUID getUserIdFromToken(final Claims claims) {
        Object userId = claims.get("userId");
        String userIdString = userId.toString();
        // UUID 형식인지 확인
        try {
            UUID.fromString(userIdString);
        } catch (IllegalArgumentException e) {
            throw new InvalidJWTUserIdException(
                    ErrorCode.INVALID_JWT_USER_ID_ERROR,
                    ErrorCode.INVALID_JWT_USER_ID_ERROR.getStatusMessage()
            );
        }
        return UUID.fromString(userIdString);
    }
}

• preSend(): WebSocket을 통해 들어오는 메시지 요청이 처리되기 전에 실행된다 이 메소드는 WebSocket 요청의 헤더를 분석하고, JWT 토큰을 인증한 후 사용자의 userId를 세션에 저장하는 역할을 한다

• StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message): STOMP 메시지의 헤더 정보를 쉽게 다룰 수 있도록 StompHeaderAccessor로 래핑한다 이를 통해 메시지의 다양한 헤더 값과 STOMP 명령을 처리할 수 있다

• 나머지는 JWT 검증, userId 추출

• Objects.requireNonNull(accessor.getSessionAttributes()).put("userId", userId): 검증된 userId를 WebSocket 세션에 저장한다. 이 userId는 이후 메시지 처리 과정에서 메시지를 보낸 사용자를 추적하거나 인증된 사용자임을 확인하는 데 사용된다

• 만약에 COMMAND가 SUBSCRIBE(구독)일 경우에 구독하는 사람이 과연 이 채팅방을 접속할 수 잇는지 검증을 한번 때린다 그리고 채널을 구독하기 위해 controller로 넘긴다 (코드는 아래에 있음)

• COMMAND가 SEND일 경우에 jwt만 검증하고(왜? 아까 SUBSCRIBE에서 각종 validation을 했으니까) controller 로 보낸다

• CONNECTION일 경우에도 일단 JWT만 검증한다

• DISCONNECT일 경우에는 jwt를 검증하고 세션을 삭제한다

@RequiredArgsConstructor
@Controller
@Slf4j
public class ChattingWSController {

    private final ChattingService chattingService;

    /**
     * websocket "/pub/chatting/message"로 들어오는 메시징을 처리한다.
     */
    @MessageMapping("/chatting/message")
    public ResponseEntity<HttpStatus> message(ChattingMessageWSRequest message,
                                              @Header("simpSessionAttributes") Map<String, Object> sessionAttributes) {
        UUID sendUserId = (UUID) sessionAttributes.get("userId");
        chattingService.sendChattingMessage(message, sendUserId); //RedisPublisher 호출;
        return ResponseEntity.ok().build();
    }

    @MessageMapping("/chatting/room/{roomId}")
    public void handleRoomSubscription(@DestinationVariable UUID roomId,
                                       @Header("simpSessionAttributes") Map<String, Object> sessionAttributes) {
        UUID userId = (UUID) sessionAttributes.get("userId");
        chattingService.enterChattingRoom(roomId, userId);
        log.info("User {} subscribed to room {}", userId, roomId);
    }

}

• @MessageMapping: WebSocket에서 특정 경로로 들어오는 메시지를 처리하는 어노테이션이다 여기서는 클라이언트가 /pub/chatting/message 경로로 WebSocket을 통해 메시지를 보내면, 이 메소드가 호출된다 이 경로는 클라이언트가 메시지를 서버로 발행할 때 사용되는 경로이다 STOMP 프로토콜을 사용하여 메시지를 전송할 때, 클라이언트는 이 경로에 메시지를 발송하게 된다

• @Header("simpSessionAttributes") Map<String, Object> sessionAttributes: WebSocket 세션에서 전송된 세션 정보를 가져오는 어노테이션 이다. 여기서 simpSessionAttributes라는 헤더에서 세션 정보를 가져오며, 이는 세션에 저장된 사용자 정보를 포함한다 원래는 세션에 담지 않고 @Header("userId")로 하려 했는데 컨트롤러까지 페이로드가 넘어오지 않아서 일단 이렇게 구현했다 ㅠ (개선 사항 1순위)

• 근데 왜 url이 pub/chatting/message가 아닐까??: 이전에 WebSocketConfig에서 setApplicationDestinationPrefixes("/pub")을 사용했었다 클라이언트가 메시지를 서버로 보낼 때 사용할 경로에 /pub이라는 프리픽스를 추가한다는 건데 클라이언트는 실제로 /pub/chatting/message 경로로 메시지를 보내지만, 서버에서는 @MessageMapping("/chatting/message")와 매칭된다 이때 @MessageMapping 경로는 프리픽스 없이 설정한다 즉, /pub은 메시지를 발행할 때 자동으로 붙는 경로의 프리픽스일 뿐, 실제로 서버에서는 이를 고려하지 않고 핸들러에서 처리할 경로만 지정하는 것이다. 그래서 컨트롤러 메소드에서는 /pub이 포함되지 않은 경로를 설정한다

@Service
@RequiredArgsConstructor
@Slf4j
public class ChattingService {

    private final RedisTemplate<String, String> redisTemplate;
    private final ObjectMapper objectMapper;
    private final UserReader userReader;
    private final ChattingValidator chattingValidator;
    private final ChattingContentWriter chattingContentWriter;
    private final RedisMessageListenerContainer redisMessageListenerContainer;
    private final MessageListenerAdapter messageListenerAdapter;
    private final ChattingRoomReader chattingRoomReader;
    private final ChattingRoomWriter chattingRoomWriter;
    private final ChattingContentReader chattingContentReader;

    public void sendChattingMessage(ChattingMessageWSRequest request, UUID sendUserId) {
        User user = userReader.readById(sendUserId);
        ChattingRoom chattingRoom = chattingValidator.validateChattingRoomAndUserExist(user, request.getRoomId());
        ChattingContent chattingContent = chattingContentWriter.write(createChattingContent(user, request.getRoomId(), request.getMessage()));
        publishMessage(chattingContent);
    }

    private void publishMessage(ChattingContent chattingContent) {
        PublishChattingMessage chattingMessage = PublishChattingMessage.builder()
                .roomId(chattingContent.getChattingRoomId())
                .sendAt(chattingContent.getCreatedDate())
                .message(chattingContent.getContent())
                .sendUserName(chattingContent.getSendUserName())
                .build();
        try {
            // ChatMessageDto를 JSON 문자열로 변환
            String messageJson = objectMapper.writeValueAsString(chattingMessage);
            String topic = "chatroom-" + chattingContent.getChattingRoomId().toString();
            // Redis에 JSON 문자열을 발행
            redisTemplate.convertAndSend(topic, messageJson);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    private ChattingContent createChattingContent(User user, UUID roomId, String message) {
        return ChattingContent.builder()
                .chattingRoomId(roomId)
                .sendUserName(user.getUserProfile().getFullName())
                .content(message)
                .build();
    }
    public void subscribeToTopic(UUID roomId) {
        ChattingRoom chattingRoom = chattingRoomReader.readById(roomId);
        ChannelTopic topic = new ChannelTopic("chatroom-" + chattingRoom.getId());
        redisMessageListenerContainer.addMessageListener(messageListenerAdapter, topic);
    }

    public void unsubscribeFromTopic(UUID roomId) {
        ChattingRoom chattingRoom = chattingRoomReader.readById(roomId);
        chattingRoomWriter.delete(chattingRoom);
        ChannelTopic topic = new ChannelTopic("chatroom-" + chattingRoom.getId());
        redisMessageListenerContainer.removeMessageListener(messageListenerAdapter, topic);
    }
}

• validateChattingRoomAndUserExist: 채팅을 보낸 유저가 요청한 채팅방 id와 실제 그 유저가 그 채팅방에 존재하는지 검증 해준다.
• 채팅 내역을 저장한다
• publishMessage: 채팅 내역을 발행한다 채팅방 생성할때 topic을 채팅방의 uuid로 설정했기 때문에 (설정한부분은 subscribeToTopic 메서드)
• 그리고 아까 말했듯이 json으로 변환하고 redis에 발행하면 된다

참고 자료: https://redis.io/docs/latest/develop/interact/pubsub/ https://medium.com/frientrip/pub-sub-%EC%9E%98-%EC%95%8C%EA%B3%A0-%EC%93%B0%EC%9E%90-de9dc1b9f739 https://nebulaisme.tistory.com/147 https://docs.spring.io/spring-framework/reference/web/websocket/stomp.html

인가: https://velog.io/@wellbeing-dough/spring-interceptor-JWT-Authentication-%EB%A6%AC%ED%8C%A9%ED%84%B0%EB%A7%81

그런데 비관적 락 -> redisson 분산락으로 구현하고 나니 (https://velog.io/@wellbeing-dough/spring-interceptor-JWT-Authentication-%EB%A6%AC%ED%8C%A9%ED%84%B0%EB%A7%81) aop 사용할 수 있을 것 같다는 생각이 들었다 그래서 이번에 확실하게 aop에 대해서 알아보고 적용해 봐보자

AOP란

AOP는 Aspect Orient Programming의 약자로 관점 지향 프로그래밍이라고 불린다 각 계층의 로직마다 핵심적인 관점(비즈니스 로직), 부가적인 관점(핵심 로직을 실행하기 위해서 행해지는 부가적인 로직)으로 나눠서 그 관점을 기준으로 모듈화 하는 것이다.

결국엔 비즈니스 로직에서 계속 반복해서 쓰는 코드들을 흩어진 관심사라고 부르고 그것들을 Aspect로 모듈화 하고 분리하여 재사용하는 것이 AOP의 목적이다

1. AOP의 개념 용어

• Aspect: 흩어진 관심사를 모듈화 한 것
• Target: Advice가 적용될 객체
• Advice: 프록시가 호출하는 실제 부가기능을 담은 구현체 (횡단 기능)
• JointPoint: Advice가 적용될 위치, 끼어들 지점, 메서드 진입 지점, 생성자 호출 시점, 필드에서 값을 꺼내올 때 다양한 시점에 적용 가능
• PointCut: JointPoint의 상세한 스펙을 정의한 것

2. AOP의 원리

1. 리플렉션

리플렉션은 런타임에 클래스와 메서드의 메타 정보를 사용해 어플리케이션을 동적으로 유연하게 만드는 기술이다, 런타임에 클래스, 인터페이스, 메서드, 필드 등의 정보를 통해 컴파일 시간에 알 수 없는 클래스나 객체에 대해 동적으로 접근할 수 있다. 또한 클래스 이름을 잉용해 런타임에 동적으로 객체를 생성할 수 있으며 런타임에 메서드 이름을 이용해 메서드를 호출할 수 있다 또한 클래스의필드 값을 런타임에 읽거나 수정할 수 있다

2. JDK 동적 프록시

동적 프록시란 런타임에 인터페이스를 구현해주는 클래스의 프록시 객체를 생성할 수 있다 이를 통해 실제 객체 대신 프록시 객체를 사용하여 메서드를 호출하고 해당 메서드 호출을 가로채어 다양한 로직을 추가할 수 있다 @Transactional이 예시이다

스프링에서 리플렉션 기반으로 동적 프록시를 사용하면 런타임 시 개발자를 대신하여 프록시를 생성해주고 다양한 동작을 할 수 있다

원래 A가 프록시를 통해 BImpl을 호출한다면 동적 프록시를 사용하면 InvocationHandler를 사용하여 BImpl을 호출한다 프록시 객체는 newProxyInstance을 사용하여 생성한다

자세하게 까보면 타킷의 인터페이스를 자체적인 검증 로직을 통해 ProxyFactory에 의해 타깃의 인터페이스를 상속한 Proxy객체를 생성하고 Proxy객체에 InvocationHandler를 포함시켜 하나의 객체로 반환한다 이렇게 핵심부분은 인터페이스를 기준으로 Proxy객체를 생성해준다는 것이다. 그래서 구현체는 무조건 인터페이스를 상속받아야 한다

3. CGLIB(Code Generator Library)

CGLIB은 프록시에서 지원하지 않는 구체 클래스만을 기반으로 프록시를 생성할 수 있게 설계되었다 클래스의 바이트 코드를 조작하여 Proxy객체를 생성해준다 CGLIB는 Enhancer라는 클래스를 통해 Proxy를 생성한다

Enhancer는 타깃의 클래스를 상속받아 그 클래스에 포함된 모든 메서드를 재정의하여 Proxy를 생성한다 재정의 때문에 Final 메서드나 클래스에 대해서 재정의를 할 수 없으므로 Proxy도 생성할 수 없다는 단점이 있다 하지만 바이트 코드로 Proxy를 생성해 주기 때문에 성능이 좋다

Enhandler로 동적 프록시 객체를 생성하고 MethodInterceptor로 메서드 호출을 가로채서 추가 로직을 실행 Callback으로 프록시 객체의 메서드 호출을 가로챔

4. JDK 동적 프록시 vs CGLIB

가끔 다른 사람들의 스프링 코드를 보다보면 Service, ServiceImpl로 구현 한 사람들을 많이 볼 수 있었다 인강을 보더라도 많았었다 그게 궁금해서 옛날에 찾아봤었는데

1. 인터페이스와 구현 클래스를 분리할 수 있다 -> 비즈니스 로직과 구현 로직을 분리하여 독립적으로 개발 가능, 구현 변경은 인터페이스만 수정
2. IoC 기능과 함께 사용 가능 -> Service인터페이스를 빈으로 등록하고 ServiceImpl에서 의존성을 주입받아서 사용
3. 과거의 관습이 이어져왔다 -> 이부분에 대해서 잘 몰랐다

알고보니 2.0 Spring에서는 AOP Proxy를 구현할 때 JDK 동적 프록시를 사용해서 인터페이스를 구현한 객체에 대해서만 AOP Proxy를 생성할 수 있었다고 한다 그런데 어느 시점부터 인터페이스를 구현하지 않은 클래스에 대해서도 AOP Proxy를 생성할 수 있는 CGLIB가 구체 클래스만을 기반으로 AOP구현이 가능해 졌다고 한다 https://docs.spring.io/spring-framework/reference/core/aop/introduction-proxies.html

5. 그러면 스프링에서는 어떻게 프록시를...?

ProxyFactory을 사용하여 직접 프록시를 정의할 필요 없이 프록시 기술에 따라서 자동으로 생성된다고 한다

위에 있듯이 JDK 동적 프록시에서 제공하는 InvocationHandler와 CGLIB에서 제공하는 MethodInterceptor를 중복으로 만들지 않기 위해 Advice라는 개념을 도입하여 Advice만을 정의하고 스프링 프록시에 추가하기만 하면 내부에서 알아서 해준다

3. 활용

기본적인 개념을 알아봤으니 적용해보자 redisson 분산락에 대해서는 https://velog.io/@wellbeing-dough/redisson-%EC%9D%84-%EC%82%AC%EC%9A%A9%ED%95%9C-%EB%B6%84%EC%82%B0-%EB%9D%BD 여기에 있으니 여기서는 AOP의 활용에 초점을 맞춰보자

예전에

    public void acceptApplyEventPublish(Long studyPostId) {

        studyPostWriter.updateStudyPostApply(studyPostId);
        RLock lock = redissonClient.getLock(studyPostId.toString());
        boolean available = false;
        try {
            available = lock.tryLock(10, 1, TimeUnit.SECONDS);
            if (!available) {
                throw new StudyApplyLockAcquisitionException();
            }
            studyPostWriter.updateStudyPostApply(studyPostId);
        } catch (InterruptedException e) {
            throw new StudyApplyLockAcquisitionException();
        } finally {
            if (available) {
                lock.unlock();
            }
        }

예전에 사용했던 스터디 게시글 잔여석 감소 로직이다

이제 여기에 AOP를 적용해 보자

package kr.co.studyhubinu.studyhubserver.common.redisson;

import kr.co.studyhubinu.studyhubserver.exception.apply.StudyApplyLockAcquisitionException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;

@Aspect
@Slf4j
@RequiredArgsConstructor
@Component
public class RedissonDistributedLockAop {

    private final RedissonClient redissonClient;

    @Around("@annotation(kr.co.studyhubinu.studyhubserver.common.redisson.RedissonDistributedLock)")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        RedissonDistributedLock redissonDistributedLock = method.getAnnotation(RedissonDistributedLock.class);
        String hashKey = getDynamicValue(signature, joinPoint, redissonDistributedLock.hashKey());
        String field = getDynamicValue(signature, joinPoint, redissonDistributedLock.field());

        RLock lock = redissonClient.getLock(hashKey + ":" + field);

        Object result;
        boolean available = false;
        try {
            available = lock.tryLock(10, 1, TimeUnit.SECONDS);
            if (!available) {
                log.warn("Redisson GetLock Timeout {}", field);
                throw new StudyApplyLockAcquisitionException();
            }

            result = joinPoint.proceed();
        } catch (InterruptedException e) {
            throw new StudyApplyLockAcquisitionException();
        } finally {
            if (available) {
                lock.unlock();
            }
        }
        return result;
    }

    // 메서드 파라미터(field와 hashkey)를 기반으로 동적으로 값을 지정
    public String getDynamicValue(MethodSignature signature, ProceedingJoinPoint joinPoint, String distributedLock) {
        return CustomSpringELParser.getDynamicValue(
                signature.getParameterNames(),
                joinPoint.getArgs(),
                distributedLock);
    }
}

@Aspect: 이 클래스가 Aspect임을 나타낸다. Aspect는 특정 조인 포인트(프로그램 실행의 특정 지점)에 적용되는 Advice(횡단 관심사)를 포함합니다.

@Around: 이 메서드가 around advice임을 나타낸다 advice는 어노테이션이 붙은 메서드의 실행 전, 후에 실행된다 그리고 이 어드바이스는 REdissonDistributedLock 어노테이션에 붙은 메서드에 적용된다

MethodSignature: 조인 포인트의 메서드 시그니처를 가져온다 조인 포인트는 프로그램 실행의 특정 지점 ex) 메서드 호출, 객체 생성 을 말한다 Method method: 실제 실행 중인 메서드를 가져온다 자바 리플렉션의 API일부이며, 해당 객체를 통해 메서드의 이름, 반환 타입, 파라미터 타입, 어노테이션 등 다양한 정보를 가져올 수 있다 RedissonDistributedLock redissonDistributedLock: 메서드에서 RedissonDistributedLock 어노테이션 인스턴스를 가져온다

result = joinPoint.proceed(); 락을 획득한 후, 원래 메서드를 실행한다

getDynamicValue: 메서드 시그니처와 조인 포인트의 인자를 사용하여 동적으로 값을 계산한다

@Slf4j
@NoArgsConstructor
public class CustomSpringELParser {
    public static String getDynamicValue(String[] parameterNames, Object[] args, String key) {
        SpelExpressionParser parser = new SpelExpressionParser();
        StandardEvaluationContext context = new StandardEvaluationContext();
        for (int i = 0; i < parameterNames.length; i++) {
            context.setVariable(parameterNames[i], args[i]);
        }
        Object value = parser.parseExpression(key).getValue(context, Object.class);
        if (value == null) {
            log.warn("CustomSpringELParser evaluated value is null for key={}", key);
            return null;
        }
        return value.toString();
    }
}

이 클래스는 SpEL을 사용하여 메서드 파라미터를 기반으로 동적으로 값을 계산하는 클래스 이다 Lock의 이름을 자유롭게 전달할 수 있다 SpelExpressionParser: SpEL 표현식을 파싱하는데 사용됨 StandardEvaluationContext: 표현식을 평가할 때 사용할 컨텍스트 parseExpression: 주어진 표현식을 파싱하여 SpEL표현식 객체를 생성 getValue: 주어진 컨텍스트에서 표현식을 평가하여 결과 값을 반환한다 이 메서드는 메서드 파라미터 이름과 값을 컨텍스트에 설정하고, 주어진 SpEL 표현식을 평가하여 결과 값을 반환 예를들어 hashKey: 'apply', field: #studyId라고 하면

hashKey 계산:

• parameterNames: ["studyPostId"]
• args: [123L]
• key: "'apply'"
• context.setVariable("studyPostId", 123L)를 통해 변수를 설정 이 단계는 파서가 값을 평가할 때 필요한 변수 설정이다
• parser.parseExpression("'apply'").getValue(context, Object.class)을 통해 리터럴 문자열 "apply" 값을 평가
• hashKey는 "apply"

field 계산:

• parameterNames: ["studyPostId"]
• args: [123L]
• key: "#studyPostId"
• context.setVariable("studyPostId", 123L)를 통해 변수를 설정
• parser.parseExpression("#studyPostId").getValue(context, Object.class)을 통해 파라미터 studyPostId의 값을 평가
• field는 123

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RedissonDistributedLock {

    String hashKey();
    String field();
}

@Target: 커스텀 어노테이션이 적용될 수 있는 요소의 타입을 지정 @Retention(RetentionPolicy.RUNTIME): 어노테이션은 커스텀 어노노테이션의 유지 정책을 지정, 이 어노테이션이 런타임 동안 유지된다는 것을 의미 이는 리플렉션을 통해 이 어노테이션에 접근할 수 있음을 나타낸다 hashKey(), field(): 어노테이션의 속성을 정의

이렇게 적용해 봤다

    @RedissonDistributedLock(hashKey = "'apply'", field = "#studyPostId")
    public void acceptApplyEventPublish(Long studyPostId) {
        studyPostWriter.updateStudyPostApply(studyPostId);
    }

이렇게 더 간단하게 코드를 바꿀 수 있고 락을 획득하고 해제하는 redisson 분산락의 흩어진 관심사를 모듈화해서 분리해 봤다

@SpringBootTest
class StudyPostApplyEventPublisherTest {

    @Autowired
    private StudyPostApplyEventPublisher studyPostApplyEventPublisher;
    @Autowired
    private StudyPostRepository studyPostRepository;

    @AfterEach
    public void after() {
        studyPostRepository.deleteAll();
    }

    @Test
    void 동시에_100개의_요청의_스터디_지원서가_수락되면_게시글의_잔여석이_줄어든다() throws InterruptedException {
        // given
        Long postedUserId = 1L;
        StudyPostEntity post = StudyPostEntityFixture.SQLD.studyPostEntity_생성(postedUserId);
        StudyPostEntity savedPost = studyPostRepository.saveAndFlush(post);
        // when
        int threadCount = 100;
        ExecutorService executorService = newFixedThreadPool(32);
        CountDownLatch latch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            executorService.submit(() -> {
                try {
                    studyPostApplyEventPublisher.acceptApplyEventPublish(savedPost.getId());
                } finally {
                    latch.countDown();
                }
            });
        }
        latch.await();

        // then
        StudyPostEntity actualStudyPost = studyPostRepository.findById(savedPost.getId()).orElseThrow();
        // 100 - (1 * 100) = 0이 되어야 함
        assertEquals(0, actualStudyPost.getRemainingSeat());

    }

후기

아직 애매한 부분이 많다 명확하지 않다 알아보던 중, 김영한님의 스프링 고급 강의에 AOP에 대한 자세한 내용이 있다더라 다음 인프런 할인때 구매해서 봐야겠다

참고 자료: https://escapefromcoding.tistory.com/822 https://www.baeldung.com/cglib https://0soo.tistory.com/256#https://helloworld.kurly.com/blog/distributed-redisson-lock/#3-%EB%B6%84%EC%82%B0%EB%9D%BD%EC%9D%84-%EB%B3%B4%EB%8B%A4-%EC%86%90%EC%89%BD%EA%B2%8C-%EC%82%AC%EC%9A%A9%ED%95%A0-%EC%88%98%EB%8A%94-%EC%97%86%EC%9D%84%EA%B9%8C https://escapefromcoding.tistory.com/823

예전에 처음 개발을 시작하고 첫 프로젝트를 했을 때, 소셜 로그인을 위해서 rest template을 사용했다 하지만 코드가 너무 복잡해 지고 Feign client라고 스프링 부트의 어노테이션을 활용하여 정말 깔끔하게 사용할 수 있다 해서 사용했었다

지금 시점에서 그때의 선택을 다시 돌아보고 feign client에서 retry와 예외처리를 정리해보자

rest template은 동기식 클라이언트 라이브러리로 Blocking I/O를 사용하여 동시성이 높은 기능에서 성능 문제가 발생할 수 있다 bloking i/o는 i/o작업이 실행되는 동안 스프링 부트 기준 timeout 지정 안하면 해당 스레드가 멈추고 그런 스레드가 많아지면 성능에 문제가 생긴다

뭐 이건 feign도 똑같긴하다 하지만 클라이언트 인터페이스를 정의 하면 쉽게 기능을 구현할 수 있고 유지 관리할 수 있다. Spring Cloud와 통합하여 여러 기능을 제공한다

WebClient는 WebFlux의 라이브러리의 일부로 Non-Blocking I/O를 사용한다 확장성과 동시 요청 처리에 좋다

지금은 부하가 그리 많지 않아서 Feign으로 하고 있는데 나중에 비동기 상황의 이벤트 기반 아키텍쳐에서 WebClient를 활용하면 좋을 것 같다

문제 상황

1. 실패하면 재시도는 어떻게 할 수 있을까? ex) 알림톡을 보냈는데 실패했을 때
2. 외부 api에서 주는 여러 에러를 어떻게 받을 수 있을까? ex) PG결제 전송 예외가 생겼을 때

해결 방법

1. 실패하면 재시도는 어떻게 할 수 있을까? ex) 알림톡을 보냈는데 실패했을 때

@FeignClient(value = "post-api-AlimTalk",
        url = "${biz-message.request.alimtalk-url}",
        configuration = AlimTalkApiClientConfig.class)
public interface AlimTalkPostApiClient {

    @PostMapping("/messages")
    SendFormMsgResponse sendBotKeywordMessage(AlimTalkFeignRequest request);

    @PostMapping("/messages")
    SendFormMsgResponse sendPriorityChoiceCompletedMessage(AlimTalkFeignRequest request);

    @PostMapping("/messages")
    SendFormMsgResponse sendRecommendationUserProfilesMessage(AlimTalkFeignRequest request);

    @PostMapping("/messages")
    SendFormMsgResponse sendMaleMatchingCompletedMessage(AlimTalkFeignRequest feignRequest);

    @PostMapping("/messages")
    SendFormMsgResponse sendFemaleMatchingCompletedMessage(AlimTalkFeignRequest feignRequest);

    @PostMapping("/messages")
    SendFormMsgResponse sendAdditionalRecommendationMessage(AlimTalkFeignRequest feignRequest);

    @PostMapping("/messages")
    SendFormMsgResponse sendBothAdditionalRecommendationMessage(AlimTalkFeignRequest feignRequest);

    @PostMapping("/messages")
    SendFormMsgResponse sendFailMatchingResult(AlimTalkFeignRequest feignRequest);

    @PostMapping("/messages")
    SendFormMsgResponse sendRecommendationScheduleAnnounce(AlimTalkFeignRequest feignRequest);

    @PostMapping("/messages")
    SendFormMsgResponse sendAlimTalk(AlimTalkFeignRequest feignRequest);

}

이런식으로 짤 수 있다 전체적인 사용 방법은 스프링의 익숙한 어노테이션들이 있어서 너무 쉽게 식별이 가능 하다

위에 configuration에 집중 해 보자

저거 왜 쓰는거임? 알림톡에 각각 설정 파일 ex) 소셜 로그인, 알림톡, PG 등 외부 api의 설정파일이 단 하나로 할 수 없다 왜 와이? 알림톡은 보내는데 실패하면 재시도 로직을 작성할건데 PG가 실패하면 재시도? 사용자에게 결제 실패 이유를 알려주고 결제를 취소 시키는게 맞다 그래서 각각 설정파일을 다르게 하는 것이다. 또한 헤더 설정도 각각 전부 스펙이 다르기 떄문에 저렇게 한다

// 여기에 @Configuration 달면 안됨 전체 적용 되버리니까
@Import({BizMsgApiHeaderConfig.class}) //헤더 설정하는건데 여기서 할 수 있지만 너무 커져서 분리 후 import
public class AlimTalkApiClientConfig {

    private static final long PERIOD = 500L; //기본 재시도 주기 밀리초 설정. 첫 번째 시도 후 다음 시도까지 500ms 대기합니다.
    private static final long MAX_PERIOD = TimeUnit.SECONDS.toMillis(3L); // 최대 재시도 주기를 초 단위로 설정하고 밀리초로 변환합니다. 결과적으로 최대 3000ms 동안 대기할 수 있습니다
    private static final int MAX_ATTEMPTS = 5; // 최대 5번 재시도합니다.

    @Bean
    public Retryer retryer() {
        return new Retryer.Default(PERIOD, MAX_PERIOD, MAX_ATTEMPTS);
    }

    @Bean
    @ConditionalOnMissingBean(value = ErrorDecoder.class)
    public AlimTalkErrorDecoder commonFeignErrorDecoder() {
        return new AlimTalkErrorDecoder();
    }

    @Bean
    public FeignFormatterRegistrar localDateFeignFormatterRegistrar() {
        return formatterRegistry -> {
            DateTimeFormatterRegistrar registrar = new DateTimeFormatterRegistrar(); //시간 찍눈 형식
            registrar.setUseIsoFormat(true);
            registrar.registerFormatters(formatterRegistry);
        };
    }

}

주석을 보면 설명이 되어있다

저기에 있는 Retryer도 구현해서 커스텀할 수 있지만 아직 재시도 로직에 대한 스펙이 크지 않아서 저상태로 냅뒀다

우리는 알림톡 디코더를 봐보자

위의 예시에서는 @ConditionalOnMissingBean(value = ErrorDecoder.class) 어노테이션을 사용하여 CustomErrorDecoder 빈을 등록한다. 만약 컨텍스트에 ErrorDecoder 타입의 빈이 이미 존재하지 않는다면, CustomErrorDecoder 빈을 등록한다 하지만 우리는 AlimTalkErrorDecoder라는 ErrorDecoder타입의 빈을 등록 해 주었다

public class AlimTalkErrorDecoder implements ErrorDecoder {

    @Override
    public Exception decode(String methodKey, Response response) {
        try {
            FeignException exception = FeignException.errorStatus(methodKey, response);
            int status = response.status();

            // 500번대는 기본적으로 리트라이
            if (HttpStatus.valueOf(status).is5xxServerError()) {
                throw new RetryableException(
                        status,
                        exception.getMessage(),
                        response.request().httpMethod(),
                        exception,
                        null,
                        response.request());
            }

            // 401 Unauthorized 에러에 대해 RetryableException을 던짐
            if (status == 401) {
                return new RetryableException(
                        status,
                        exception.getMessage(),
                        response.request().httpMethod(),
                        exception,
                        null,
                        response.request());
            }

            // retry 정책 제외 에러는 AlimTalkUnHandleException 을 던짐
            String responseBody = new String(response.body().asInputStream().readAllBytes(), "UTF-8");
            if (status >= 400 && status < 500) {
                return new AlimTalkApiRequestException(ErrorCode.ALIMTALK_API_CLIENT_ERROR, responseBody);
            }

            if (status >= 500) {
                return new AlimTalkApiRequestException(ErrorCode.ALIMTALK_API_SERVER_ERROR, responseBody);
            }

            return exception;
        } catch (IOException e) {
            throw new AlimTalkUnHandleException(ErrorCode.ALIMTALK_UNHANDLE_ERROR, e.getMessage());
        }
    }
}

주석에 잘 설명 했다 이런식으로 Feign에서 일어나는 외부 api에 대한 재시도 로직을 작성할 수 있다

2. 외부 api에서 주는 여러 에러를 어떻게 받을 수 있을까? ex) PG결제 전송 예외가 생겼을 때

방금 것과 비슷하다 PG에는 retry전략을 뺏다 왜? FORBIDDEN_CONSECUTIVE_REQUEST 반복적인 요청은 허용되지 않습니다. 잠시 후 다시 시도해주세요. 애초에 반복적인 요청은 허용되지 않을 뿐더러 그게 아니라 해도

INVALID_REJECT_CARD 카드 사용이 거절되었습니다. 카드사 문의가 필요합니다. 이런 반복해서 될 문제가 아닌 에러들 밖에 없기 때문이다

그냥 retry뺴고

public class TossPaymentsErrorDecoder implements ErrorDecoder {

    private final ObjectMapper objectMapper = new ObjectMapper();


    @Override
    public Exception decode(String methodKey, Response response) {
        try {
            InputStream bodyStream = response.body().asInputStream();
            JsonNode body = objectMapper.readTree(bodyStream);

            String message = body.get("message").asText();

            return new TossPaymentsConfirmException(ErrorCode.TOSS_PAYMENTS_CONFIRM_ERROR, message);
        } catch (IOException e) {
            throw new TossPaymentsUnHandleException(ErrorCode.TOSS_PAYMENTS_UNHANDLE_ERROR, e.getMessage());
        }
    }
}

예외 메시지를 프론트에게 그대로 보내주는 방식을 채택했다 (이 부분은 추후 문제가 생길시 변경될 수 있음)

추가

AlimTalkApiClientConfig에 보면 헤더 설정이 아예 없던데 어캄?

@Import({BizMsgApiHeaderConfig.class}) //헤더 설정하는건데 여기서 할 수 있지만 너무 커져서 분리 후 import 여기서 설정할 수 있다

• 예전에 그룹 스터디 참여 로직에 Pessimistic Lock(이하 비관적 락)을 이용하여 동시성 이슈를 처리했었다 링크: https://velog.io/@wellbeing-dough/%EC%8A%A4%ED%94%84%EB%A7%81-%EB%B6%80%ED%8A%B8-%EB%8F%99%EC%8B%9C%EC%84%B1-%EC%B2%98%EB%A6%AC

• 하지만 비관적 락은 락을 거는 동안 다른 트랜잭션이 대기 상태가 되어 성능 저하가 발생할 수 있다 락을 거는 동안 다른 트랜잭션이 대기 상태가 되기 때문이다 특히 트랜잭션이 길어질 경우 문제가 된다.

• 여러 트랜잭션이 동시에 락을 요청할 때 교착 상태가 발생할 수 있다 이를 해결하기 위해 복잡한 트랜잭션 관리와 교착 상태 회피 알고리즘이 필요하다

• 데이터베이스 서버의 부하가 증가할 수 있다. 스터디헙 인프라는 오토 스케일링으로 ec2는 늘릴 수 있지만 데이터베이스 이중화는 되어있지 않다.

• 비관적 락은 단일 데이터베이스 서버의 자원에 의존하므로, 대규모 분산 시스템에서는 확장성이 제한될 수 있다

이런 단점으로 redis를 사용해서 락을 구현하기로 했다

대표적으로 lettuce와 redisson 이 있는데 차이점은

lettuce는 스핀 락 방식으로 구현되어있고 분산 락 기능이 따로 제공되지 않아 따로 직접 구현이 필요하다 하지만 redisson은 pub/sub 구조로 분산락 기능이 기본적으로 제공된다 스핀락 방식은 lock이 해제되었는지 주기적으로 retry를 해야하므로 CPU 리소스를 많이 소모할 수 있다 분산 락 방식은 락에 TTL(Time-To-Live)을 설정하여 프로세스 중단 시 락이 자동으로 해제될 수 있게 한다. 이는 락의 영구적 잔류 문제를 방지한다

그래서 redisson을 사용하여 동시성 이슈를 해결해 보겠다

문제 해결

동시성 이슈에 대한 설명은 예전에 https://velog.io/@wellbeing-dough/%EC%8A%A4%ED%94%84%EB%A7%81-%EB%B6%80%ED%8A%B8-%EB%8F%99%EC%8B%9C%EC%84%B1-%EC%B2%98%EB%A6%AC 여기서 알아봤으니 따로 하지 않고 구현만 해보자

일단 pub/sub 에 대해서 알아보면 기본적으로 Publisher: 특정 채널에 메시지를 발행하는 클라이언트입니다. 발행자는 메시지를 특정 채널에 보낼 뿐, 누가 그 메시지를 수신하는지는 알지 못합니다. Subscriber: 특정 채널을 구독하는 클라이언트입니다. 구독자는 자신이 구독한 채널에 발행된 모든 메시지를 수신합니다.

두개의 터미널에서 redis-cli로 접속 해 보자

첫번째 터미널에서는 subscribe ch1을해보자 이러면 ch1이라는 채널을 구독한다

두번째 터미널에서는 publish ch1 hello!라고 해보자 그러면 첫번째 터미널 즉 ch1을 구독하는 곳에서 hello!라는 메세지를 확인할 수 있다

redis는 자신이 점유하고 있는 락을 해제할 때 채널에 메세지를 보내줘서 락을 획득해야 하는 스레드들에게 락을 획득하라고 전해준다 그러면 그 메세지를 받은 스레드들은 락 획득을 시도하게 된다

이렇게 하면 retry를 사용한 스핀락 방식 보다 redis의 부하를 줄여준다

redisson 라이브러리를 추가했는데

    implementation 'org.redisson:redisson-spring-boot-starter:3.17.4'

"Failed to start bean 'documentationPluginsBootstrapper'; nested exception is java.lang.NullPointerException"

무슨 스웨거에서 빌드 에러가 뜬다

https://www.inflearn.com/questions/625844/%EC%84%A0%EC%83%9D%EB%8B%98%EC%9D%B4-%ED%94%84%EB%A1%9C%EC%A0%9D%ED%8A%B8-spring-%EB%B2%84%EC%A0%84-%EA%B8%B0%EC%A4%80%EC%9C%BC%EB%A1%9C-%EC%8B%A4%EB%AC%B4-%ED%86%B5%ED%95%B4-redisson-%EC%A0%81%EC%9A%A9%EC%8B%9C-%EB%AC%B8%EC%A0%9C

구글링 해봤는데 이런 도움이.... 너무 감사합니다

https://github.com/springfox/springfox/issues/3462#issuecomment-979548234 여기에 해결책이 나와있다

@Component
@RequiredArgsConstructor
@Slf4j
public class StudyPostApplyEventPublisher {

    private final RedissonClient redissonClient;
    private final StudyPostWriter studyPostWriter;
    private final StudyPostReader studyPostReader;

        @Transactional
    public void acceptApplyEventPublish(Long studyId) {
        StudyPostEntity studyPost = studyPostRepository.findByIdWithPessimisticLock(studyId).orElseThrow(PostNotFoundException::new);
        studyPost.decreaseRemainingSeat();
        studyPost.closeStudyPostIfRemainingSeatIsZero();
        studyPostRepository.save(studyPost);
    }


}

기존에 이렇게 낙관적 락을 수행한 부분에서

@Component
@RequiredArgsConstructor
@Slf4j
public class StudyPostApplyEventPublisher {

    private final RedissonClient redissonClient;
    private final StudyPostWriter studyPostWriter;
    private final StudyPostReader studyPostReader;

    @Timer
    public void acceptApplyEventPublish(Long studyPostId) {
        StudyPostEntity studyPost = studyPostReader.readById(studyPostId);
        RLock lock = redissonClient.getLock(studyPost.getId().toString());
        boolean available = false;
        try {
            available = lock.tryLock(10, 1, TimeUnit.SECONDS);
            if (!available) {
                throw new StudyApplyLockAcquisitionException();
                return;
            }
            studyPostWriter.updateStudyPostApply(studyPost.getId());
        } catch (InterruptedException e) {
            throw new StudyApplyLockAcquisitionException();
        } finally {
            if (available) {
                lock.unlock();
            }
        }

    }
}

• RLock lock = redissonClient.getLock(studyPost.getId().toString()); -> studyPostId를 문자열로 변환하여 락 객체를 생성한다. 이 ID는 Redis에서 락을 구분하는 키로 사용된다

• available = lock.tryLock(10, 1, TimeUnit.SECONDS); -> 락 획득을 시도한다 첫번째 매개변수 10: 최대 대기 시간. 락을 얻기 위해 최대 10초 동안 기다린다 두번째 매개변수 1: 락의 임대 시간. 락을 획득한 후 1초 동안 유지된다 TimeUnit.SECONDS: 시간 단위로 초를 사용한다 락을 획득하면 available이 true 그렇지 않으면 false

• 락을 획득하지 못하면 예외 처리를 했다

그리고 업데이트 하는 로직은

@Component
@RequiredArgsConstructor
@Slf4j
public class StudyPostWriter {

    private final StudyPostRepository studyPostRepository;
    private final StudyPostReader studyPostReader;

    @Transactional
    public void updateStudyPostApply(Long studyPostId) {
        StudyPostEntity studyPost = studyPostRepository.findById(studyPostId).orElseThrow(PostNotFoundException::new);
        studyPost.decreaseRemainingSeat();
        studyPost.closeStudyPostIfRemainingSeatIsZero();
        studyPostRepository.saveAndFlush(studyPost);
    }

}

여기 잘 해놨다

    @Test
    void 동시에_100개의_요청의_스터디_지원서가_수락되면_게시글의_잔여석이_줄어든다() throws InterruptedException {
        // given
        Long postedUserId = 1L;
        StudyPostEntity post = StudyPostEntityFixture.SQLD.studyPostEntity_생성(postedUserId);
        StudyPostEntity savedPost = studyPostRepository.saveAndFlush(post);
        // when
        int threadCount = 100;
        ExecutorService executorService = newFixedThreadPool(32);
        CountDownLatch latch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            executorService.submit(() -> {
                try {
                    studyPostApplyEventPublisher.acceptApplyEventPublish(post.getStudyId());
                } finally {
                    latch.countDown();
                }
            });
        }
        latch.await();

        // then
        StudyPostEntity actualStudyPost = studyPostRepository.findById(savedPost.getId()).orElseThrow();
        // 100 - (1 * 100) = 0이 되어야 함
        assertEquals(0, actualStudyPost.getRemainingSeat());

    }

테스트 코드도 잘 통과한다

락 획득하는 내부 코드 한번 까보자 근데 pub/sub을 알아서 redisson 내부에서 해주는건가? 처음에 얼마나 많은 스레드가 실패해서 대기중인게 궁금해서 AtomicLong을 스태틱으로 넣어놓고 락 획득 실패한 부분에서 fail++ 를 하여서 몇번 실패했는지 알아봤는데 0번이 나왔다... 어?? 아 그러면 available = lock.tryLock(10, 1, TimeUnit.SECONDS); 이부분에서 알아서 락을 pub/sub으로 락을 획득하고 끝끝내 실패한 경우에 락 획득 실패 부분으로 가는구나...

tryLock 라이브러리 코드 까보자 RedissonLock.class

    public boolean tryLock() {
        return (Boolean)this.get(this.tryLockAsync());
    }

    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        return this.evalWriteAsync(this.getRawName(), LongCodec.INSTANCE, command, "if (redis.call('exists', KEYS[1]) == 0) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; return redis.call('pttl', KEYS[1]);", Collections.singletonList(this.getRawName()), new Object[]{unit.toMillis(leaseTime), this.getLockName(threadId)});
    }

    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
        if (ttl == null) {
            return true;
        } else {
            time -= System.currentTimeMillis() - current;
            if (time <= 0L) {
                this.acquireFailed(waitTime, unit, threadId);
                return false;
            } else {
                current = System.currentTimeMillis();
                CompletableFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId);

                try {
                    subscribeFuture.get(time, TimeUnit.MILLISECONDS);
                } catch (TimeoutException | ExecutionException var20) {
                    if (!subscribeFuture.cancel(false)) {
                        subscribeFuture.whenComplete((res, ex) -> {
                            if (ex == null) {
                                this.unsubscribe(res, threadId);
                            }

                        });
                    }

                    this.acquireFailed(waitTime, unit, threadId);
                    return false;
                }

                try {
                    time -= System.currentTimeMillis() - current;
                    if (time <= 0L) {
                        this.acquireFailed(waitTime, unit, threadId);
                        boolean var22 = false;
                        return var22;
                    } else {
                        boolean var16;
                        do {
                            long currentTime = System.currentTimeMillis();
                            ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
                            if (ttl == null) {
                                var16 = true;
                                return var16;
                            }

                            time -= System.currentTimeMillis() - currentTime;
                            if (time <= 0L) {
                                this.acquireFailed(waitTime, unit, threadId);
                                var16 = false;
                                return var16;
                            }

                            currentTime = System.currentTimeMillis();
                            if (ttl >= 0L && ttl < time) {
                                ((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture)).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                            } else {
                                ((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture)).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                            }

                            time -= System.currentTimeMillis() - currentTime;
                        } while(time > 0L);

                        this.acquireFailed(waitTime, unit, threadId);
                        var16 = false;
                        return var16;
                    }
                } finally {
                    this.unsubscribe((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture), threadId);
                }
            }
        }
    }

    protected CompletableFuture<RedissonLockEntry> subscribe(long threadId) {
        return this.pubSub.subscribe(this.getEntryName(), this.getChannelName());
    }

    protected void unsubscribe(RedissonLockEntry entry, long threadId) {
        this.pubSub.unsubscribe(entry, this.getEntryName(), this.getChannelName());
    }

    public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
        return this.tryLock(waitTime, -1L, unit);
    }

    public void unlock() {
        try {
            this.get(this.unlockAsync(Thread.currentThread().getId()));
        } catch (RedisException var2) {
            RedisException e = var2;
            if (e.getCause() instanceof IllegalMonitorStateException) {
                throw (IllegalMonitorStateException)e.getCause();
            } else {
                throw e;
            }
        }
    }

비동기적으로 락을 시도하고, 결과를 동기적으로 반환 tryLockAsync() 메소드를 호출하여 비동기적으로 락을 시도하고, 그 결과를 get() 메소드를 통해 동기적으로 반환

tryLockInnerAsync여기서는 Redis Lua 스크립트를 실행하여 락을 시도하고 락이 없으면(redis.call('exists', KEYS[1]) == 0), 락을 설정하고(hincrby), 만료 시간을 설정(pexpire). 현재 스레드가 이미 락을 가지고 있으면(redis.call('hexists', KEYS[1], ARGV[2]) == 1), 락 횟수를 증가시키고(hincrby), 만료 시간을 갱신한다(pexpire). 그렇지 않으면 현재 TTL(Time To Live)을 반환한다(pttl).

가장 중요한 tryLock 메서드는 지정된 대기 시간(waitTime) 동안 락을 시도하고, 락을 획득한 후 지정된 시간(leaseTime) 동안 락을 유지한다

    Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
    if (ttl == null) {
        return true;
    }

ttl: tryAcquire 메소드를 호출하여 락을 시도 락을 획득하면 null을 반환 락을 획득한 경우 true를 반환하고 메소드를 종료한다

    time -= System.currentTimeMillis() - current;
    if (time <= 0L) {
        this.acquireFailed(waitTime, unit, threadId);
        return false;
    }

락을 획득하지 못한 경우, 남은 대기 시간을 계산 남은 시간이 없으면 acquireFailed 메소드를 호출하고 false를 반환

    current = System.currentTimeMillis();
    CompletableFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId);

    try {
        subscribeFuture.get(time, TimeUnit.MILLISECONDS);
    } catch (TimeoutException | ExecutionException var20) {
        if (!subscribeFuture.cancel(false)) {
            subscribeFuture.whenComplete((res, ex) -> {
                if (ex == null) {
                    this.unsubscribe(res, threadId);
                }
            });
        }
        this.acquireFailed(waitTime, unit, threadId);
        return false;
    }

락 이벤트를 구독하여 락 해제를 기다린다 구독이 완료될 때까지 남은 시간 동안 기다린다 TimeoutException 또는 ExecutionException이 발생하면 구독을 취소하고, acquireFailed 메소드를 호출하여 false를 반환한다

    try {
        time -= System.currentTimeMillis() - current;
        if (time <= 0L) {
            this.acquireFailed(waitTime, unit, threadId);
            boolean var22 = false;
            return var22;
        } else {
            boolean var16;
            do {
                long currentTime = System.currentTimeMillis();
                ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
                if (ttl == null) {
                    var16 = true;
                    return var16;
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0L) {
                    this.acquireFailed(waitTime, unit, threadId);
                    var16 = false;
                    return var16;
                }

                currentTime = System.currentTimeMillis();
                if (ttl >= 0L && ttl < time) {
                    ((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture)).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    ((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture)).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= System.currentTimeMillis() - currentTime;
            } while(time > 0L);

            this.acquireFailed(waitTime, unit, threadId);
            var16 = false;
            return var16;
        }
    } finally {
        this.unsubscribe((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture), threadId);
    }

이제 반복적으로 락 획득을 시도 tryAcquire 메소드를 호출하여 락을 시도하고, null을 반환하면 락을 획득 남은 시간이 없으면 acquireFailed 메소드를 호출하고 false를 반환 락을 획득하지 못하면 이벤트를 기다리고 다시 시도한다 최종적으로 락을 해제 unsubscribe

결론적으로 간단하게 정리하면

1. tryAcquire를 이용하여 락을 시도하고 획득되면 true 반환
2. 락 획득에 실패했을 때, ttl, 대기시간이 남아있으면 그 스레드는 락이 사용 가능하다는 알림을 받는 채널을 구독
3. 그 스레드는 락이 사용 가능하다는 메시지가 도착할때 까지 대기 대기하다가 ttl이 오버되면 즉, TimeoutException 또는 ExecutionException이 발생하면 구독을 취소하고, acquireFailed 메소드를 호출하여 false를 반환한다
4. 성공적으로 락을 획득하거나, 시간이 초과되서 실패하면 구독 해지

오호 이런 내부 구조를 갖고있다

참고: https://www.inflearn.com/questions/625844/%EC%84%A0%EC%83%9D%EB%8B%98%EC%9D%B4-%ED%94%84%EB%A1%9C%EC%A0%9D%ED%8A%B8-spring-%EB%B2%84%EC%A0%84-%EA%B8%B0%EC%A4%80%EC%9C%BC%EB%A1%9C-%EC%8B%A4%EB%AC%B4-%ED%86%B5%ED%95%B4-redisson-%EC%A0%81%EC%9A%A9%EC%8B%9C-%EB%AC%B8%EC%A0%9C

https://github.com/springfox/springfox/issues/3462#issuecomment-979548234

어떻게 1초동안 10번을 보낸건진 모르겠다...

그리고 중복 요청 이슈가 발생하여 알림톡이 두번 나갔는데 이것도 유저 입장에선 우리 서비스가 너무 허술해 보이기도 하고 알림톡 자체가 비싸진 않지만 유료인 점을 감안해서 중복 요청 방지도 구현했었다

이미 코드는 짜져있고 이것을 커스텀 어노테이션으로 좀 이쁘게 바꿔보자

문제 해결

public class WebConfig implements WebMvcConfigurer {

    @Value("${cors.allowed.origins:}")
    private String[] ALLOWED_CORS_URLS;
    private final TokenInterceptor tokenInterceptor;
    private final AdminTokenInterceptor adminTokenInterceptor;
    private final ApiThrottlingInterceptor apiThrottlingInterceptor;
    private final UserIdentifierArgumentResolver userIdentifierArgumentResolver;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(tokenInterceptor)
                .order(1);

        registry.addInterceptor(adminTokenInterceptor)
                .order(2);

        registry.addInterceptor(apiThrottlingInterceptor)
                .order(3);
    }

이런식으로 일단 해당 인터셉터를 타게 했다

우리가 볼 내용은 ApiThrottlingInterceptor이다

일단 어노테이션 두개 커스텀 해서 만들어준다

@HandleDuplicateRequest: 중복 요청을 방지하겠다는 뜻 @ApiThrottled: rate limit를 걸었다는 뜻

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface HandleDuplicateRequest {
}

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface ApiThrottled {
}

인터셉터 내용이다

@Component
@RequiredArgsConstructor
public class ApiThrottlingInterceptor implements HandlerInterceptor {

    private final RateLimitTracker rateLimitTracker;
    private final DuplicateRequestTracker duplicateRequestTracker;

    @Override
    public boolean preHandle(final HttpServletRequest request,
                             final HttpServletResponse response,
                             final Object handler) {
        if (!(handler instanceof final HandlerMethod handlerMethod)) {
            return true;
        }

        boolean isDuplicateRequest = handlerMethod.hasMethodAnnotation(HandleDuplicateRequest.class);
        boolean isApiThrottled = handlerMethod.hasMethodAnnotation(ApiThrottled.class);
        String clientIp = ClientIpUtil.getClientIp(request);

        if (isDuplicateRequest) {
            duplicateRequestTracker.resolveBucket(clientIp);
        }

        if (isApiThrottled) {
            rateLimitTracker.resolveBucket(clientIp);
        }

        return true;
    }
}

일단 if (!(handler instanceof final HandlerMethod handlerMethod)) { return true; }

이 코드는 먼저 handler가 HandlerMethod의 인스턴스인지 확인한다 쉽게 말해서 handler가 컨트롤러인지 확인, mvc에서 정적 파일 요청도 가능하다 컨트롤러가 아니면 뒤의 로직 스킵한다

다음 부분은

boolean isDuplicateRequest = handlerMethod.hasMethodAnnotation(HandleDuplicateRequest.class); boolean isApiThrottled = handlerMethod.hasMethodAnnotation(ApiThrottled.class);

hasMethodAnnotation은 HandlerMethod 클래스의 메서드로, 핸들러 메서드에 지정된 어노테이션이 있는지 확인 한다 handlerMethod.hasMethodAnnotation(HandleDuplicateRequest.class)는 핸들러 메서드에 @HandleDuplicateRequest 어노테이션이 있는지 확인한다

handlerMethod.hasMethodAnnotation(ApiThrottled.class)는 핸들러 메서드에 @ApiThrottled 어노테이션이 있는지 확인한다

이제 그 다음 로직 중 rateLimitTracker는 Bucket4J를 사용하여 Rate Limit를 하는건데 저번에 기록 했으니 따로 기록하진 말자

    @ApiThrottled
    @HandleDuplicateRequest
    @Operation(description = "나에게 추천된 유저 프로필 조회")
    @GetMapping("/v3/recommendations/profile")
    public ResponseEntity<GetRecommendedUserProfileInfoResponse> getRecommendedUserProfile(생략) {
        return ResponseEntity.ok().body(recommendationService.getRecommendedUserProfile(생략));
    }

이제 이런식으로 api 컨트롤러 메서드에 rate limit이나 중복 요청 방지를 쏙쏙 골라서 할 수 있다

후기

처음엔 이걸 AOP로 구현할까 생각해서 구현해봤는데 막상 구현해 보니 인터셉터를 사용하여 레이트 리미팅을 구현하는 것이 더 나아 보였다 스프링 인터셉터는 요청이 컨트롤러에 도달하기 전에 처리되기 때문에, 이를 사용하여 레이트 리미팅을 구현하면 코드가 더 명확하고 재사용성이 높아진다고 생각해서 인터셉터와 커스텀 어노테이션 선에서 하는 것으로 했다

구글링을 해보니 scouter server collector와 scouter client의 서버 시간이 달라서 그렇다고 한다

scouter server의 timezone을 Asia/Seoul로 변경했다.

스카우터 서버의 date 명령어엔 분명하게 KST시간이 나온다

같은 문제가 계속 발생했다...

고민하다가 scoueter server의 startup.sh 실행파일의 JVM -Duser.timezone 옵션으로 시간대를 설정해 보았다

nohup java -Xmx2048m -Xmx2048m -classpath ./scouter-server-boot.jar scouter.boot.Boot ./lib > nohup.out &
sleep 1
tail -100 nohup.out

이걸

nohup java -Xmx2048m -Xmx2048m -Duser.timezone=Asia/Seoul -classpath ./scouter-server-boot.jar scouter.boot.Boot ./lib > nohup.out &
sleep 1
tail -100 nohup.out

이렇게 -Duser.timezone=Asia/Seoul 옵션으로...

잘 된다 ㅎ 구글링해도 이 방법은 없어보여 올려봤다

어플리케이션의 성능 문제를 실시간으로 감지하고, 장애의 근본 원인을 빠르게 찾을 수 있도록 도와주는 APM(어플리케이션의 성능을 모니터링하고 관리하는 도구)을 도입하기로 했다 핀포인트, 스카우터, 다이나트레이스, 뉴 레릭, 인스타나, 와탭 등등 여러 도구가 있었지만 돈이 없기 때문에 오픈소스인 핀포인트, 스카우터중 골라야 했고 그중에 스카우터는 문서도 잘 되어있고 사람들이 많이 사용해서 참고 자료도 많았다 그래서 스카우터로 도입하기로 했다

1. 인프라 현황

우리는 Elastic beanstalk(이하 빈스톡) + github acions를 활용해서 CICD배포를 진행하고 있다

scouter를 도입하는 방법자체는 ec2에 들어가서 scouter wget하고 설정파일 설정하고 실행하는 간단한 작업이지만 cicd를 하고 있는 서비스에서 배포할때마다 ec2에 접근해서 저 일련과정을 하는것 자체가 비효율적이라고 생각했고 .ebextensions라고 웹 어플리케이션 소스 코드에 빈스톡 구성 파일을 추가해서 사용하기로 했다 (벌써부터 느껴지는 삽질의 기운...)

스카우터의 개념과 작동방식, 용어와 스카우터를 설치하고 설정하고 실행하는 방법은 너무 잘 나와있어서 여기에 굳이 쓰진 않겠다

2. 도입 시작

스카우터의 개념과 작동방식, 용어와 스카우터를 설치하고 설정하고 실행하는 방법은 너무 잘 나와있어서 여기에 굳이 쓰진 않겠다

1. 계획

스카우터 서버(collector), 스카우터 호스트 에이전트, 스카우터 자바 에이전트를 전부 어플리케이션이 돌아가는 서버(이하 상용 서버)에 넣는거 자체가 좀 성능에 누를 끼치지 않을까 고려해서 스카우터 서버는(collector) 새로운 ec2를 파기로 했다 (이것도 돈 아까우니까 새로운 계정 파서 일단 1년 프리티어) 왜 와이? 제대로 도입하기전에 실제로 한번 ec2에 접속해서 서버, 에이전트, 스프링 3개를 돌리니까 cpu 메모리가 고갈되는 현상을 확인했기 때문 스카우터 서버 ec2는 고정 ip를 설정해줬다 참고로 고정 ip 할당하면 프리티어까진 돈이 안든다

2. 스카우터 클라이언트 도입

그냥 로컬에 설치하고 실행하면 된다

실행 안되는 이슈

스카우터 클라이언트를 내 로컬 컴퓨터에 설치하는중에 실행이 안되는것을 보았다 권한이 없었다 했나 맥북에서 발생하는 이슈였다

실행 안되는 이슈 해결

맥용 클라이언트는 자바 11이상인데 나는 17이라서 괜찮고 xattr -cr scouter.client.app 이 명령어를 입력하니까 잘 됬다 scouter.client.app 디렉토리와 그 안의 모든 파일 및 디렉토리의 확장 속성을 제거하는 명령어인데 인터넷에서 다운로드한 애플리케이션이나 파일이 실행되지 않을 때, "확장 속성"을 제거하여 실행할 수 있게 한다고 한다 해결!

3. 스카우터 서버(collector) 도입

설정파일에 6100포트와 db_dir, log_dir 설정해주고 실행했다 물론 스카우터 서버 ec2에 인바운드 규칙에 6100 포트 대상은 상용 서버 ip로 해서 tcp, udp다 열어주었고, 아웃바운드 규칙에 6100포트 대상은 내 로컬 컴퓨터 ip 로 tcp, udp 다 열어주었다 스카우터 서버는 도입할때 딱히 이슈는 없었다

4. 스카우터 호스트 에이전트 도입

상용 서버의 아웃바운드 규칙에 6100 포트 대상은 스카우터 서버(collector) 고정ip로 tcp, udp다 열어주자

여기서부터 삽질 시작이였다 ebextensions에 03-install-scouter-host-agent.config

files:
  "/opt/elasticbeanstalk/hooks/appdeploy/pre/03-install-scouter-host-agent.sh":
    mode: "000755"
    owner: root
    group: root
    content: |
      #!/bin/bash

      SCOUTER_VERSION="2.20.0"
      SCOUTER_DIR="/opt/scouter"
      SCOUTER_URL="https://github.com/scouter-project/scouter/releases/download/v$SCOUTER_VERSION/scouter-all-$SCOUTER_VERSION.tar.gz"
      SCOUTER_HOST_AGENT_DIR="$SCOUTER_DIR/agent.host"
      SCOUTER_HOST_AGENT_CONF="$SCOUTER_HOST_AGENT_DIR/conf/scouter.conf"

      mkdir -p $SCOUTER_DIR
      wget $SCOUTER_URL -O /tmp/scouter-all.tar.gz
      tar -xzf /tmp/scouter-all.tar.gz -C $SCOUTER_DIR --strip-components=1

      cat <<EOF > $SCOUTER_HOST_AGENT_CONF
      net_collector_ip= {스카우터 서버(collector) 의 고정 ip}
      net_collector_udp_port=6100
      net_collector_tcp_port=6100
      EOF

      # Ensure the script has execution permissions
      chmod +x $SCOUTER_HOST_AGENT_DIR/host.sh

      cd $SCOUTER_HOST_AGENT_DIR
      ./host.sh start

commands:
  01_run_scouter_install_script:
    command: sudo /opt/elasticbeanstalk/hooks/appdeploy/pre/03-install-scouter-host-agent.sh

스카우터 호스트 에이전트 메모리 이슈

이렇게 해줬는데 배포가 안됬다... github actions에 로그는

Error: Deployment failed: Error: Environment still has health Yellow 30 seconds after update finished!

이렇다

1. 03-install-scouter-host-agent.config의 코드 문제일까? 1) 03-install-scouter-host-agent.config에서 command 삭제하고 배포, 그다음에 ec2 직접 접속해서 수동 실행 해보기

1-1 번 방법으로 해보니 cpu 그래프에 파란 선이 보인다 잘 되는 것이다 ec2 상태 모니터링을 확인했는데 스토리지가 갑자기 치솟으더니 cpu도 치솟고 서버가 죽었다 일단 메모리 문제라고 생각이 들었다

스카우터 호스트 에이전트 메모리 이슈 해결

1. 샘플링 단위가 1초라서 너무 많은 메모리를 사용하고 있는건 아닐까? 1) 일단 scouter.host 의 conf에 설정할때 샘플링 단위를 널널하게 10초로 둬보자 net_collector_ip=3.36.225.29 net_collector_udp_port=6100 net_collector_tcp_port=6100 hook_profile_fullstack_method_sampling_interval_ms=10000 hook_args_sampling_interval_ms=10000

1-1) 해도 같은 이슈 발생

2. 스카우터가 실행되는 Java 애플리케이션에 할당할 최대 힙 메모리 크기를 제한해보자 1) 메모리 설정 값인 -Xmx256m 사용

2-1) 해도 같은 이슈 발생

3. t3.micro의 메모리가 1G라서 그런가? 1) t3.small로 변경

3-1 실행 하니 이젠 다른 에러가 터짐

스카우터 호스트 에이전트 timeout 이슈

16:29:46 WARN: The following instances have not responded in the allowed command timeout time (they might still finish eventually on their own): [{인스턴스id}].
16:29:46 INFO: Command execution completed on all instances. Summary: [Successful: 0, TimedOut: 1].
16:29:46 ERROR: Unsuccessful command execution on instance id(s) 'i-{인스턴스id}'. Aborting the operation.
16:29:46 ERROR: Failed to deploy application.
16:29:52 ERROR: Deployment failed! Current State: Version: ***-2024-06-03T19-24-02, Health: Green, Health Status: Ok
Error: Deployment failed: Error: Deployment failed! Current State: Version: ***-2024-06-03T19-24-02, Health: Green, Health Status: Ok

그러면 아까랑 똑같이 03-install-scouter-host-agent.config에서 command 삭제하고 배포, 그다음에 ec2 직접 접속해서 수동 실행 해보기를 해보자 -> 이렇게 하니 계속 잘 된다.... 메모리 이슈는 해결된 것 같다 메모리 1기가로는 scouter와 스프링을 같이 돌리기 힘든 것 같다 성능 최적화는 나중에 해보고 일단 timeout이슈를 해결해보자

거의 하루종일 구글링을 해본 결과 https://theholyjava.wordpress.com/2015/07/29/fixing-a-mysterious-ebextensions-command-time-out-aws-elastic-beanstalk/ 해당 블로그 글을 참고할 수 있었다

나랑 같은 이슈가 생겼고 &를 이용하여 백그라운드로 시작하는 명령어의 stdin stderr의 출력을 /dev/null로 redirection하라는 소리였다

스카우터 호스트 에이전트 timeout 이슈 해결

1. &를 이용하여 백그라운드로 시작하는 명령어의 stdin stderr의 출력을 /dev/null로 redirection 1) &를 이용하여 백그라운드로 시작하는 명령어의 stdin stderr의 출력을 /dev/null로 redirection

1-1) 그래도 계속 timeout이슈 발생

2. commands가 인스턴스를 시작할때 실행되니까 인스턴스가 힘든거 아닐까? 1) container_command를 이용해서 어플리케이션 코드가 배포된 후 실행해보자

2-1) 이렇게 하니까 해결됬다!

files:
  "/opt/elasticbeanstalk/hooks/appdeploy/pre/03-install-scouter-host-agent.sh":
    mode: "000755"
    owner: root
    group: root
    content: |
      #!/bin/bash

      SCOUTER_VERSION="2.20.0"
      SCOUTER_DIR="/opt/scouter"
      SCOUTER_URL="https://github.com/scouter-project/scouter/releases/download/v$SCOUTER_VERSION/scouter-all-$SCOUTER_VERSION.tar.gz"
      SCOUTER_HOST_AGENT_DIR="$SCOUTER_DIR/agent.host"
      SCOUTER_HOST_AGENT_CONF="$SCOUTER_HOST_AGENT_DIR/conf/scouter.conf"

      mkdir -p $SCOUTER_DIR
      wget $SCOUTER_URL -O /tmp/scouter-all.tar.gz
      tar -xzf /tmp/scouter-all.tar.gz -C $SCOUTER_DIR --strip-components=1

      cat <<EOF > $SCOUTER_HOST_AGENT_CONF
      net_collector_ip={스카우터 서버(collector) 의 고정 ip}
      net_collector_udp_port=6100
      net_collector_tcp_port=6100
      hook_profile_fullstack_method_sampling_interval_ms=10000
      hook_args_sampling_interval_ms=10000
      hook_dbsql_enabled=true
      hook_http_enabled=true
      hook_thread_enabled=true
      hook_profile_fullstack_method_sampling_enabled=true
      hook_args_enabled=true
      EOF

      # Ensure the script has execution permissions
      chmod +x $SCOUTER_HOST_AGENT_DIR/host.sh

      cd $SCOUTER_HOST_AGENT_DIR
      sed -i 's/^JAVA_OPTS=.*$/JAVA_OPTS="-Xmx256m"/' ./host.sh
      nohup ./host.sh start > /dev/null 2>&1 &

container_commands:
  01_run_scouter_install_script:
    command: sudo /opt/elasticbeanstalk/hooks/appdeploy/pre/03-install-scouter-host-agent.sh

최종 코드

잘 된다....... 해결!

5. 스카우터 자바 에이전트 도입

04-install-scouter-java-agent.config

files:
  "/opt/elasticbeanstalk/hooks/appdeploy/pre/04-install-scouter-java-agent.sh":
    mode: "000755"
    owner: root
    group: root
    content: |
      #!/bin/bash

      # Variables
      SCOUTER_VERSION="2.20.0"
      SCOUTER_AGENT_DIR="/opt/scouter"
      SCOUTER_JAVA_AGENT_URL="https://github.com/scouter-project/scouter/releases/download/v$SCOUTER_VERSION/scouter-agent-java-$SCOUTER_VERSION.tar.gz"
      SCOUTER_JAVA_AGENT_CONF="/opt/scouter/agent.java/conf/scouter.conf"

      # Download and extract Scouter agent
      # Configure Scouter agent
      cat <<EOF > $SCOUTER_JAVA_AGENT_CONF
      obj_name=java
      net_collector_ip=3.36.225.29
      net_collector_udp_port=6100
      net_collector_tcp_port=6100
      EOF

container_commands:
  01_run_scouter_install_script:
    command: sudo /opt/elasticbeanstalk/hooks/appdeploy/pre/04-install-scouter-java-agent.sh

스카우터 자바 에이전트는 호스트 에이전트와 다르게 독립적인 실행이 아니라 java 어플리케이션이 실행될 때 attach되서 실행된다 한다

실행 방법은

1. 톰캣의 시작 스크립트에 스카우터 정보 넣기
2. 어플리케이션 시작할때 scouter 명령행 옵션을 포함하여 jar 파일을 실행

2번으로 실행해보자 00-makeFiles.config

files:
    "/sbin/appstart":
        mode: "000755"
        owner: webapp
        group: webapp
        content: |
            #!/usr/bin/env bash
            JAR_PATH=/var/app/current/application.jar
            SCOUTER_AGENT_PATH=/opt/scouter/agent.java/scouter.agent.jar
            SCOUTER_CONF_PATH=/opt/scouter/agent.java/conf/scouter.conf

            # run app
            java -javaagent:$SCOUTER_AGENT_PATH \
             -Dscouter.config=$SCOUTER_CONF_PATH \
             -jar \
             -Dspring.profiles.active=prod -Dfile.encoding=UTF-8 -jar $JAR_PATH

스카우터 자바 에이전트의 obj_name을 넣어줄 수 있지만 안해도 돌아가긴한다 그냥 패스!

굳

참고자료 https://gunsdevlog.blogspot.com/2017/07/scouter-apm-1.html https://theholyjava.wordpress.com/2015/07/29/fixing-a-mysterious-ebextensions-command-time-out-aws-elastic-beanstalk/ https://github.com/scouter-project/scouter/blob/master/README_kr.md

1. 우리는 jwt를 사용해서 인가를 하고 interceptor에서 한다 그래서 WebConfig.class에서

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(tokenInterceptor)
                .order(1)
                .addPathPatterns("/v1/registration-step")
                .addPathPatterns("/v1/male/career-image")
                .addPathPatterns("/v1/optional-certification")
                .addPathPatterns("/v1/payment")
                .addPathPatterns("/v1/female/career-image")
                .addPathPatterns("/v1/univ-image")
                .addPathPatterns("/v1/income-image")
                .addPathPatterns("/v1/complete/optional-certification")
                .addPathPatterns("/v1/user-image");
   
        registry.addInterceptor(apiThrottlingInterceptor)
                .order(2)
                .addPathPatterns("/**");
    }

   이렇게 하나하나 인가가 필요한 url을 달아서 InterceptorRegistry를 사용하여 인터셉터를 추가한다

이렇게 하니까 각 패키지에 흩어져있는 controller 메서드 하나하나 WebConfig를 통해서 인가를 관리해야 한다 -> 귀찮고 유지보수 않좋고 휴먼 에러 나기 참 쉽다

@Slf4j
@RequiredArgsConstructor
@Component
public class TokenInterceptor implements HandlerInterceptor {

    private final JwtProvider jwtTokenProvider;

    @Override
    // 컨트롤러 호출전에 호출
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        if (request.getMethod().equals(HttpMethod.OPTIONS.name())) {  //preflight 통과하도록 설정
            return true;
        }
        String authorizationHeader = request.getHeader(HttpHeaders.AUTHORIZATION);
        parseTokenAndTransferUserId(request, authorizationHeader);
        return true;
    }

기존에는 이렇게 구성했다

2. jwt에서 유저아이디 꺼내오는것도

   
    private void parseTokenAndTransferUserId(HttpServletRequest request, String authorizationHeader) {
        HashMap<String, Object> parseJwtTokenMap = jwtTokenProvider.parseJwtToken(request ,authorizationHeader);
        Long userId = getUserIdFromToken(parseJwtTokenMap);
        request.setAttribute("userId", userId);
    }

   이렇게 하고 controller에서 @RequestAttribute Long userId 를 매개변수로 받아서 사용했는데 이것도 직관적이지 않다

2. 문제 해결

1. 모든 인터셉터를 열어두고
2. Authenticated.class를 커스텀 어노테이션으로 만들고
3. handlerMethod.hasMethodAnnotation(Authenticated.class)를 사용해서 컨트롤러 위에 @Authenticated 가 있으면 인가를 처리하도록 해보자

@Configuration
@RequiredArgsConstructor
public class WebConfig implements WebMvcConfigurer {

    @Value("${cors.allowed.origins:}")
    private String[] ALLOWED_CORS_URLS;
    private final TokenInterceptor tokenInterceptor;
    private final AdminTokenInterceptor adminTokenInterceptor;
    private final ApiThrottlingInterceptor apiThrottlingInterceptor;
    private final UserIdentifierArgumentResolver userIdentifierArgumentResolver;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(tokenInterceptor)
                .order(1);

        registry.addInterceptor(adminTokenInterceptor)
                .order(2);

        registry.addInterceptor(apiThrottlingInterceptor)
                .order(3);
    }

이런식으로 일단 모든 인터셉터를 열어두고

@RequiredArgsConstructor
@Component
public class TokenInterceptor implements HandlerInterceptor {

    private final AuthService authService;

    @Override
    public boolean preHandle(final HttpServletRequest request,
                             final HttpServletResponse response,
                             final Object handler) {
        if (request.getMethod().equals(HttpMethod.OPTIONS.name())) {  //preflight 통과하도록 설정
            return true;
        }

        if (!(handler instanceof final HandlerMethod handlerMethod)) { // handler가 컨트롤러인지 확인, mvc에서 정적 파일 요청도 가능 -> 컨트롤러가 아니면 뒤의 로직 스킵
            return true;
        }

        if (handlerMethod.hasMethodAnnotation(Authenticated.class)) { // 컨트롤러 메서드가 인증처리가 필요한 메서드인 경우
            String token = authService.getAuthorizationToken(request.getHeader("Authorization"));
            if (!authService.isValidToken(token)) { // 토큰 유효성 검사
                response.setStatus(HttpServletResponse.SC_UNAUTHORIZED); // 401 상태 코드 설정
                return false;
            }
        }

        return true;
    }

}

이런식으로 인가를 했다 토큰에서 페이로드 꺼내오는 로직은 알규먼트 리졸버에서 처리하도록 하고 일단 검증은 이렇게 했다

@Component
@RequiredArgsConstructor
public class UserIdentifierArgumentResolver implements HandlerMethodArgumentResolver {

    private final AuthService authService;

    @Override
    public boolean supportsParameter(final MethodParameter parameter) {
        if (!parameter.hasMethodAnnotation(Authenticated.class)) {
            throw new AuthenticationException(ErrorCode.INVALID_AUTHENITCATED_METHOD, ErrorCode.INVALID_AUTHENITCATED_METHOD.getStatusMessage());
        }

        return parameter.getParameterType().equals(Long.class) &&
                parameter.hasParameterAnnotation(UserIdentifier.class);
    }

    @Override
    public Long resolveArgument(final MethodParameter parameter,
                                final ModelAndViewContainer mavContainer,
                                final NativeWebRequest webRequest,
                                final WebDataBinderFactory binderFactory) {
        final String authorizationHeader = webRequest.getHeader("Authorization");
        String authorizationToken = authService.getAuthorizationToken(authorizationHeader);
        return extractUserIdFromJwtToken(authorizationToken);
    }

알규먼트 리졸버도 Authenticated.class 커스텀 어노테이션 만들어서 작성했다

@Component
@RequiredArgsConstructor
public class UserIdentifierArgumentResolver implements HandlerMethodArgumentResolver {

    private final AuthService authService;

    @Override
    public boolean supportsParameter(final MethodParameter parameter) {
        if (!parameter.hasMethodAnnotation(Authenticated.class)) {    //
            throw new AuthenticationException(ErrorCode.INVALID_AUTHENITCATED_METHOD, ErrorCode.INVALID_AUTHENITCATED_METHOD.getStatusMessage());
        }

        return parameter.getParameterType().equals(Long.class) &&
                parameter.hasParameterAnnotation(UserIdentifier.class);
    }

    @Override
    public Long resolveArgument(final MethodParameter parameter,
                                final ModelAndViewContainer mavContainer,
                                final NativeWebRequest webRequest,
                                final WebDataBinderFactory binderFactory) {
        final String authorizationHeader = webRequest.getHeader("Authorization");
        String authorizationToken = authService.getAuthorizationToken(authorizationHeader);
        return extractUserIdFromJwtToken(authorizationToken);
    }

간단하게 코드 설명하면 Spring MVC는 MethodParameter 객체를 사용하여 컨트롤러 메서드의 각 매개변수에 대한 메타데이터를 제공한다 메서드에 @Authentication 있는지 확인하고 매개변수에 @UserIdentifier가 있는지 확인한다 그렇다면 supportsParameter가 true를 return

resolveArgument에서는 이제 토큰에서 페이로드 까서 반환한다 우리는 userId를 사용한다

한가지 방법이 더있다

@Getter
@AllArgsConstructor
public final class UserId {

    @ApiModelProperty(hidden = true)
    private final Long id;

}

리졸버에다가 인가가 끝나고 꺼내온 userId를 저기다가 담아서 controller에 던져주는거다 걍 다 똑같다 resolveArgument메서드에서 반환할때 Long이 아니라 UserId라는 객체 생성에서 반환해주면 된다 사용해보니가 controller에서 userId.getId()해줘야 되기도 하고 한다해도 아직까지는 크게 효과가 좋은 것 같지 않아서 패스

3. 결과

    @Operation(summary = "어쩌고 저쩌고")
    @PostMapping(value = "/v1/test")
    @Authenticated
    public ResponseEntity<HttpStatus> uploadUserImages(@UserIdentifier Long userId) {
        webRegisterService.test(userId);
        return ResponseEntity.ok().build();
    }

운영중인 코드는 좀 그래서 간단하게 만들어봤다 이렇게 하나만 보면 큰 의미가 없어보이는데 70개 api넘어가니까 확실히 관리하기 편리하다는 생각이 들었다 그리고 이제부터 인가가 필요한 api를 짤 때 WebConfig들어가서 url매핑해주는 일을 하지 않고 그냥 컨트롤러에 어노테이션 하나 박으면 된다

물론 모든 api가 인터셉터를 한번은 타서 성능상 약간은 손해인거 같긴 하지만 우리 서비스는 80퍼센트 이상이 인가를 필요로 하는 서비스라서 어쩌피 타야된다 아주 조금의 성능과 가독성, 유지보수성을 트레이드 오프 했다

우리 서비스는 카카오, 네이버 로그인을 지원한다 대부분의 소셜 로그인 써드파티 제공 업체는 회원가입(정보 이용 동의) -> 토큰 발급 -> 토큰을 기반으로 유저 정보 조회 -> 유저 정보 반환 형식이다 d

문제 상황

1. 요구사항에서 소셜 로그인 업체를 추가한다면, 코드 너무 많이 변경이 일어난다 이것을 추상화로 해결 해 보자

문제 해결

    @Operation(summary = "소셜 로그인")
    @PostMapping("/signup/social")
    public ResponseEntity<TokenResponse> socialLogin(@Valid @RequestBody SocialLoginRequest request) {
        LoginToken loginToken = userService.signupByThirdParty(request.toDomain());
        return ResponseEntity.status(HttpStatus.OK).body(new TokenResponse(loginToken));
    }

presentation layer도 서드파티 싹다 통합하고 싶지만 이건 회사마다 다르다

@Data
public class SocialLoginRequest {

    @Schema(description = "Oauth 서버에서 받아온 인가코드", example = "인가코드")
    @NotBlank
    private String code;

    /**
     * redirectUrl 은 인가코드를 받아올 redirectUrl을 의미하며 여기서  redirectUrl은 로그인시 요청한 redirectUrl과 동일한 값으로 받아와야함
     * 리다이렉트 유알엘을 받는 이유는 로컬, 배포 , 테스트 환경에서 유동적으로 실행할수있게 하기 위함임
     */
    @Schema(description = "로그인후 리다이렉트 받을 주소 경로", example = "<http://localhost:3000/login/kakaoLoginProcess>")
    private String redirectUrl;

    private ProviderType type;

    @Schema(description = "소셜 로그인 타입", example = "KAKAO")
    @NotNull
    private ProviderType providerType;

    public ThirdPartySignupInfo toDomain() {
        Map<String, String> propertiesValues = new HashMap<>();
        propertiesValues.put("code", code);
        propertiesValues.put("redirectUrl", redirectUrl);
        return new ThirdPartySignupInfo(type, propertiesValues);
    }

}

여기서 ProviderType을 클라이언트에게 받는다 그리고 그것을

@Getter
public class ThirdPartySignupInfo {
    private final ProviderType providerType;
    private final Map<String, String> propertiesValues;

    public ThirdPartySignupInfo(ProviderType providerType, Map<String, String> propertiesValues) {
        this.providerType = providerType;
        this.propertiesValues = propertiesValues;
    }
}

해당 객체에 넣는다 이제 여기서부터 모든 서드파티가 통합이다 그래서 Map을 사용해서 각 서드파티에 필요한 요소값들을 넣어주었다 참고로 카카오는 redirect url이 필요하고 네이버 로그인은 state가 필요하다

이제 sevice layer로 가보자

    public LoginToken signupByThirdParty(ThirdPartySignupInfo request) {
        ThirdPartyUserInfo userInfo = requestProviderIdFromThirdParty(request);
        boolean isNewUser = userAuthManager.registerIfNeed(userInfo, request.getProviderType());
        return tokenGenerator.generate(userInfo.getProviderId(), isNewUser);
    }

    private ThirdPartyUserInfo requestProviderIdFromThirdParty(ThirdPartySignupInfo request) {
        ThirdPartyAuthorizer authorizer = thirdPartyAuthorizerProvider.get(request.getProviderType());
        String accessToken = authorizer.getAccessToken(request);
        return authorizer.getUserInfo(accessToken);
    }

여기서 이제 가장 중요한건 requestProviderIdFromThirdParty메서드 이다

@Component
@AllArgsConstructor
public class ThirdPartyAuthorizerProvider {
    private final List<ThirdPartyAuthorizer> thirdPartyAuthorizers;

    public ThirdPartyAuthorizer get(ProviderType providerType) {
        return thirdPartyAuthorizers.stream()
                .filter(authorizer -> authorizer.getProviderType() == providerType)
                .findFirst()
                .orElseThrow(() -> new RuntimeException("해당하는 제공자가 없습니다."));
    }
}

public interface ThirdPartyAuthorizer {
    String getAccessToken(ThirdPartySignupInfo signupInfo);

    ThirdPartyUserInfo getUserInfo(String accessToken);

    ProviderType getProviderType();

}

여기 보면 ThirdPartyAuthorizer 라는 인터페이스를 만들어서 소셜 로그인 제공자를 위한 공통된 메서드들을 정의했다 결국엔 모든 네이버, 카카오, 구글 등이 OAuth서버로부터 엑세스 토큰을 얻고, 엑세스 토큰을 사용하여 사용자 정보를 반환하는 구조기 때문이다 getProviderType으로 해당 타입 제공자가 카카오인지, 네이버인지, 구글인지 알 수 있게 해놨다

ThirdPartyAuthorizerProvider는 주어진 ProviderType에 맞는 적절한 ThirdPartyAuthorizer를 반환한다

예를 들어, ProviderType.KAKAO가 주어지면 KakaoAuthorizer 객체를 반환하면 된다

@Component
@RequiredArgsConstructor
public class KakaoAuthorizer implements ThirdPartyAuthorizer {

    private final KakaoAuthClient kakaoAuthClient;
    private final KakaoApiClient kakaoApiClient;
    @Value("${kakao.client-id}")
    private String clientId;
    @Value("${kakao.client-secret}")
    private String client_secret;

    @Override
    public String getAccessToken(ThirdPartySignupInfo signupInfo) {
        Map<String, String> propertiesValues = signupInfo.getPropertiesValues();

        KakaoTokenResponse response = kakaoAuthClient.generateToken(
                "authorization_code",
                clientId,
                propertiesValues.get("redirectUrl"),
                propertiesValues.get("code"),
                client_secret
        );

        return response.getAccessToken();
    }

    @Override
    public ThirdPartyUserInfo getUserInfo(String accessToken) {
        KakaoUserInfo kakaoUserInfo = kakaoApiClient.getUserInfo(new BearerAuthHeader(accessToken).getAuthorization());
        return new ThirdPartyUserInfo(kakaoUserInfo.getId().toString(), kakaoUserInfo.getName(),
                kakaoUserInfo.getNickName(), kakaoUserInfo.getProfileImage(), kakaoUserInfo.getEmail(),
                kakaoUserInfo.getPhoneNumber(), kakaoUserInfo.getGender(), kakaoUserInfo.getBirthDay());
    }

    @Override
    public ProviderType getProviderType() {
        return ProviderType.KAKAO;
    }

}

이렇게 하면 된다 그러면 프론트가 KAKAO값을 주면 적절한 ThridPartyAuthorizer 타입의 변수인 KakaoAuthorizer객체를 참조할 수 있다 이러면 구체적인 구현 클래스에 의존하지 않고 인터페이스 타입으로 객체를 다룰 수 있다

서비스 코드인

    private ThirdPartyUserInfo requestProviderIdFromThirdParty(ThirdPartySignupInfo request) {
        ThirdPartyAuthorizer authorizer = thirdPartyAuthorizerProvider.get(request.getProviderType());
        String accessToken = authorizer.getAccessToken(request);
        retrun authorizer.getUserInfo(accessToken);
    }

여기서!

다시말해 ThirdPartyAuthorizer 인터페이스의 KakaoAuthorizer와 같은 구현체는 ThirdPartyAuthorizer 타입으로 참조될 수 있으며, service layer에서는 이를 통해 다양한 소셜 로그인들을 처리할 수 있다 (리스코프 치환 원칙)

그러면 여기서 네이버 로그인 추가하면 어떻게 될까?

결국엔 카카오랑 똑같이 signupByThirdParty메서드만 호출하고 카카오랑 다른 요소값만 Map형식으로 보내주면 된다

그리고 ProviderType에

public enum ProviderType implements EnumModel {

    KAKAO("kakao"),
    NAVER("naver"),

    NORMAL("일반회원가입");

    private final String value;

    ProviderType(String value) {
        this.value = value;
    }

    @Override
    public String getKey() {
        return name();
    }

    @Override
    public String getValue() {
        return value;
    }
}

네이버 추가해주고 ThirdPartyAuthorizer를 구현하는

@Component
@RequiredArgsConstructor
public class NaverAuthorizer implements ThirdPartyAuthorizer {

    private final NaverAuthClient naverAuthClient;
    private final NaverApiClient naverApiClient;
    @Value("${spring.oauth2.client.registration.naver.client-id}")
    private String clientId;
    @Value("${spring.oauth2.client.registration.naver.client-secret}")
    private String client_secret;

    @Override
    public String getAccessToken(ThirdPartySignupInfo signupInfo) {

        Map<String, String> propertiesValues = signupInfo.getPropertiesValues();

        NaverTokenResponse response = naverAuthClient.generateToken(
                "authorization_code",
                clientId,
                client_secret,
                propertiesValues.get("code"),
                propertiesValues.get("state")
        );

        return response.getAccess_token();
    }

    @Override
    public ThirdPartyUserInfo getUserInfo(String accessToken) {

        NaverUserInfo naverUserInfo = naverApiClient.getUserInfo(new BearerAuthHeader(accessToken).getAuthorization());

        return new ThirdPartyUserInfo(naverUserInfo.getId().toString(), naverUserInfo.getName(),
                naverUserInfo.getNickName(), naverUserInfo.getProfileImage(), naverUserInfo.getEmail(),
                naverUserInfo.getPhoneNumber(), naverUserInfo.getGender(), naverUserInfo.getBirthDay());
    }

    @Override
    public ProviderType getProviderType() {
        return ProviderType.NAVER;
    }

}

이친구만 추가해주면 끝이다 구글도 같은 방식으로 처리하면 된다

이렇게 기획에서 소셜이 추가되더라도 객체지향적으로 코드를 작성하면 도메인과 서비스 레이어의 코드를 1도 건들이지 않고 시스템은 견고하면서 확장이 가능하다

근데 뭐 소셜 로그인 추가가 자주 일어날라나..... 어떻게 생각해보면 자주 일어날거같고 어떻게 생각해보면 자주 일어나지 않을 것 같다.

그리고 객체지향적으로 잘 짠다는 것은 결국엔 인터페이스 활용도가 아닌가 라는 생각도 들었다

그리고 역시 단순 독서보다 직접 활용하는게 나한텐 맞다

라고 정리했었는데 구글 로그인이 스펙에 추가 되었다

@Component
@RequiredArgsConstructor
public class GoogleAuthorizer implements ThirdPartyAuthorizer {

    private final GoogleAuthClient googleAuthClient;
    private final GoogleApiClient googleApiClient;

    @Value("${google.client-id}")
    private String clientId;
    @Value("${google.client-secret}")
    private String client_secret;

    @Override
    public String getAccessToken(final ThirdPartySignupInfo signupInfo) {
        Map<String, String> propertiesValues = signupInfo.getPropertiesValues();
        String code = propertiesValues.get("code");
        String decodedCode = URLDecoder.decode(code, StandardCharsets.UTF_8); // 구글 oauth 서버로부터 받은 인가코드는 디코딩 한번 해줘야함
        GoogleTokenRequest googleTokenRequest = new GoogleTokenRequest(
                "authorization_code",
                clientId,
                propertiesValues.get("redirectUrl"),
                decodedCode,
                client_secret);

        GoogleTokenResponse response = googleAuthClient.generateToken(googleTokenRequest);
        return response.accessToken();
    }

    @Override
    public ThirdPartyUserInfo getUserInfo(final String accessToken) {
        GoogleUserInfo googleUserInfo = googleApiClient.getUserInfo(new BearerAuthHeader(accessToken).getAuthorization());

        return new ThirdPartyUserInfo(googleUserInfo.getId().toString(), googleUserInfo.getName(),
                googleUserInfo.getNickName(), googleUserInfo.getProfileImage(), googleUserInfo.getEmail(),
                googleUserInfo.getPhoneNumber(), googleUserInfo.getGender(), googleUserInfo.getBirthDay());
    }

    @Override
    public ProviderType getProviderType() {
        return ProviderType.GOOGLE;
    }

}

정말 간단하게 구글 로그인을 추가할 수 있었다 굳굳

우리 서비스의 1기 매칭 버전1에서 서로 매칭을 해주는 과정에서 생겼던 이슈이다 매칭 과정을 이렇다 0. 트랜잭션 시작

1. 남자 기준 여자와의 추천 데이터가 있는지 검증 -> 이미 있으면 예외
2. 여자 기준 남자와의 추천 데이터가 있는지 검증 -> 이미 있으면 예외
3. 남자 기준 여자와의 추천 데이터가 5개가 넘는지 검증 -> 넘으면 예외
4. 여자 기준 남자와의 추천 데이터가 5개가 넘는지 검증 -> 넘으면 예외
5. 남자 기준 여자와의 추가 추천 객체 생성
6. 여자 기준 남자와의 추가 추천 객체 생성
7. 5, 6 저장
8. 5를 기반으로 추가 추천 결제 객체 생성
9. 6를 기반으로 추가 추천 결제 객체 생성
10. 8, 9 저장
11. 남자에게 알림톡 쏘고 로그 저장
12. 여자에게 알림톡 쏘고 로그 저장
13. 트랜잭션 종료

문제점 1

매칭을 했는데 12번에서 알림톡 전송에 실패한 것이다..... 알림톡에 실패 로그가 찍혔는데 이게 싹다 롤백되서 로그가 찍혀도 누가누구랑 매칭을 돌리다가 실패했는지 찾기가 힘들고 심지어 11번 남성에게는 알림톡이 갔다.... 이미 간 알림톡은 롤백을 할 수 없다 그리고 알림톡전송이 실패했다 해도 매칭 데이터가 날라가버리는건 우리가 원하는 트랜잭션 롤백이 아니다

문제점 2

그리고 또 다른 문제는 알림톡을 보내는 과정이

1. ncloud에 알림톡을 보내는 api
2. ncloud에 알림톡이 잘 갔는지 확인하는 알림톡 결과 발송 api
3. 알림톡 결과 로그 저장 이렇게 두번을 보내야 한다 한번만 보내면 알림톡이 잘 갔는지 안갔는지 확인할 수 없다 생각보다 이게 오래걸린다 그동안 디비 커넥션을 계속 붙들고 있어서 매칭이 많이 일어나는 시간대에 간단한 api까지도 디비 커넥션을 취득하지 못해서 성능이 안좋아진다

문제점 3

또한 예전에 트랜잭션 전파 관리를 제대로 하지 않아 디비 커넥션이 계속 열려서 https://velog.io/@wellbeing-dough/MySQL-ALTER-TABLE-%EB%AA%85%EB%A0%B9%EC%96%B4-%EC%95%88%EB%90%98%EB%8A%94-%EB%AC%B8%EC%A0%9C 이런 이슈도 발생했다

결론 트랜잭션 관리를 확실하게 하자

해결

예전에 서비스 레이어의 세부 구현을 impl 레이어로 분리하여 리팩토링을 했었다 링크: https://velog.io/@wellbeing-dough/%EC%8A%A4%ED%94%84%EB%A7%81-%EB%B6%80%ED%8A%B8-%EC%84%9C%EB%B9%84%EC%8A%A4-%EB%A0%88%EC%9D%B4%EC%96%B4-%EB%A6%AC%ED%8E%99%ED%86%A0%EB%A7%81

이번에도 이 메서드를 예시를 들면

    @Transactional
    public void createPayBothAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        additionalRecommendationValidator.validIsAlreadyExistBothAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        additionalRecommendationValidator.validBothAdditionalRecommendCountOver(request.getUserId(), request.getPartnerUserId(), ADDITIONAL_RECOMMENDATION_COUNT);
        paymentAdditionalRecommendationManager.createBothPaymentAdditionalRecommendation(request);
        sendAlimTalkAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        sendAlimTalkAdditionalRecommendation(request.getPartnerUserId(), request.getUserId());
    }

@Component
@RequiredArgsConstructor
public class AdditionalRecommendationValidator {

    private final AdditionalRecommendationRepository additionalRecommendationRepository;
    private final AdditionalRecommendationReader additionalRecommendationReader;

    public void validBothAdditionalRecommendCountOver(Long userId, Long partnerUserId, Long additionalRecommendationCount) {
        Long userRecommendationCount = additionalRecommendationRepository.countByUserAndType(userId);
        Long partnerUserRecommendationCount = additionalRecommendationRepository.countByUserAndType(partnerUserId);

        if (userRecommendationCount >= additionalRecommendationCount || partnerUserRecommendationCount >= additionalRecommendationCount) {
            throw new AdditionalRecommendationExceedException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR.getStatusMessage()
            );
        }
    }

    public void validIsAlreadyExistBothAdditionalRecommendation(Long userId, Long partnerUserId) {
        if (additionalRecommendationReader.isAdditionalRecommendationExistWithPartnerUser(userId, partnerUserId) ||
                additionalRecommendationReader.isAdditionalRecommendationExistWithPartnerUser(partnerUserId, userId)) {
            throw new AdditionalRecommendationAlreadyExistException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR.getStatusMessage()
            );
        }
    }

}

@Component
@RequiredArgsConstructor
public class PaymentAdditionalRecommendationManager {

    private final AdditionalRecommendationRepository additionalRecommendationRepository;
    private final RecommendationPaymentRepository recommendationPaymentRepository;


    public void createBothPaymentAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        List<AdditionalRecommendation> bothAdditionalRecommendation = request.toPayBothAdditionalRecommendations();
        additionalRecommendationRepository.saveAll(bothAdditionalRecommendation);
        List<RecommendationPayment> bothRecommendationPayment = createBothRecommendationPayment(
                bothAdditionalRecommendation.get(0).getId(),
                bothAdditionalRecommendation.get(1).getId()
        );
        recommendationPaymentRepository.saveAll(bothRecommendationPayment);
    }

    private List<RecommendationPayment> createBothRecommendationPayment(Long recommendationId, Long anotherRecommendationId) {
        RecommendationPayment recommendationPaymentForUser = RecommendationPayment.builder()
                .recommendationId(recommendationId)
                .paymentStatus(RecommendationPaymentStatus.BEFORE_PAYMENT)
                .build();

        RecommendationPayment anotherRecommendationPaymentForUser = RecommendationPayment.builder()
                .recommendationId(anotherRecommendationId)
                .paymentStatus(RecommendationPaymentStatus.BEFORE_PAYMENT)
                .build();
        return Arrays.asList(recommendationPaymentForUser, anotherRecommendationPaymentForUser);
    }

}

    private void sendAlimTalkAdditional(Long userId, Long partnerUserId) {
        UserProfile userProfile = userProfileReader.readByUserId(userId);
        UserProfile partnerUserProfile = userProfileReader.readByUserId(partnerUserId);
        AlimTalkFeignRequest userAlimTalkFeignRequest = additionalRecommendationAlimTalkClassifier.classifyByEssentialRecommendation(userProfile, partnerUserProfile);
        alimTalkManager.sendAlimTalkAndSaveLog(userId, userAlimTalkFeignRequest);
    }

이렇게 모든 작업이 하나의 트랜잭션으로 묶여서 문제점 1, 2가 발생한다 그리고 아무생각없이 서비스 계층에 별생각 없이 당연하듯이 @Transactional을 박으니까 문제점 3이 발생한다

그래서 impl계층에 트랜잭션을 걸어서

1. 남녀 상호 추천데이터가 있는지 검증하는 트랜잭션 시작
2. 남녀 상호 추천 데이터가 있는지 검증 -> 이미 있으면 예외
3. 남녀 상호 추천데이터가 있는지 검증하는 트랜잭션 종료
4. 남녀 상호 추천 데이터가 5개가 넘는지 검증하는 트랜잭션 시작
5. 남녀 상호 추천 데이터가 5개가 넘는지 검증 -> 넘으면 예외
6. 남녀 상호 추천 데이터가 5개가 넘는지 검증하는 트랜잭션 종료
7. 남녀 상호 추천 데이터와 추천 결제 데이터 생성하는 트랜잭션 시작
8. 남녀 상호 추천 데이터와 추천 결제 데이터 생성
9. 남녀 상호 추천 데이터와 추천 결제 데이터 생성하는 트랜잭션 종료
10. 남자에게 알림톡 쏘고 로그 저장하는 트랜잭션 시작
11. 남자에게 알림톡 쏘고 로그 저장
12. 남자에게 알림톡 쏘고 로그 저장하는 트랜잭션 종료
13. 여자에게 알림톡 쏘고 로그 저장하는 트랜잭션 시작
14. 여자에게 알림톡 쏘고 로그 저장
15. 여자에게 알림톡 쏘고 로그 저장하는 트랜잭션 종료

이 플로우를 코드에 적용해보자

    public void createPayBothAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        additionalRecommendationValidator.validIsAlreadyExistBothAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        additionalRecommendationValidator.validBothAdditionalRecommendCountOver(request.getUserId(), request.getPartnerUserId(), ADDITIONAL_RECOMMENDATION_COUNT);
        paymentAdditionalRecommendationManager.createBothPaymentAdditionalRecommendation(request);
        sendAlimTalkAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        sendAlimTalkAdditionalRecommendation(request.getPartnerUserId(), request.getUserId());
    }

위 서비스 레이어에는 트랜잭션을 전부 제거 하였다

@Component
@RequiredArgsConstructor
@Transactional(readOnly = true)
public class AdditionalRecommendationValidator {

    private final AdditionalRecommendationRepository additionalRecommendationRepository;
    private final AdditionalRecommendationReader additionalRecommendationReader;


    public void validBothAdditionalRecommendCountOver(Long userId, Long partnerUserId, Long additionalRecommendationCount) {
        Long userRecommendationCount = additionalRecommendationRepository.countByUserAndType(userId);
        Long partnerUserRecommendationCount = additionalRecommendationRepository.countByUserAndType(partnerUserId);

        if (userRecommendationCount >= additionalRecommendationCount || partnerUserRecommendationCount >= additionalRecommendationCount) {
            throw new AdditionalRecommendationExceedException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR.getStatusMessage()
            );
        }
    }

    public void validIsAlreadyExistBothAdditionalRecommendation(Long userId, Long partnerUserId) {
        if (additionalRecommendationReader.isAdditionalRecommendationExistWithPartnerUser(userId, partnerUserId) ||
                additionalRecommendationReader.isAdditionalRecommendationExistWithPartnerUser(partnerUserId, userId)) {
            throw new AdditionalRecommendationAlreadyExistException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR.getStatusMessage()
            );
        }
    }

}

상세 구현 컴포넌트에 validator는 검증을 한다, 읽기만하지 쓰기작업은 하지 않는다 그래서 @Transactional(readOnly = true)을 클래스에 달았다

@Component
@RequiredArgsConstructor
@Transactional
public class PaymentAdditionalRecommendationManager {

    private final AdditionalRecommendationRepository additionalRecommendationRepository;
    private final RecommendationPaymentRepository recommendationPaymentRepository;


    public void createBothPaymentAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        List<AdditionalRecommendation> bothAdditionalRecommendation = request.toPayBothAdditionalRecommendations();
        additionalRecommendationRepository.saveAll(bothAdditionalRecommendation);
        List<RecommendationPayment> bothRecommendationPayment = createBothRecommendationPayment(
                bothAdditionalRecommendation.get(0).getId(),
                bothAdditionalRecommendation.get(1).getId()
        );
        recommendationPaymentRepository.saveAll(bothRecommendationPayment);
    }

    private List<RecommendationPayment> createBothRecommendationPayment(Long recommendationId, Long anotherRecommendationId) {
        RecommendationPayment recommendationPaymentForUser = RecommendationPayment.builder()
                .recommendationId(recommendationId)
                .paymentStatus(RecommendationPaymentStatus.BEFORE_PAYMENT)
                .build();

        RecommendationPayment anotherRecommendationPaymentForUser = RecommendationPayment.builder()
                .recommendationId(anotherRecommendationId)
                .paymentStatus(RecommendationPaymentStatus.BEFORE_PAYMENT)
                .build();
        return Arrays.asList(recommendationPaymentForUser, anotherRecommendationPaymentForUser);
    }

}

그리고 추천과 추천 결제 데이터 생성은 여기에 달았고 쓰는 작업을 하기 때문에 @Transactional을 달았다

이렇게 서비스 레이어에 트랜잭션을 안걸고 구현 컴포넌트에만 트랜잭션을 각각 달기로 정했다

이렇게 하면

1. 트랜잭션으로 묶여야 되는 것들은 트랜잭션으로 묶고 트랜잭션에 묶이지 않아야 될것들은 묶이지 않을 수 있다 -> 직접 트랜잭션을 확실하게 관리하여 잘못 롤백되거나 롤백이 안되는 이슈가 생기지 않는다
2. 알림톡을 쏘고 로그를 저장하는것도 따로 트랜잭션으로 묶어서 매칭 트래픽이 많이 발생하는 경우에 디비 커넥션에 영향을 주지 않는다
3. 트랜잭션 전파를 정확하게 관리하여 문제점 3같은 경우도 일어나지 않는다

또한 예상치 못하게 추가적인 장점이 있는데 서비스 레이어에 트랜잭션이 없다보니까 하나의 트랜잭션으로 묶여야 하는 기능들을 세부 구현 컴포넌트에서 같은 트랜잭션으로 묶어야 했다 이렇게 하다 보니까 서비스 레이어의 코드 부분들이 명확하게 분리되고 비즈니스 흐름이 더 명확해져 가독성이 좋아지고 책임을 더 잘 분리할 수 있게 되었다

ex)

    public void createPayBothAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        additionalRecommendationValidator.validIsAlreadyExistBothAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        additionalRecommendationValidator.validBothAdditionalRecommendCountOver(request.getUserId(), request.getPartnerUserId(), ADDITIONAL_RECOMMENDATION_COUNT);

        AdditionalRecommendation additionalRecommendation = request.toPayAdditionalRecommendation();
        AdditionalRecommendation anotherAdditionalRecommendation = request.toPayAdditionalAnotherRecommendation();
        additionalRecommendationWriter.write(additionalRecommendation);
        additionalRecommendationWriter.write(anotherAdditionalRecommendation);

        RecommendationPayment recommendationPayment = craeteRecommendationPayment(additionalRecommendation.getId());
        RecommendationPayment anotherRecommendationPayment = craeteRecommendationPayment(anotherAdditionalRecommendation.getId());
        recommendationPaymentWriter.write(recommendationPayment);
        recommendationPaymentWriter.write(anotherRecommendationPayment);

        sendAlimTalkAdditionalMatching(request.getUserId(), request.getPartnerUserId());
        sendAlimTalkAdditionalMatching(request.getPartnerUserId(), request.getUserId());
    }

이 코드에서 트랜잭션으로 무조건 묶여야 할 부분이 있다 추가 추천 객체를 생성하고 저장, 추가 추천 결제 객체를 생성하고 저장 하지만 이렇게하면 추가 추천 결제 객체를 생성하고 저장하다가 실패하면 추가 추천 객체가 롤백되지 않는다 이럼 안된다

    public void createPayBothAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        additionalRecommendationValidator.validIsAlreadyExistBothAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        additionalRecommendationValidator.validBothAdditionalRecommendCountOver(request.getUserId(), request.getPartnerUserId(), ADDITIONAL_RECOMMENDATION_COUNT);
        paymentAdditionalRecommendationManager.createBothPaymentAdditionalRecommendation(request);
        sendAlimTalkAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        sendAlimTalkAdditionalRecommendation(request.getPartnerUserId(), request.getUserId());
    }

이렇게 추가 추천을 생성하고 저장, 추가 추천 결제를 생성하고 저장 하는 과정을 하나의 세부 구현 컴포넌트로 분리하여 하나의 트랜잭션으로 묶으니 서비스 계층의 메서드가 더 좋아졌다

이렇게 트랜잭션을 잘게 쪼갰는데 api가 클라이언트에게 응답될 때 까지 영속성 컨텍스트를 유지하기 때문에 디비 커넥션 또한 계속 가지고 있다 우리 서비스는 알림톡을 보내고 받은 유저가 행위를 하는 시간이 몰려 있는(순간 트래픽)이 중요한 서비스이기 때문에 OSIV를 false로 해서 transaction이 끝나면 영속성 컨텍스트가 닫히게 하였다

이렇게 하면 지연로딩이 닫힐 수 있지만 우리는 연관관계를 맺는 것에 대한 엄격한 기준을 새워서 현재까지 연관관계를 걸지 않았다 그래서 문제가 없다(지연로딩 자체가 없다)

1. 하나의 서비스 레이어가 다른 모든 repository와 다른 서비스 레이어를 의존함 -> 스파게티 코드 이렇게 되자 서비스 클래스의 메서드 하나하나가 책임과 역할이 커지고 서비스 레이어의 클래스 하나를 변경하자 다른곳에서 에러가 터지는 이슈 발생
2. 서비스 레이어의 하나의 메서드 비즈니스 로직이 길어지면서 가독성이 떨어지고 책임과 역할이 커지고 유지보수가 힘들어지는 문제 발생

이게 전부 비즈니스 로직이라 하면 비즈니스 로직이라 할 수 있다 하지만 서비스 클래스가 이렇게되면 유지보수 하기도 너무 힘들고 서비스 클래스의 책임과 역할이 너무 커져버린다.... 학교 선배님들의 조언과 구글링을 해본 끝에 해결할 수 있었다 "서비스 클래스의 메서드는 영어를 조금 알면서 코딩을 아예 모르는 사람이 와서 봐도 대충 이게 뭐하는지 알아야 한다"

해결책

여러가지 알아보고 나서 아키텍쳐부터 뜯어 고치기로 했다 service 계층에서 직접적으로 레포지토리 호출을 하지 않고 그 사이에 implementation계층(이하 impl계층)이라는 컴포넌트를 만들었다 이는 service계층끼리 서로 의존하는 스파게티 코드를 막으면서 서비스 계층의 메서드 하나하나가 책임이 커지는걸 방지한다

기존 코드의 하나의 메서드만 가져와봤다

기존 코드는 이렇다

@RequiredArgsConstructor
@Service
@Slf4j
@Transactional(readOnly = true)
public class AdminRecommendationServiceTmp {

    private static final Long ADDITIONAL_RECOMMENDATION_COUNT = 5L;
    private final AlimTalkManager alimTalkManager;
    private final AdditionalRecommendationAlimTalkClassifier additionalRecommendationAlimTalkClassifier;

    private final UserProfileRepository userProfileRepository;
    private final AdditionalRecommendationRepository additionalRecommendationRepository;
    private final RecommendationRepository recommendationRepository;
    private final RecommendationPaymentRepository recommendationPaymentRepository;

    private void sendAlimTalkAdditionalMatching(Long userId, Long partnerUserId) {
        UserProfile userProfile = userProfileRepository.findByUserId(userId);
        UserProfile partnerUserProfile = userProfileRepository.findByUserId(partnerUserId);
        AlimTalkFeignRequest userAlimTalkFeignRequest = additionalRecommendationAlimTalkClassifier.classifyByEssentialRecommendation(userProfile, partnerUserProfile);
        alimTalkManager.sendAlimTalkAndSaveLog(userId, userAlimTalkFeignRequest);
    }

    @Transactional
    public void createPayBothAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        if (additionalRecommendationRepository.isAdditionalRecommendationExistWithPartnerUser(request.getUserId(), request.getPartnerUserId())) {
            throw new AdditionalRecommendationAlreadyExistException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR.getStatusMessage()
            );
        }
        if (additionalRecommendationRepository.isAdditionalRecommendationExistWithPartnerUser(request.getPartnerUserId(), request.getUserId())) {
            throw new AdditionalRecommendationAlreadyExistException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR.getStatusMessage()
            );
        }
        Long recommendationCount = additionalRecommendationRepository.countByUserAndType(request.getUserId());
        if (recommendationCount >= ADDITIONAL_RECOMMENDATION_COUNT) {
            throw new AdditionalRecommendationExceedException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR.getStatusMessage()
            );
        }
        Long partnerRecommendationCount = additionalRecommendationRepository.countByUserAndType(request.getPartnerUserId());
        if (partnerRecommendationCount >= ADDITIONAL_RECOMMENDATION_COUNT) {
            throw new AdditionalRecommendationExceedException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR.getStatusMessage()
            );
        }
        AdditionalRecommendation additionalRecommendation = request.toPayAdditionalRecommendation();
        AdditionalRecommendation anotherAdditionalRecommendation = request.toPayAdditionalAnotherRecommendation();
        additionalRecommendationRepository.save(additionalRecommendation);
        additionalRecommendationRepository.save(anotherAdditionalRecommendation);

        RecommendationPayment recommendationPayment = craeteRecommendationPayment(additionalRecommendation.getId());
        RecommendationPayment anotherRecommendationPayment = craeteRecommendationPayment(anotherAdditionalRecommendation.getId());
        recommendationPaymentRepository.save(recommendationPayment);
        recommendationPaymentRepository.save(anotherRecommendationPayment);

        sendAlimTalkAdditionalMatching(request.getUserId(), request.getPartnerUserId());
        sendAlimTalkAdditionalMatching(request.getPartnerUserId(), request.getUserId());
    }

    private RecommendationPayment createRecommendationPayment(Long recommendationId) {
        return RecommendationPayment.builder()
                .recommendationId(recommendationId)
                .paymentStatus(RecommendationPaymentStatus.BEFORE_PAYMENT)
                .build();
    }

딱봐도 너무 복잡하다 이 코드를 보면 당최 무엇을 하려는지 파악이 되질 않는다

1. 남자 기준 여자와의 추천 데이터가 있는지 검증 -> 이미 있으면 예외
2. 여자 기준 남자와의 추천 데이터가 있는지 검증 -> 이미 있으면 예외
3. 남자 기준 여자와의 추천 데이터가 5개가 넘는지 검증 -> 넘으면 예외
4. 여자 기준 남자와의 추천 데이터가 5개가 넘는지 검증 -> 넘으면 예외
5. 남자 기준 여자와의 추가 추천 객체 생성
6. 여자 기준 남자와의 추가 추천 객체 생성
7. 5, 6 저장
8. 5를 기반으로 추가 추천 결제 객체 생성
9. 6를 기반으로 추가 추천 결제 객체 생성
10. 8, 9 저장
11. 남자에게 알림톡 쏘고 로그 저장
12. 여자에게 알림톡 쏘고 로그 저장

이 모든 것이 비즈니스 로직이긴 하지만, 서비스 클래스가 이렇게 되면 유지보수가 너무 힘들어진다 이 비즈니스 로직중에서 상세 구현 부분을 impl계층에 넣어서 처리하고 repository를 직접 호출하는 방식또한 implementation계층

일단 서비스 코드에서 모든 repository계층을 주입받는 방식을 제외하고 사이에 impl계층에 넣어서 해당 계층의 메서드에 상세 구현을 책임지게 해보자

@Transactional
    public void createPayBothAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        additionalRecommendationValidator.validIsAlreadyExistBothAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        additionalRecommendationValidator.validBothAdditionalRecommendCountOver(request.getUserId(), request.getPartnerUserId(), ADDITIONAL_RECOMMENDATION_COUNT);
        paymentAdditionalRecommendationManager.createBothPaymentAdditionalRecommendation(request);
        sendAlimTalkAdditionalRecommendation(request.getUserId(), request.getPartnerUserId());
        sendAlimTalkAdditionalRecommendation(request.getPartnerUserId(), request.getUserId());
    }

이 메서드를 이렇게 줄였다 학교 선배님의 말씀인 "서비스 클래스의 메서드는 영어를 조금 알면서 코딩을 아예 모르는 사람이 와서 봐도 대충 이게 뭐하는지 알아야 한다" 라는 말에 어느정도 부합하는거같다 validIsAlreadyExistBothAdditionalRecommendation는 뭔진 모르겠지만 대충 상호 추가 추천이 이미 존재하는지 검증하는구나 validBothAdditionalRecommendCountOver는 뭔진 모르겠지만 상호 추가 추천의 개수가 오버됬는지 검증하는구나 createBothPaymentAdditionalRecommendation는 뭔진 모르겠지만 상호 유료 추가 추천을 생성하는 구나 sendAlimTalkAdditionalMatching 뭔진 모르겠지만 추가 추천 알림톡을 발송하는 구나

이제 impl계층의 상세 구현을 봐보자

@Component
@RequiredArgsConstructor
public class AdditionalRecommendationValidator {

    private final AdditionalRecommendationRepository additionalRecommendationRepository;
    private final AdditionalRecommendationReader additionalRecommendationReader;

    public void validBothAdditionalRecommendCountOver(Long userId, Long partnerUserId, Long additionalRecommendationCount) {
        Long userRecommendationCount = additionalRecommendationRepository.countByUserAndType(userId);
        Long partnerUserRecommendationCount = additionalRecommendationRepository.countByUserAndType(partnerUserId);

        if (userRecommendationCount >= additionalRecommendationCount || partnerUserRecommendationCount >= additionalRecommendationCount) {
            throw new AdditionalRecommendationExceedException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_EXCEED_ERROR.getStatusMessage()
            );
        }
    }

    public void validIsAlreadyExistBothAdditionalRecommendation(Long userId, Long partnerUserId) {
        if (additionalRecommendationReader.isAdditionalRecommendationExistWithPartnerUser(userId, partnerUserId) ||
                additionalRecommendationReader.isAdditionalRecommendationExistWithPartnerUser(partnerUserId, userId)) {
            throw new AdditionalRecommendationAlreadyExistException(
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR,
                    ErrorCode.ADDITIONAL_RECOMMENDATION_ALREADY_EXIST_ERROR.getStatusMessage()
            );
        }
    }

}

이렇게 상호 추천이 존재하는지, 개수를 넘어섰는지 확인한다

@Component
@RequiredArgsConstructor
public class PaymentAdditionalRecommendationManager {

    private final AdditionalRecommendationRepository additionalRecommendationRepository;
    private final RecommendationPaymentRepository recommendationPaymentRepository;


    public void createBothPaymentAdditionalRecommendation(CreatePayBothAdditionalRecommendationRequest request) {
        List<AdditionalRecommendation> bothAdditionalRecommendation = request.toPayBothAdditionalRecommendations();
        additionalRecommendationRepository.saveAll(bothAdditionalRecommendation);
        List<RecommendationPayment> bothRecommendationPayment = createBothRecommendationPayment(
                bothAdditionalRecommendation.get(0).getId(),
                bothAdditionalRecommendation.get(1).getId()
        );
        recommendationPaymentRepository.saveAll(bothRecommendationPayment);
    }

    private List<RecommendationPayment> createBothRecommendationPayment(Long recommendationId, Long anotherRecommendationId) {
        RecommendationPayment recommendationPaymentForUser = RecommendationPayment.builder()
                .recommendationId(recommendationId)
                .paymentStatus(RecommendationPaymentStatus.BEFORE_PAYMENT)
                .build();

        RecommendationPayment anotherRecommendationPaymentForUser = RecommendationPayment.builder()
                .recommendationId(anotherRecommendationId)
                .paymentStatus(RecommendationPaymentStatus.BEFORE_PAYMENT)
                .build();
        return Arrays.asList(recommendationPaymentForUser, anotherRecommendationPaymentForUser);
    }

}

유료 상호 추천에 대해서 생성하고 저장한다

가독성 향상: 코드를 이해하기 쉬워졌으며, 각 메서드와 클래스의 책임이 명확해졌다 유지보수성 개선: 특정 기능을 수정해야 할 경우, 해당 기능과 관련된 구체적인 구현 부분만을 찾아 수정하기 쉬워졌다 확장성 강화: 새로운 기능 추가가 용이해졌으며, 기존 코드에 영향을 미치지 않고 확장할 수 있는 구조가 되엇다 책임의 명확한 분리: 각 계층과 컴포넌트가 자신의 책임에 집중할 수 있게 되었다

또한 서로 엮이는 스파게티 코드를 만들지 않기 위해서

1. presentation layer
2. service layer
3. impl layer
4. data access layer 이 참조의 흐름에서 역방향으로 참조되지 않고, 하위 레이어를 건너 뛰지 말고, 동일 레이어 간의 참조를 하지 않기로 했다

그리고 사이드 이펙트를 줄이기 위해서 슈퍼 객체, 슈퍼 메서드 (슈퍼는 이제 책임과 역할이 몰리는)를 만들지 않고 정말 100퍼센트 같은 기능을 한다는 가정하에 재사용 하고, 그걸 정확하게 파악하기 위해서 메서드 이름이 조금 길어지더라도 메서드 명명을 정말 확실하게 하기로 했다

급하게 요구사항이 변경되었고 개발일정을 넉넉하게 말씀드렸다 요구사항이 변경됨에 따라 ALTER TABLE을 할 일이 생겨서 기능 구현을 어느정도 끝마치고 RDS에 ALTER TABLE을 했는데 무슨 해당 쿼리가 끝나지 않고 계속 돌아가는 것이였다

근데 또 DML작업은 가능한데 DDL작업만 안된다.... 사실 해당 테이블에 데이터가 아예 없어서 DROP하고 다시 만들생각도 했었는데 그것도 안됬다...

2. 해결

여러가지 문제가 있을 수 있지만 알아본 결과

1. 대용량 데이터라서 시간이 오래걸릴 수 있다 -> 대용량 데이터도 아닐 뿐더러 얼마나 돌아가는지 봤는데 5분을 넘겼다
2. 테이블 메타데이터 락으로 다른 세션이 사용중이라 안될 수 있다 -> 그럴싸하다
3. 서버 부하 -> 새벽 4시고 하고 지금은 광고를 끈 상태라서 부하가 심하지 않다
4. 하드웨어 문제 -> 다른 테이블은 잘만 된다
5. 다른 트랜잭션이 제대로 종료되지 않아서 -> 그럴싸하다

그래서 2번과 5번을 위주로 알아보기로 하고 2번부터 보면

SHOW PROCESSLIST;

명령어를 사용하여 출력의 State열이 "Waiting for table metadata lock" 과 같은 메시지를 보여주면 나의 프로세스가 락 떄문에 대기 중인지 알 수 있다

하지만 그런 프로세스는 없었다....

5번으로 넘어가 보면

SELECT * 
FROM information_schema.innodb_trx;

해당 명령어로 trx_state 칼럼을 확인하여 트랜잭션이 "RUNNING" 상태인지 혹은 "LOCK WAIT" 상태인지 확인할 수 있다

결과

304044843212120|RUNNING|....생략|REPEATABLE READ|....생략|

찾았다....

이제 여기서 trx_mysql_thread_id 칼럼에 있는 id를 기반으로

KILL [id];

해주면 된다 하지만 이 방법은 돌아가고 있는 트랜잭션의 변경사항이 전부 롤백되기 때문에 예상치 못한 부작용이 발생할 수 있다

해당 트랜잭션을 강제 종료하니까 잘 된다....

3. 왜지?

데이터 변경하는 트랜잭션이 종료되지 않은 이유는 여러가지가 있을 수 있다

1. @Transactional: 어노테이션의 경계가 제대로 설정되지 않거나, 트랜잭션 전파가 적절하지 못하거나, 예외처리가 적절하지 못하면 트랜잭션이 종료되지 않을 수 있다
2. 예외처리 누락: @Transactional의 어노테이션은 예외가 발생하면 롤백된다 하지만 try catch 블록에서 예외를 명시적으로 다시 던지는 경우 트랜잭션이 종료되지 않을 수 있다

이정도 인거 같다

확인 결과 해당 테이블에 트랜잭션을 거는 api가 단하나 있었는데 해당 api가 디비에 저장하고 알림톡을 보내는 과정에서 트랜잭션 경계를 명확하게 하지 않아서 발생한 이슈였다

트랜잭션 처리가 명확한 코드를 작성해야겠다

• study hub 서버 개발을 하면서 유저가 스터디에 참여를 하면 스터디 게시글(studyPost)에 remainingSeat(잔여석)이 -1되어야 한다
• 멀티 쓰레드나 서버가 여러대일 경우에는 studyPost의 동시성 처리를 해야 동시에 참여한 유저가 발생할 경우 잔여석 감소 로직이 잘 돌아간다

StudyPostApplyEvenPublisher.class

@Component
@RequiredArgsConstructor
public class StudyPostApplyEventPublisher {

    private final StudyPostRepository studyPostRepository;

    @Transactional
    public void acceptApplyEventPublish(Long studyId) {
        StudyPostEntity studyPost = studyPostRepository.findByStudyId(studyId).orElseThrow(PostNotFoundException::new);
        studyPost.decreaseRemainingSeat();
        studyPostRepository.save(studyPost);
    }
}

StudyPostEntity.class

    public void decreaseRemainingSeat() {
        if (this.remainingSeat - 1 < 0) {
            throw new NoRemainingSeatsException();
        }
        this.remainingSeat -= 1;
    }

StudyPostApplyEventPublisherTest.class

    @Test
    void 동시에_100개의_요청의_스터디_지원서가_수락되면_게시글의_잔여석이_줄어든다() throws InterruptedException {
        // given
        Long postedUserId = 1L;
        StudyPostEntity post = StudyPostEntityFixture.SQLD.studyPostEntity_생성(postedUserId);
        StudyPostEntity savedPost = studyPostRepository.saveAndFlush(post);
        // when
        int threadCount = 100;
        ExecutorService executorService = newFixedThreadPool(32);
        CountDownLatch latch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            executorService.submit(() -> {
                try {
                    studyPostApplyEventPublisher.acceptApplyEventPublish(post.getStudyId());
                } finally {
                    latch.countDown();
                }
            });
        }
        latch.await();

        // then
        StudyPostEntity actualStudyPost = studyPostRepository.findById(savedPost.getId()).orElseThrow();
        assertEquals(0, actualStudyPost.getRemainingSeat());

    }

테스트 코드는 잔여석이 100명으로 지정된 스터디를 100명이 동시에 지원하는 경우의 테스트 코드이다 ExecutorService는 비동기로 실행하는 작업을 단순화하여 사용할 수 있게 해주는 자바 api이다 CountDownLatch는 100개의 요청이 끝날떄까지 기다려야 하므로 countDownLatch사용했다, 다른 쓰레드에서 수행중인 작업이 완료될때 까지 대기할 수 있도록 해주는 클래스이다

잔여석이 0개여야하는데 88개의 잔여석이 남았다

왜 이런 문제가 발생할까?

간단하게 쓰레드가 두개라고 가정해보자

쓰레드 1이 데이터를 가져가서 데이터를 갱신한 값을 쓰레드 2가 가져가서 갱신한다고 생각했지만 실제로는 쓰레드1이 데이터를 가져가서 갱신하기 전에 쓰레드 2가 데이터를 가져간다 하지만 둘다 재고가 100인 상태에서 1을 줄인 값을 갱신하기 때문에 갱신이 누락된다 두개 이상의 쓰레드가 공유 데이터에 접근할 수 있고 동시에 값을 변경하려 해서 동시성 문제가 발생한 것이다

2. 해결책

하나의 쓰레드가 데이터를 가져가고 갱신한 후에 다른 쓰레드가 접근할 수 있게 해야 한다

1. Synchronized로 해결

자바에서는 Synchronized 사용하면 해당 메서드를 한개의 쓰레드만 접근할 수 있게 한다

    @Transactional
    public synchronized void acceptApplyEventPublish(Long studyId) {
        StudyPostEntity studyPost = studyPostRepository.findByStudyId(studyId).orElseThrow(PostNotFoundException::new);
        studyPost.decreaseRemainingSeat();
        studyPostRepository.save(studyPost);
    }

그대로 테스트 코드 돌려보면 안쓸때보다 잔여석은 줄었지만 그래도 잘 동작하지 않는다 이유는 @Transactional어노테이션 때문이다 동작 과정을 단순하게 풀면

startTansaction();
acceptApplyEventPublish(Long studyId);
commitTransaction();

이렇게 @Transactional이 붙은 메서드가 호출되면, Spring은 CGLIB으로 런타임에 동적으로 프록시 객체를 생성한고 그 객체는 원본 객체를 감싸는 형태이다

acceptApplyEventPublish메서드가 미처 트랜잭션에 커밋되지 못하고 undo로그나 트랜잭션 로그에 있을때(잔여석 - 1 을 레코드에 반영하기 직전)에 다른 쓰레드의 acceptApplyEventPublish메서드가 실행 되기 때문이다

해결책은 acceptApplyEventPublish메서드에 @Transactional을 제거하거나 @Transactional을 제거하고 implementations 계층에 decrease()메서드를 추가하고 거기에 @Transactional을 박아도되지만 이거는 단일 서버에서 가능하고 우리는 오토스케일링을 사용하여 부하가 발생할 시 최대 2개까지 ec2를 늘릴 수 있기 때문에 적절한 해결 방법이 아니다

2. Pessimistic Lock

실제로 데이터에 베타lock을 걸어서 다른 트랜잭션에서 lock이 해재되기 전까지 데이터에 접근할 수 없게 된다 하지만 데드락이 걸릴 수 있기 때문에 조심해야 한다

쓰레드1에서 락을 걸고 데이터를 가져가고 점유중이라서 쓰레드2는 대기하게 된다 쓰레드1이 update가 끝나면 쓰레드2가 데이터를 가져가고 점유한다

    @Transactional
    public void acceptApplyEventPublish(Long studyId) {
        StudyPostEntity studyPost = studyPostRepository.findByIdWithPessimisticLock(studyId).orElseThrow(PostNotFoundException::new);
        studyPost.decreaseRemainingSeat();
        studyPostRepository.save(studyPost);
    }

    @Lock(LockModeType.PESSIMISTIC_WRITE)
    @Query("select s from StudyPostEntity s where s.studyId = :studyId")
    Optional<StudyPostEntity> findByIdWithPessimisticLock(@Param("studyId") Long studyId);

이렇게 하면 된다

테스트 코드도 잘 통과한다 하지만 Pessimisitic lock은 별도의 lock을 걸기 때문에 성능 감소가 일어날 수 있다

3. Optimistic Lock

lock을 사용하지 않고 version을 이용하여 정합성을 보장한다 스키마에 version을 추가한다 간단한 동작 방식은

이렇게 버전을 추가해서 update할 때마다 1씩 버전을 올려주는데 쓰레드2의 update문은 실패한다 왜? 레코드의 버전은 2인데 where문에 버전 1 이니까 그럼 쓰레드2의 업데이트는 어칼건데? 별도의 로직을 작성해주어야 한다

    @Lock(LockModeType.OPTIMISTIC)
    @Query("select s from StudyPostEntity s where s.studyId = :studyId")
    Optional<StudyPostEntity> findByIdWithOptimisticLock(@Param("studyId") Long studyId);

        while (true) {
            try{
                studyPostApplyEventPublisher.acceptApplyEventPublish(study.getId());
                break;
            } catch (Exception e) {
                Thread.sleep(50);
            }
        }

이렇게 version이 맞지않으면 예외에서 쓰레드를 잠시 멈추고 다시 시도하게 하는 로직을 추가했다 성공했다 하지만 Optimisitc lock은 별도의 lock이 없어서 Pessimsitic lock보다 성능상 이점이 있을 수 있지만 업데이트가 실패했을때 로직을 따로 작성해야 하고 충돌이 빈번하면 Pessimsitic lock보다 성능이 떨어진다 또한 쓰레드가 메서드를 실행하는 순서가 아닌 재시도 로직의 성공 순서가 되어서 5개의 잔여석에 6명이 지원하면 요청한 순서가 아닌 재시도 성공 순서라서 공정하지 않다

4. Named Lock

이름을 갖는 metadata lock을 사용한다 트랜잭션이 종료될 때 lock이 자동으로 해제하지 않는다 Pessimistic락이랑 비슷하지만 Pessimisitic lock은 row나 table단위로 걸지만 named lock은 메타 데이터에 건다

mySQL에서는 getLock명령어로 lock을 획득하고 release명령어로 lock을 해제한다

쉽게 말해서 이전에는 StudyPost의 레코드에 락을 걸었다면 별도의 metadata에 lock을 거는거다

LockRepository로 사용해보자

public interface LockRepository extends JpaRepository<StudyPostEntity, Long> {

    @Query(value = "select get_lock(:key, 3000)", nativeQuery = true)
    void getLock(String key);

    @Query(value = "select release_lock(:key)", nativeQuery = true)
    void releaseLock(String key);

}

편의성을 위해 StudyPostEntity에 native query를 사용했는데 제대로 사용하려면 별도의 jdbc를 사용해야 한다

        try {
            lockRepository.getLock(id.toString());
            studyPostApplyEventPublisher.acceptApplyEventPublish(study.getId());
        } finally {
            lockRepository.releaseLock(id.toString());
        }

이렇게 lockRepository에 명시적으로 lock을 걸고 끝나면 finally로 lock을 해제해주는 로직을 작성 해 준다

테스트 코드도 잘 돌아간다

주로 분산 락을 구현할때 사용한다 하지만 실제로 제대로 사용하려면 락 해제와 세션관리와 커넥션을 잘 관리해야 하고 같은 데이터소스를 사용하기 때문에 커넥션 풀 사이즈도 관리해야 한다 많이 복잡하다

3. 결론

1. Synchronized로 해결 -> 오토 스케일링으로 최대 ec2 두대라서 안됨
2. Pessimistic Lock -> 비즈니스 로직상 스터디헙에서 가장 중요한건 스터디 참여기 때문에 데이터의 정확성과 일곤성이 매우 중요해서 적합, 하지만 충돌 가능성이 그다지 높진 않아서 별도의 Lock을 걸기 때문에 성능이 그다지 좋지않음 Optimistic Lock이 더 좋을거같음
3. Optimistic Lock -> 스터디 참여 순서가 메서드를 실행하는 순서가 아닌 재시도 로직의 성공 순서가 되어서 5개의 잔여석에 6명이 지원하면 요청한 순서가 아닌 재시도 성공 순서라서 공정하지 않음
4. Named Lock -> 제대로 구현하려면 너무 복잡함(다음주 론칭임....) 같은 사유로 Lettuce, Redisson 도 안쓸 예정 Pessimistic Lock을 쓰자