최근 프로젝트 아키텍처를 변경하였다. 기존에는 어떤 단일 요청이 발생하였을 떄, 하나의 단일 서버에서 모든 과정을 순차적으로 관제하고 트랜잭션을 마감했다면, 바뀐 구조에서는 서버를 쪼개서 작업과정을 분산시켰다.

이로 인해 여러가지 고려해야 될 사항이 생겼는데, 그 과정을 공유해보자.

기존 작업 흐름

기존에는 클라이언트로부터 요청이 오면, 요청 스레드가 아닌 다른 스레드에게 작업을 넘기고 즉시 응답을 시켰다. 넘겨받은 스레드에서는 해당 작업이 순차적으로 이어졌으며, 최대 9번의 외부 API/SDK 호출과, 최대 4번의 파일 I/O 작업, 최대 5번의 DB I/O 작업을 하나의 트랜잭션으로 관리하고 있었다.

단일 요청에 대한 함수로 꽤나 부하가 걸리는 흐름이므로, 새로 바뀐 구조에서는 이 작업과정을 쪼개서 여러 대의 서버에 분산처리를 시킴과 동시에 병렬구조로 실행하는 걸로 방향성을 잡았다.

각 작업에 대한 수행을 비동기로 개별 서버에게 분산시키고, 각자 할당된 작업이 끝나면 DB에 관련결과를 업데이트한다. DB 해당 칼럼의 STEP이 일정단계를 초과했을 경우 상태를 완료처리한 걸로 간주한다.

JPA Dirty Checking X

해당 함수마다 각자의 스텝이 끝나면 DB에 상태를 반영시키도록 해야했는데 기존에는 JPA 의 변경감지 기능을 적극 사용했다. 다만 분산 서버 환경에서는 이 기능을 사용하기가 몹시 까다로워졌다. 각각의 서버마다 트랜잭션이 다르고, 병렬로 수행하다 보니, DB에서 불러온 엔티티의 필드가, 동시에 다른 트랜잭션 내에서 불러온 Entity의 필드를 업데이트했으면 반영하지 못하는 경우가 발생하는 것. 따라서 썡쿼리를 날리는 걸로 쇼부봤다.

fun increaseStep(id: Long): Int {

        val sql =
            """
            update 
            테이블
            set 
            sequence = sequence + 1, 
            status = case
                WHEN sequence >= ${finalSequence} THEN ?
                ELSE status
                end           
            where id = ?
            """.trimIndent()

        return jdbcTemplate.update(
            sql,
            Status.SUCCESS.name,
            id
        )
    }

대충 이런 식으로 쿼리를 해당 함수가 끝날때마다 호출하는 식으로 작업했다.
쌩쿼리를 날리는 게 영 찝찝하다면, QueryDsl을 사용할 수 도 있다. 그래도 타입안정성은 챙겨갈 수 있다. 아니다. QueryDsl을 쓰면 에러가 발생한다. 이거 나중에 정리해보겠다..

        val caseBuilder = CaseBuilder()
            .`when`(큐타입엔티티.sequence.goe(finalSequence))
            .then(Status.SUCCESS)
            .otherwise(큐타입엔티티.status)

        queryFactory
            .update(큐타입엔티티)
            .set(큐타입엔티티.sequence, 큐타입엔티티.sequence.add(1))
            .set(큐타입엔티티.status, caseBuilder)
            .where(
                큐타입엔티티.id.eq(id)
            )
            .execute()

JPA의 영속성 생명주기를 사용하는 것이 아니기 때문에 당연히 JPA의 여러가지 편의기능( EntityListner.. ) 등을 사용하지 못한다. 따라서 쿼리 내에서 자체적으로 case 절을 도입했다.

SSE Event => POLLING

기존에는 해당 작업이 모두 완료되었으면 다 끝났다는 신호를 Server Sent Event를 통해, 서버에서 단방향으로 클라이언트에게 쏴주곤 했다. 다만 역시 분산 환경에서 이 부분을 컨트롤하기가 몹시 까다로워졌다. 일단 클라이언트의 요청을 제일 앞단에서 받는 웹 서버도 로드밸런싱 되어있어서, 여러 대의 서버로 분산 처리될 뿐 아니라, 해당 작업들을 전파할 다른 서버들도 로드밸런싱 될 가능성이 있었다.

Server Sent Event 같은 경우, 클라이언트와(이 경우 브라우저) 커넥션을 해당 서버가 지속적으로 물고 있어야 되므로, 커넥션을 맺은 해당 서버의 메모리에서 메시지를 발행해야 된다. 따라서 클러스터링이 매우 힘든 구조이다.

최초에는 이 모든 서버를 메시지 큐 서버 (Amazon MQ) 에 묶고 브로드캐스팅 방식으로, 마무리 단계일 경우, 한 번의 Publish 로 모든 구독된 서버가 알람을 받고 클라이언트에게 모두 쏘는 형태로 생각을 해보았으나, 구현상 복잡도가 기능의 단순함에 비해 지나치게 과하다는 생각이 들어 그만뒀다.

해당 서버의 메모리에 저장된 Emitter를 끄집어낼 고유식별자를 어딘가 저장하고, 메시지를 발행할때, 로드 밸런싱되어있는 모든 서버에게 알리는 구조로 설계를 해야했는데, 애매한 것이 지금 서버는 이 Emitter를 구별할 식별자를 결정하기가 힘든 구조였다. 해당 서비스는 모든 유저, 즉 로그인 유저와 비로그인 유저 모두에게 Notify를 줘야 되는 형태였고, 따라서 비로그인 유저 같은 경우는 클라이언트에서 랜덤한 식별자를 생성해 서버에 전달해주면, 서버는 그걸 가지고 Emitter를 저장하는 형태였다.

이 비로그인 유저들의 식별자를 매번 DB에 저장하기도 애매하니, 계속 식별자를 모든 서버에게 넘기면서 유지를 해야 되는데, 이 또한 귀찮은 과정이며, 이렇게 넘겨받은 EmitterId를 가지고 개별 서버 모두 메시징 큐 서버에 묶어놓고 pub 요청을 해야 되는데, 너무나 과하다는 생각이 들었다. 더군다나, Status 가 성공이란 걸 판단할려면, 매번 업데이트 후 검색하는 쿼리를 날려야 되는데, 이럴 거면 프론트에서 바로 POLLING 요청으로 DB의 상태를 바라보는 것과 크게 다르지 않다는 생각이 들었다.

그래서 최종적으로 프론트에서 해당작업을 요청한 후의 바로 ID를 넘겨받아 로컬 스토리지에 저장, 2초 주기로 인터벌로 요청을 날려 최종 상태를 판단하고 성공 또는 실패가 뜰시 유저에게 알려주고 인터벌을 중단시키는 형태로 작업을 바꿨다.

비동기 SDK 작업 강제로 동기화

해당 작업 중 AWS 미디어컨버팅 API를 호출하는 부분이 있는데, 관련 응답기록을 DB에 저장시켜야 되서, while문을 사용해 강제로 블락킹시켰다. 해당 SDK 내부가 비동기로 작성되어있어서 어디 콜백받을 수 있는 인터페이스를 따로 제공하지 않을까 찾아보았지만 관련문서도 빈약하고 내 빈약한 머리로 찾을 수 가 없더라 ㅠㅠ

    fun getJobUntilComplete(
        jobResponse: CreateJobResult,
        outPutName: String,
        userMastering: UserMastering
    ){


        var job = jobResponse.job
        log.info {  "convertJob.job().id() = " + job.id}

        while ( job.status == null
            || job.status == JobStatus.SUBMITTED.name
            || job.status == JobStatus.PROGRESSING.name ) {
            job = getConvertJob(job.id).job
            log.info("convertJob.job().status() = " + job.status)
            Thread.sleep(5000)
        }

        log.info{"job is complete = " + job.status}


        if (job.status == JobStatus.COMPLETE.name
            || job.status == JobStatus.CANCELED.name
            || job.status == JobStatus.ERROR.name
            ) {

              //update job status to DB
              
            if (job.status != JobStatus.COMPLETE.name){
                  // 실패처리
            }else{
                // increaseStep
            }

        }

    }

(conn=806895) Deadlock found when trying to get lock; try restarting transaction

java.sql.SQLTransactionRollbackException