바닐라 비동기 큐 가이드

비동기 큐는 큐 가이드에서 설명한 순서, 용량, 우선순위, 만료, 시작 또는 중지 제어를 유지합니다. 여기에 동시성 제어, Promise를 고려한 처리, 항목별 재시도, 결과 콜백, 활성 작업 제어 기능을 추가합니다.

수락된 모든 항목을 최종적으로 실행해야 하지만 한 번에 제한된 수의 비동기 작업만 활성화해야 할 때 사용합니다. 초과 호출을 기다리게 하는 대신 시간 창에 따라 거부해야 한다면 비동기 요청률 제한을 사용합니다.

비동기 큐의 작동 방식

항목은 두 단계를 거칩니다.

pending queue                 active work, concurrency: 2

[ A, B, C, D ]    start A    [ A ]
[ B, C, D ]       start B    [ A, B ]
[ B, C, D ]    B finishes    [ A ]
[ C, D ]          wait, C    [ A, C ]

concurrency는 자동으로 예약되는 활성 항목의 수를 제한합니다. 기본값은 1입니다. wait: 0이면 항목 하나가 완료된 후 빈 슬롯을 채웁니다. wait가 양수이면 항목 하나가 완료된 후 해당 시간만큼 기다린 다음 추가 작업을 확인합니다.

큐는 시작 순서를 제어합니다. 동시성이 1보다 크면 완료 순서는 작업 자체에 따라 달라집니다.

빠른 시작

큐에 항목을 추가하는 작업만 필요하다면 asyncQueue를 사용합니다.

import { asyncQueue } from '@tanstack/pacer'

const enqueue = asyncQueue(
  async (job: Job) => {
    await processJob(job)
  },
  { concurrency: 2 },
)

const accepted = enqueue(job)

반환된 함수는 바인딩된 addItem() 메서드입니다. 항목이 대기 큐에 들어가면 true를, 큐에서 거부되면 false를 반환합니다. 해당 항목의 결과에 대한 Promise는 반환하지 않습니다.

생명주기 메서드, 콜백, 상태가 필요하다면 AsyncQueuer를 사용합니다.

import { AsyncQueuer } from '@tanstack/pacer'

const queue = new AsyncQueuer(processJob, {
  concurrency: 2,
  maxSize: 100,
  onSuccess: (result, item) => {
    console.log('Completed:', item.id, result)
  },
  onError: (error, item) => {
    console.error('Failed:', item.id, error)
  },
  onReject: (item) => {
    console.warn('Queue full:', item.id)
  },
})

queue.addItem(job)

생성 시점에 작업이 이미 있다면 initialItems를 전달합니다. 큐는 일반적인 삽입 및 용량 규칙을 적용하며 started: false를 설정하지 않았다면 자동 처리가 즉시 시작될 수 있습니다.

대기 중인 항목의 순서 지정

동기 순서 지정 규칙이 그대로 적용됩니다.

• 기본적으로 뒤에 추가하고 앞에서 읽어 FIFO 순서를 만듭니다.
• LIFO 순서에는 getItemsFrom: 'back'을 설정합니다.
• 삽입을 제어하려면 addItemsTo를 설정하거나 addItem()에 위치를 전달합니다.
• 숫자 우선순위가 높은 항목을 먼저 배치하려면 getPriority(item)을 설정합니다.

우선순위 정렬은 앞 또는 뒤에서 제거하는 규칙보다 우선합니다.

const queue = new AsyncQueuer(processJob, {
  started: false,
  concurrency: 2,
  getPriority: (job) => job.priority,
})

queue.addItem({ id: 'low', priority: 1 })
queue.addItem({ id: 'high', priority: 10 })
queue.addItem({ id: 'medium', priority: 5 })
queue.start()

높은 우선순위와 중간 우선순위 작업이 먼저 시작됩니다. 두 작업의 상대적인 완료 순서는 보장되지 않습니다.

용량 및 거부

maxSize는 활성 항목이 아닌 대기 중인 항목을 제한합니다. 항목이 시작되면 대기 큐에서 빠져나가 대기 슬롯 하나가 비게 됩니다. 대기 큐가 가득 차면 addItem()은 false를 반환하고 onReject를 호출합니다. 큐가 내부적으로 사용 가능한 항목이 없음을 나타내는 데 사용하므로 undefined도 거부됩니다.

if (!queue.addItem(job)) {
  saveForLater(job)
}

호출자가 빈 슬롯을 기다리지 않으므로 용량만으로는 배압을 제공하지 않습니다. 항목을 버릴 수 없다면 거부를 명시적으로 처리합니다.

결과 및 오류

자동으로 예약된 작업은 백그라운드에서 실행됩니다. 결과는 onSuccess나 queue.store.state.lastResult를 통해 관찰하고 실패는 onError와 오류 카운터를 통해 관찰합니다.

직접 제어하려면 execute()로 대기 중인 항목 하나를 제거하고 처리합니다. 이 메서드의 Promise는 래핑된 함수의 결과가 아니라 처리한 항목으로 이행됩니다. 결과는 onSuccess에 전달되고 lastResult로 저장됩니다.

onError가 없으면 throwOnError의 기본값은 true입니다. 백그라운드 예약은 콜백과 상태를 업데이트한 후 해당 거부를 포착하므로 큐가 계속 동작할 수 있습니다. execute()나 flush()를 직접 호출하면 호출자에게 거부가 전달될 수 있습니다. onError를 제공하면 기본값이 false로 바뀝니다.

콜백은 다음과 같습니다.

• onSuccess(result, item, queue)는 항목이 성공한 후 실행됩니다.
• onError(error, item, queue)는 해당 항목의 재시도가 실패한 후 실행됩니다.
• onSettled(item, queue)는 어느 결과든 완료된 후 실행됩니다.
• onItemsChange(queue)는 대기 컬렉션이 변경될 때 실행됩니다.
• onReject(item, queue)와 onExpire(item, queue)는 실행되지 않는 항목에 대해 실행됩니다.

항목은 해당 함수가 시작되기 전에 대기 큐에서 제거됩니다. 실패한 항목은 자동으로 다시 추가되지 않습니다.

항목 재시도

시작된 각 항목은 자체 재시도기를 사용합니다.

const queue = new AsyncQueuer(processJob, {
  concurrency: 2,
  asyncRetryerOptions: {
    maxAttempts: 3,
    backoff: 'exponential',
    baseWait: 500,
    jitter: 0.2,
  },
})

재시도는 동일한 활성 항목에 속하며 계속 동시 실행 슬롯을 차지합니다. maxAttempts에는 첫 시도가 포함됩니다. 부수 효과가 있는 작업을 재시도하기 전에 비동기 재시도 가이드를 참고합니다.

시작, 중지 및 플러시

started: false를 설정하지 않으면 큐가 자동으로 시작됩니다.

• stop()은 새로운 자동 시작을 막고 대기 중인 타이머를 지웁니다. 활성 항목을 중단하거나 대기 중인 항목을 제거하지는 않습니다.
• start()는 자동 처리를 재개합니다.
• clear()는 대기 중인 항목을 제거합니다. 활성 항목에는 영향을 주지 않습니다.
• flush(count?)는 일반적인 대기 간격 없이 대기 중인 항목을 즉시 시작합니다.
• flushAsBatch(fn)은 대기 중인 모든 항목을 제거하고 하나의 비동기 배치 함수에 전달합니다.

flush()는 직접 실행을 사용하며 설정된 concurrency보다 많은 작업을 시작할 수 있습니다. 일반 예약이 아닌 의도적인 비우기 작업으로 사용합니다.

직접 플러시된 항목 중 하나라도 throwOnError: true 상태에서 거부되면 요청된 모든 실행이 완료된 후 flush()가 거부됩니다. 큐가 실행 중이면 남아 있는 대기 작업이 그 후 재개됩니다.

만료

대기 중인 항목은 expirationDuration 또는 getIsExpired(item, addedAt)를 통해 만료될 수 있습니다. 만료는 항목별 전용 타이머가 아니라 큐 틱에서 확인합니다. 따라서 중지된 큐는 재개된 후 오래된 항목을 평가합니다.

만료된 항목은 제거되고 onExpire를 호출하며 처리 함수에는 절대 도달하지 않습니다. 활성 항목은 만료되지 않습니다.

활성 작업 중단

abort()는 모든 활성 실행의 재시도기를 중단합니다. 대기 중인 항목은 지우지 않습니다. 처리 함수에서 신호를 사용하는 경우에만 취소가 기반 API에 전달됩니다.

const queue = new AsyncQueuer(
  async (job: Job) => {
    return fetch(`/api/jobs/${job.id}`, {
      method: 'POST',
      signal: queue.getAbortSignal() ?? undefined,
    })
  },
  { concurrency: 2 },
)

queue.abort()

여러 실행이 겹칠 때 특정 실행의 신호가 필요하다면 executeCount를 getAbortSignal()에 전달합니다.

안전하게 재설정하기

reset()은 빈 대기 큐와 실행 상태를 포함한 기본 상태를 복원합니다. 큐의 대기 타이머를 지우거나 활성 기반 작업의 중지를 보장하지는 않습니다. 먼저 명시적인 생명주기 메서드를 사용합니다.

queue.stop()
queue.abort()
queue.reset()

설정 및 상태

concurrency와 wait에는 값 또는 큐 인스턴스를 받는 함수를 지정할 수 있습니다. setOptions()는 새 옵션을 병합하고 asyncQueuerOptions()는 재사용할 수 있고 타입 검사를 거친 옵션 객체를 생성합니다.

initialState는 애플리케이션이 유지한 일부 큐 상태를 복원할 수 있습니다. items가 포함되어 있으면 initialItems보다 우선하며 initialState.isRunning도 마찬가지로 started보다 우선합니다. 지속 가능한 필드만 복원합니다. 대기 중인 타이머와 활성 실행은 복원되지 않습니다.

일반적인 상태는 다음과 같습니다.

• items와 size: 대기 중인 작업입니다.
• activeItems: 현재 활성 상태로 추적되는 작업입니다.
• isRunning, isIdle, status: 스케줄러 상태입니다.
• isFull과 rejectionCount: 대기 용량 상태입니다.
• successCount, errorCount, settledCount: 실행 결과입니다.
• lastResult: 가장 최근의 성공한 처리 결과입니다.

복사된 항목 배열에는 peekPendingItems(), peekActiveItems(), peekAllItems()를 사용합니다. 모든 메서드와 상태는 AsyncQueuer API 레퍼런스를 참고합니다.