NSQ 메시지 큐의 디스크 저장소: DiskQueue 설계 상세

개요 및 역할

NSQ 아키텍처에서 DiskQueue는 비휘발성 스토리지 역할을 수행하는 핵심 컴포넌트입니다. 메모리 기반 처리의 속도와 디스크 기반 영구성의 안정성을 결합하여, 시스템 재부팅 후에도 손실되지 않도록 메시지를 파일로 관리합니다. 이 구성 요소는 메모리와 디스크 사이의 효율적인 데이터를 이동시키는 메커니즘을 제공하며, 특히 고可用 환경에서 데이터 일관성을 유지하는 데 중요한 역할을 합니다.

이 글에서는 라이브러리로 분리되어 관리되는 go-diskqueue의 핵심 로직과 내부 설계 패턴을 해부합니다. 주요 초점은 쓰기 전략, 읽기 흐름, 그리고 시스템 장애 시 복구가 어떻게 이루어지는지에 맞춰져 있습니다.

인터페이스와 아키텍처

저장소 계층은 표준화된 인터페이스를 따릅니다. 메시지 처리 단위는 바이트 배열이며, 이는 직렬화/역직렬화 부담을 상위로 위임함으로써 스토리지 레벨의 복잡도를 낮춥니다.

// 추상화된 영구 저장 인터페이스
type DataPersistInterface interface {
    Submit(payload []byte) error
    GetOutputStream() <-chan []byte
    Shutdown() error
    Clear() error
    CountItems() int64
}

내부 상태 관리를 위한 구조체는 여러 개의 통신 채널을 통해 메인 스레드와 작업을 담당하는 워커 루틴 간의 동기화를 처리합니다.

type PersistentDisk struct {
    // ... 기타 필드
    
    // 현재 누적된 메시지 개수 전송용
    sizeNotify       chan int64
    // 신규 메시지 입력용
    inputStream      chan []byte
    // 작업 결과 응답용
    resultStream     chan error
    // 전체 데이터 삭제 요청용
    flushRequest     chan struct{}
    // 종료 신호
    shutdownSignal   chan bool
}

파일 회전 (Log Rotation) 와 메타데이터

단일 파일이 무한히 커지는 것을 방지하기 위해, 일정 크기 제한에 도달하면 자동으로 새 파일로 전환되는 로그 회전 전략을 사용합니다. 이는 불필요하게 성장한 파일을 정리하고 삭제하는 과정을 단순화합니다.

회전 및 복구 상태를 추적하기 위해 다음 세 가지 메타데이터가 유지됩니다:

현재 대기 중인 총 메시지 수
읽기 커서 위치 (파일 번호 및 오프셋)
쓰기 커서 위치 (파일 번호 및 오프셋)

메타데이터의 신뢰성은 매우 중요합니다. 중단의 경우 데이터 손실을 막기 위해 원자적(A tomic) 쓰기 기법을 적용합니다. 직접 덮어쓰기를 하는 대신, 임시 파일에 먼저 저장한 후 성공 시 원본 파일명으로 이름을 변경(rename)하는 방식을 사용함으로써 부정한 상태의 파일 생성을 방지합니다.

쓰기 프로세스 (Write Path)

외부로부터 메시지가 들어오면 Submit 메서드를 통해 호출됩니다. 실제로 디스크 IO 를 수행하는 것은 별도의 루틴이 담당합니다.

func (pd *PersistentDisk) Submit(msg []byte) error {
    pd.LockRead()
    defer pd.UnlockRead()

    if pd.isStopped {
        return ErrSystemShutdown
    }

    pd.inputStream <- msg
    return <-pd.resultStream
}

워커 루틴 내에서 각 메시지를 파일에 기록할 때는 버퍼링과 파일 크기 검사를 함께 수행합니다.

func persistMessage(pd *PersistentDisk, payload []byte) error {
    dataLength := int32(len(payload))
    totalSize := int64(4 + dataLength)

    // 단일 파일 크기 초과 시 회전 처리
    if pd.currentWritePos > 0 && pd.currentWritePos+totalSize > maxFileSize {
        pd.rotateFileIfNeeded()
    }

    // 파일 핸들링이 비어있으면 초기화
    if pd.writeHandle == nil {
        filename := constructFileName(pd.fileIndex)
        handle, err := os.OpenFile(filename, os.O_RDWR|os.O_CREATE, 0600)
        if err != nil {
            return err
        }
        pd.writeHandle = handle
        pd.seekOffset(pd.currentWritePos)
    }

    // 레코드 길이와 실제 데이터를 바인딩하여 버퍼에 담음
    pd.buffer.Reset()
    binary.Write(&pd.buffer, binary.BigEndian, dataLength)
    pd.buffer.Write(payload)

    // 디스크 기록
    _, err := pd.writeHandle.Write(pd.buffer.Bytes())
    if err != nil {
        pd.closeHandle(false)
        return err
    }

    pd.currentWritePos += totalSize
    pd.itemCount++
    return nil
}

여기서 주의할 점은 파일 회전 과정 자체가 완전히 원자적으로 보장되지 않는다는 것입니다. 절차는 다음과 같이 나뉩니다:

기존 파일 동기화 (Sync)
메타데이터 갱신 (새 파일 인덱스 반영)
새로운 데이터 파일 생성

만약 단계 1 과 2 사이에서 정전이 발생하면 기존 파일 크기가 메타데이터의 기록된 위치보다 클 수 있습니다. 이때는 복구 로직이 이를 감지하고 자동으로 회전을 트리거하도록 설계되어 있습니다. 단계 2 와 3 사이에 문제가 생기면 해당 파일이 누락되지만, 최초 쓰기 시점에 자동으로 재생성되어 정상 동작합니다.

읽기 프로세스 (Read Path)

메시지 소모자는 읽기 채널을 통해 데이터를 받습니다. 내부적으로는 워커 루틴이 데이터를 읽어 채널로 방출하는 방식으로 연결됩니다.

func retrieveMessage(pd *PersistentDisk) ([]byte, error) {
    var length int32

    if pd.readHandle == nil {
        filename := constructFileName(pd.readFileIndex)
        handle, err := os.OpenFile(filename, os.O_RDONLY, 0600)
        if err != nil {
            return nil, err
        }
        pd.readHandle = handle
        pd.setReader(bufio.NewReader(handle))
        
        // 읽기 포지션 설정 및 최대 크기 계산
        pd.seekOffset(pd.readPos)
        pd.maxReadLimit = getMaxReadableBytes()
    }

    // 헤더부터 크기 정보 확인
    if err := binary.Read(pd.reader, binary.BigEndian, &length); err != nil {
        pd.closeHandle(true)
        return nil, err
    }

    // 실제 페이로드 획득
    buf := make([]byte, length)
    if _, err := io.ReadFull(pd.reader, buf); err != nil {
        pd.closeHandle(true)
        return nil, err
    }

    pd.nextReadPos = pd.readPos + 4 + int64(length)

    // 현재 파일 끝까지 읽었으면 다음으로 이동
    if pd.nextReadPos >= pd.maxReadLimit {
        pd.advanceToFileNext()
    }

    return buf, nil
}

동기화 전략과 데이터 손실 리스크

입력 성능을 극대화하기 위해 매번 쓰기에 즉시 fsync 를 수행하지는 않습니다. 다음 조건 중 하나를 만족할 때만 물리적 디스크에 강제로 플러시합니다.

로그 회전 시점 도달
정해진 횟수의 쓰기 이후 (syncInterval)
시간 초과 발생 (syncTimeout)

이 설계는 운영체제의 페이지 캐시를 활용하여 높은 처리량을 제공합니다. 그러나 동시에, 서버가 갑자기 강제 종료될 경우 OS 페이지 캐시에 남아 있었으나 디스크로 아직 쓰이지 않은 데이터는 사라질 가능성이 있음을 의미합니다. 즉, '영속성'을 약속하지만 완벽한 데이터 손실 보호 수준까지는 아니라는 점을 이해해야 합니다.

동시성 최적화: 깊이 (Depth) 측정

현재 대기 중인 메시지 수를 조회하는 기능은 경합을 최소화하기 위해 채널을 이용합니다. 공유 변수에 접근할 때 항상 락을 걸면 성능 저하가 발생할 수 있으므로, 이를 피하는 방법을 사용합니다.

func (pd *PersistentDisk) CountItems() int64 {
    select {
    case val := <-pd.sizeNotify:
        return val
    default:
        return pd.cachedItemCount
    }
}

워커 루틴은 주기적으로 혹은 상태 변화 시마다 최신 카운트를 sizeNotify 채널로 푸시합니다. 조회 측에서는 채널이 닫혔거나 비어있는 경우 캐시된 값을 반환하도록 하여, 비동기적 업데이트 상황에서 락 없는 효율적인 조회가 가능하도록 설계되었습니다.

태그: nsq Golang message-queue distributed-systems persistent-storage

8월 1일 19:28에 게시됨

괴물 클럽