Python Queue.join()이 끝났는데 작업은 실패했을 때: task_done과 결과 분리

Queue.join()이 반환됐다는 이유만으로 작업이 모두 성공했다고 처리하면 실패한 배치를 정상으로 넘길 수 있습니다. 이 글의 일반적인 Queue 처리 흐름에서 join은 put한 항목에 대해 task_done 호출이 맞게 돌아왔는지 기다립니다. 성공·실패 결과를 판단하거나 작업자의 예외를 호출자에게 다시 던지는 기능은 아닙니다.

작업 두 개 중 하나가 실패해도 finally에서 처리 확인을 반환하면 join은 끝납니다. 아래 실습은 오류를 별도 결과 큐에 남기고, 작업자 종료까지 확인한 뒤 호출자가 실패를 판정합니다. 실제 파일·DB·외부 API를 건드리지 않는 합성 작업입니다.

큐 항목의 처리 확인이 모두 돌아와도 실패한 작업 결과가 따로 남을 수 있다는 개념 그림 · 실행 수치나 실제 서비스 화면이 아닌 개념 그림입니다.

대기 중인 항목 수와 실패한 작업 수를 나눕니다

Queue.get은 항목을 꺼내는 동작입니다. 일반적인 작업 추적에서는 성공적으로 꺼낸 항목마다 처리 정책을 마친 뒤 task_done을 한 번 호출합니다. 큐가 비어 있다는 것과 이미 꺼낸 작업의 처리가 끝났다는 것은 다릅니다. 다른 작업자가 아직 실행 중일 수 있으므로 empty나 qsize만으로 전체 완료를 판단하지 않습니다.

실패를 기록하고 더 이상 이번 큐에서 처리하지 않기로 했다면 그 항목에도 처리 확인을 반환할 수 있습니다. 이때 실패가 성공으로 바뀌는 것이 아닙니다. 재시도할지 중단할지는 별도 결과와 정책으로 결정해야 합니다. task_done은 성공 판정 도장으로 사용하지 않습니다.

큐 처리 확인 기록과 작업 성공·실패 기록을 별도로 보존하는 개념 그림 · 실행 수치나 실제 서비스 화면이 아닌 개념 그림입니다.

결과 큐와 finally를 둔 작은 작업자를 실행합니다

Python 3.14.7에서 검증했습니다. queue_outcomes_demo.py로 저장해 python3 queue_outcomes_demo.py로 실행하세요. 단일 작업자가 ok·bad를 순서대로 받고, bad는 고정 ValueError를 일으킵니다. 종료 표식 stop도 put한 항목이므로 finally에서 task_done을 호출합니다.

from queue import Queue, Empty
from threading import Thread

# Local synthetic jobs only; no file, database, or external API effects.
def run_jobs(items):
    jobs, outcomes = Queue(), Queue()
    stop = object()

    def worker():
        while True:
            item = jobs.get()
            try:
                if item is stop:
                    return
                if item == 'bad':
                    raise ValueError('synthetic task failure')
                outcomes.put((item, 'OK', None))
            except Exception as exc:
                outcomes.put((item, 'FAILED', type(exc).__name__))
            finally:
                jobs.task_done()

    for item in items:
        jobs.put(item)
    jobs.put(stop)
    thread = Thread(target=worker)
    thread.start()
    jobs.join()
    print('queue.join returned')
    thread.join()
    records = []
    while True:
        try:
            records.append(outcomes.get_nowait())
        except Empty:
            break
    return records

if __name__ == '__main__':
    records = run_jobs(['ok', 'bad'])
    print('records=', records)
    failed = [item for item, status, _ in records if status == 'FAILED']
    print('failed jobs=', failed)
    assert records == [('ok', 'OK', None), ('bad', 'FAILED', 'ValueError')]
    try:
        if failed:
            raise RuntimeError('job group failed: ' + ', '.join(failed))
    except RuntimeError as exc:
        print('caller rejected:', str(exc))

실제 출력: join은 끝나지만 실패 기록은 남습니다

queue.join returned
records= [('ok', 'OK', None), ('bad', 'FAILED', 'ValueError')]
failed jobs= ['bad']
caller rejected: job group failed: bad

queue.join returned 뒤에도 bad의 FAILED·ValueError 기록이 존재합니다. Thread.join은 작업자가 끝날 때까지 기다리지만, 작업자에서 처리하지 않은 예외를 호출 스레드에 자동으로 다시 던져 주는 메서드가 아닙니다. 이번 코드는 작업 예외를 작업자 안에서 기록하고, 호출자가 결과를 모아 실패를 판정합니다.

끝의 RuntimeError를 잡아 출력한 부분은 이 실습의 결과를 눈으로 확인하기 위한 것입니다. 실제 배치에서는 이 실패를 상위 호출자에게 전달하거나 실패 종료 상태로 반환해야 합니다. 메시지를 출력한 뒤 성공 경로로 계속 진행하면 실패를 숨기는 문제가 다시 생깁니다.

오류 기록을 보존한 뒤 finally에서 해당 큐 항목의 처리 확인을 반환하는 개념 그림 · 실행 수치나 실제 서비스 화면이 아닌 개념 그림입니다.

finally 하나로 모든 중단 상황이 해결되지는 않습니다

성공한 get마다 task_done을 한 번이라는 짝을 유지하세요. 빠뜨리면 join이 계속 기다릴 수 있고, 더 많이 호출하면 ValueError가 발생할 수 있습니다. get 자체가 성공하기 전부터 무조건 task_done을 호출하는 구조도 피해야 합니다. 예제는 get 다음에 try·finally를 시작합니다.

작업이 끝나기 전에 확인을 반환하면 join이 실제 처리보다 먼저 끝날 수 있습니다. 반대로 작업자가 멈추거나 프로세스가 강제 종료되면 finally나 실패 기록이 반드시 남는다고 보장할 수 없습니다. 이 메모리 큐는 작업 영속성·재시도 횟수·중복 실행 방지·제한 시간까지 제공하는 작업 시스템이 아닙니다.

Python 3.13 이상 Queue.shutdown(immediate=True)는 남은 항목의 일반 처리 흐름과 다른 즉시 종료 규칙을 가집니다. 작업이 실행되지 않아도 join이 풀릴 수 있으므로, join 반환을 모든 상황의 처리 확인이나 성공 증거로 절대화하지 않습니다. 이번 예제는 그 메서드를 사용하지 않는 종료 표식 방식입니다.

종료 표식의 처리 확인과 작업자 종료를 확인한 뒤 결과를 판정하는 개념 그림 · 실행 수치나 실제 서비스 화면이 아닌 개념 그림입니다.

종료와 결과 판정을 별도로 확인합니다

예제에서는 jobs.join으로 큐의 처리 확인을 기다리고 Thread.join으로 작업자 종료를 기다린 뒤, 더 이상 결과를 넣는 생산자가 없는 상태에서 get_nowait·Empty로 결과를 꺼냅니다. 일반적인 동시 생산 상황에서 empty 확인 뒤 get으로 안전하게 꺼낼 수 있다고 일반화하지 않습니다.

여러 작업자로 확장하면 각 작업자의 종료 방법, 결과의 순서와 작업 ID, 누락된 결과, 재시도·실패 저장 정책이 더 필요합니다. 작업량이 크면 결과 큐·기록 저장소의 용량도 정해야 합니다. 큐 자체의 용량 정책은 deque(maxlen)과 Queue의 보존·대기 차이와 연결됩니다.

큐의 처리 확인, 작업자 종료, 업무 성공을 따로 기록하세요. join이 끝났는지와 실패가 없는지를 각각 확인해야 배치 결과를 검토할 수 있습니다.