Preact 비동기 큐잉 가이드

비동기 큐잉은 큐잉 가이드에서 설명한 순서, 용량, 우선순위, 만료, 시작 또는 중지 제어를 유지합니다. 여기에 제어 가능한 동시 실행, 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보다 크면 완료 순서는 작업 자체에 따라 달라집니다.

API 선택하기

• 반응형 대기 항목이 필요하면 useAsyncQueuedState
• 동시 실행, 순서, 수명 주기 제어가 필요하면 useAsyncQueuer

Preact 예제

import { useAsyncQueuedState } from '@tanstack/preact-pacer'

function UploadQueue() {
  const [pending, queue] = useAsyncQueuedState(
    uploadFile,
    { concurrency: 2 },
    (state) => ({
      items: state.items,
      activeItems: state.activeItems,
    }),
  )

  return (
    <button onClick={() => queue.addItem(nextFile())}>
      Upload ({pending.length} waiting, {queue.state.activeItems.length} active)
    </button>
  )
}

이 가이드에서 이후에 다루는 핵심 코드 조각은 useAsyncQueuer를 사용하며 컴포넌트나 다른 훅 내부에서 실행한다고 가정합니다.

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

대기 항목 순서 지정하기

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

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

우선순위 정렬은 앞이나 뒤에서 제거하는 설정보다 우선합니다.

const queue = useAsyncQueuer(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를 사용하거나 어댑터 상태에서 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 = useAsyncQueuer(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 = useAsyncQueuer(
  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()

Preact 수명 주기

어댑터는 소유자가 제거될 때 자동 처리를 중지하고 활성 작업을 중단합니다. onUnmount를 제공하면 이 기본 정리 동작을 대체하므로 사용자 정의 콜백에서 필요한 모든 수명 주기 작업을 수행해야 합니다. 사용자 정의 정리 과정에서 작업을 플러시하면 컴포넌트가 제거되는 동안 사용자 콜백이 실행될 수 있다는 점에 유의해야 합니다.

설정과 반응형 상태

어댑터는 셀렉터 인수가 반환한 상태만 구독합니다. 셀렉터가 없으면 어댑터 상태는 비어 있습니다. 컴포넌트나 다른 훅의 최상위 수준에서 유틸리티를 만들고 뷰에서 사용하는 필드만 선택합니다.

const queue = useAsyncQueuer(processJob, { concurrency: 2 }, (state) => ({
  size: state.size,
  activeItems: state.activeItems,
  status: state.status,
}))

console.log(queue.state.size, queue.state.activeItems, queue.state.status)

옵션 함수와 수명 주기 콜백은 기반 공개 유틸리티 인스턴스를 받습니다. 위 예제처럼 해당 콜백 안에서 .store.state를 읽는 방식을 지원합니다. 렌더링 코드는 여기에 나온 선택된 어댑터 상태를 읽어야 합니다.

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()를 사용합니다. 어댑터 시그니처는 Preact API 레퍼런스를, 전체 옵션 및 상태 타입은 공개 코어 레퍼런스를 참고합니다.