−백프레셔(backpressure)는 소비자가 처리할 수 있는 양에 맞춰, 생산자가 다음 데이터를 내보내는 속도나 양을 조절하도록 만드는 흐름 제어다. 다운로드한 파일을 처리하는 쪽보다 읽어들이는 쪽이 빠르면 대기열이 계속 커질 수 있는데, 대기열이 찼을 때 생산자도 기다리게 하면 적체가 무한히 늘어나는 것을 막을 수 있다.
여기서는 한 프로세스의 비동기 대기열로 원리를 살펴본다. asyncio.Queue(maxsize=2)가 찼을 때 await put()은 항목이 빠질 때까지 해당 생산자 코루틴을 기다리게 한다. 그동안 이벤트 루프는 다른 작업을 실행할 수 있다. Python Queue 문서
[이미지]
핵심은 produce가 각 put의 완료를 기다린다는 점이다. put 작업을 끝없이 별도 Task로 만들어 두면 대기열 밖에서 기다리는 작업 수가 다시 늘어난다.
꽉 찬 대기열 앞에서 기다리는 생산자
import asyncio
async def produce(queue):
peak = 0
for item in range(6):
await queue.put(item)
peak = max(peak, queue.qsize())
await queue.put(None)
return peak
async def consume(queue):
values = []
while True:
item = await queue.get()
try:
if item is None:
return values
await asyncio.sleep(0.01)
values.append(item)
finally:
queue.task_done()
async def main():
queue = asyncio.Queue(maxsize=2)
async with asyncio.TaskGroup() as group:
producer = group.create_task(produce(queue))
consumer = group.create_task(consume(queue))
await producer
await queue.join()
assert consumer.result() == list(range(6))
assert producer.result() <= 2
print(consumer.result())
if __name__ == "__main__":
asyncio.run(main())
Python 3.11 이상에서 실행한다. 생산자가 넣는 여섯 값은 0부터 5이고, None은 입력 값으로 사용하지 않는 종료 표지다. 소비자는 표지를 포함해 get으로 꺼낸 항목마다 task_done을 호출한다. 생산자가 마지막 표지를 넣은 뒤 join으로 모든 항목의 완료를 확인하고, TaskGroup을 나가며 작업 종료도 기다린다.
소비자가 CancelledError를 제외한 일반 예외로 실패하면 TaskGroup이 생산자도 취소하므로, 꽉 찬 대기열 앞에서 계속 기다리는 작업을 남기지 않는다. 소비자 작업 하나만 따로 취소하면 이 실패 전파 규칙이 적용되지 않아 생산자가 계속 기다릴 수 있다. 요청을 중단할 때 두 작업을 함께 끝내려면 부모 작업을 취소하는 등 소유 범위 전체의 종료를 연결해야 한다. TaskGroup의 예외·취소 규칙
task_done은 이 예제의 대기열 회계를 맞추며, 실패한 업무가 성공했다는 뜻으로 해석하지 않는다. 실패는 작업의 예외로 전달된다.
비차단 흐름 제어를 설명하는 원문
[이미지]
Reactive Streams 공식 소개의 첫 문장 부분을 2026년 10월 28일 캡처했다. non-blocking back pressure는 받는 쪽을 무제한 버퍼에 떠넘기지 않으면서 비동기 흐름을 유지하려는 목표와 연결된다. 원문, 공식 명세와 MIT-0 이용 조건.
Reactive Streams의 Publisher·Subscriber·Subscription은 소비자의 요청량을 신호로 전달하는 프로토콜을 정의한다. 위 Queue 예제는 같은 문제를 제한된 버퍼와 기다림으로 보여주며, 그 명세의 인터페이스나 적합성 검사를 구현한 것은 아니다. 명세의 Publisher 규칙 1.1
무엇의 개수를 제한하는가
장치 | 직접 다루는 대상 | 살펴볼 예
레이트 리미터 | 일정 시간에 허용할 요청량 | 초당 접수 요청 수
벌크헤드·동시 실행 상한 | 동시에 자원을 사용하는 작업 수 | 외부 API 호출 칸
백프레셔 | 생산자와 소비자 사이의 흐름 | 버퍼가 차면 다음 생산을 기다림
세 장치는 함께 사용할 수 있다. 대기열 길이가 2여도 소비자가 이미 꺼내 처리 중인 값과 생산자가 다음에 넣으려는 값은 별도로 존재한다. 항목 크기는 서로 다를 수 있으므로, maxsize=2가 제한하는 항목 개수와 프로세스 전체 메모리의 바이트 상한을 구별해야 한다.
생산자를 멈출 수 없는 입력에서는 버리기·최신 값으로 합치기·디스크에 저장하기·접수 거절 같은 정책을 정해야 한다. 대기열 상한만 추가하고 누락된 항목을 성공으로 알리면 업무 계약이 달라진다. 이 예제는 모든 입력 값을 순서대로 처리하는 대기 정책을 사용한다.
함께 읽기
구조적 동시성과 취소
레이트 리미터 패턴
벌크헤드 패턴
+파일을 읽는 생산자가 초당 100개를 내보내고 다음 단계의 소비자가 초당 60개를 처리한다고 해보자. 차이가 계속되면 초마다 40개가 쌓인다. 버퍼를 키워 잠깐 더 받아둘 수는 있어도 소비 속도가 그대로라면 그 버퍼도 결국 찬다.
백프레셔(backpressure)를 적용하면 받는 쪽의 여유에 맞춰 보내는 쪽이 생산이나 전송을 조절한다. 데이터는 생산자에서 소비자로 흐르고 더 받을 수 있는지 알리는 신호는 상류로 거슬러 올라간다. TCP의 수신 윈도, 비동기 큐의 대기, 스트림의 요청량은 이 문제를 서로 다른 단위와 실행 방식으로 다룬다.
수신자가 받을 수 있는 양을 알려준다
1981년 TCP 명세인 RFC 793에서 수신 측은 지금 받을 수 있는 바이트 수를 윈도(window)로 알려준다. 송신 측은 보낼 데이터가 더 있어도 이 범위를 고려해 전송한다. 비동기 스트림 라이브러리가 나오기 전부터 수신자의 여유를 송신자에게 알려주는 프로토콜을 사용했던 것이다.[각주 1]
수신 측 TCP가 데이터를 받아줬다고 해서 애플리케이션이 그 내용의 주문 처리나 파일 변환을 마친 것은 아니다. 애플리케이션이 소켓에서 계속 읽어 무제한 큐에 넣는다면 TCP 수신 버퍼를 비워주는 동안에도 내부 작업은 쌓인다. 그래서 소켓 바깥의 생산자와 소비자까지 어느 단계에서 더 기다리게 할지 연결해야 한다.
통신 분야에서는 흐름 제어와 혼잡 제어도 구분한다. 수신자는 자신의 수용 능력을 알리고 송신 측의 혼잡 제어는 네트워크 경로가 감당할 전송량을 다룬다. 현재 TCP 기본 명세인 RFC 9293도 송신 윈도와 혼잡 윈도를 구별한다.[각주 2]
비동기로 나누면 어디에서 기다릴까
한 스레드에서 함수를 차례로 호출하면 앞선 호출이 끝날 때까지 다음 호출을 기다린다. 생산자가 작업을 큐나 별도 Task로 넘기면 소비자의 완료를 기다리지 않고 다음 일을 만들 수 있다. 그 사이의 데이터를 어디에 얼마나 보관할지도 정해야 한다.
생산자가 큐의 각 put 완료를 기다리면 큐가 찼을 때 생산자도 다음 항목을 넣지 못하고 멈춰 선다. 반대로 put을 수행할 Task를 계속 새로 만들면 큐는 작게 유지하더라도 그 바깥에 대기하는 Task가 늘어날 수 있다.
[이미지]
위쪽 화살표는 데이터가 흐르는 방향이고 아래쪽은 대기나 수요 신호가 상류로 전달되는 방향이다. 큐에서 대기하든 요청량을 알리든 그 신호가 생산자의 다음 행동까지 이어진다.
단계별 큐를 제어 지점으로 삼은 서버
Welsh·Culler·Brewer는 2001년 SEDA 논문에서 서버를 여러 처리 단계로 나누고 단계 사이의 큐를 명시적으로 드러냈다. 각 단계의 적체를 보고 스레드 수나 처리 묶음을 조절할 수 있도록 만든 것이다.[각주 3]
[이미지]
(원문 출처)
이벤트는 입력 큐를 거쳐 처리기로 들어가고 처리기는 다른 단계로 이벤트를 보낸다. 제어기는 이 단계의 자원 사용과 스케줄링을 조절한다.
저자들은 §3.3에서 큐가 새 항목을 거절할 수 있게 하고 애플리케이션이 가득 찬 큐 앞에서 기다리거나 이벤트를 버리는 정책을 고를 수 있다고 설명한다. 큐를 둔 것만으로 상한과 정책이 저절로 정해지지는 않으며 애플리케이션이 그 제어 지점을 이용해 생산을 늦추거나 접수를 줄인다.
이렇게 처리 단계를 나누면 각 단계를 따로 조정하기 쉬워지는 대신 한 단계가 늦어졌을 때 그 제약을 앞단까지 어떻게 전달할지도 정해야 한다. 소켓의 흐름 제어에서 다루던 문제를 서버 내부에서도 만나는 것이다.
라이브러리 사이에서 주고받는 수요
Reactive Streams 참여자들은 비동기 스트림 라이브러리가 수요와 취소를 공통된 규칙으로 주고받도록 인터페이스를 정했다. 1.0.0 태그가 가리키는 커밋은 2015년 4월의 기록으로, 공식 발표에서 비차단 백프레셔를 갖춘 비동기 스트림 처리를 표준화하겠다는 목표를 밝힌다.[각주 4]
[이미지]
(원문 출처)
공식 소개 첫 문장은 참여자들이 내세운 목표를 요약하며 개별 신호의 세부 계약은 명세에서 정의한다.[각주 5]
Publisher가 원소를 보내고 Subscriber가 받는다. Subscriber는 Subscription의 request(n)을 호출해 추가로 받을 수 있는 개수를 알린다. Publisher 규칙 1.1은 누적 전달량이 누적 요청량을 넘지 못하게 하고 Subscription 규칙 3.8은 새 요청량을 남은 수요에 더하도록 정한다.[각주 6]
Subscriber가 request(2)를 호출하고 원소 하나를 받았다면 남은 수요는 하나다. 이어서 request(3)을 호출하면 넷이 된다. 여기서 세는 값은 전달을 허용한 원소 수다. 초당 네 개 같은 시간당 속도는 별도로 정하며 생산자는 데이터가 없거나 스트림을 끝낼 때 요청량보다 적게 보낼 수 있다.
남은 수요가 0인 동안에는 추가 데이터 원소를 보내지 않는다. 새 요청에는 양수를 쓰며 request(0)은 일시 정지 명령이 아니라 잘못된 요청이다. 완료와 오류는 데이터 원소와 다른 신호이므로 수요가 없어도 전달할 수 있다.[각주 7]
Java 9의 java.util.concurrent.Flow도 Publisher·Subscriber·Subscription·Processor를 제공한다. 공식 문서는 이 인터페이스들이 Reactive Streams에 대응한다고 설명한다. 라이브러리 작성자는 데이터를 보내는 메서드뿐 아니라 수요·취소·종료의 계약까지 맞춰야 서로 연결할 수 있다.[각주 8]
기다림을 표현하는 방법
방식 | 하류의 제약을 상류에 전달하는 방법 | 살펴볼 범위
동기 호출·블로킹 쓰기 | 호출이 돌아오기 전까지 다음 전송을 기다린다 | 대기하는 스레드와 자원
제한된 비동기 큐 | 큐가 차면 생산자의 put을 보류한다 | 생산자가 각 완료를 기다리는지
pull 방식 | 소비자가 다음 값을 요청할 때 생산을 진행한다 | 내부 선읽기와 별도 큐
수요 기반 스트림 | 남은 요청량 안에서 원소를 전달한다 | 전달 원소 수와 실제 처리 중인 작업
블로킹 구현에서는 스레드가 대기할 수 있고 비동기 구현에서는 코루틴을 보류한 동안 다른 작업을 실행할 수 있다. Reactive Streams는 비동기 구성 요소를 연결하면서 제어 신호 때문에 스레드를 계속 붙잡지 않도록 비차단을 요구한다.[각주 9]
수요를 셀 때는 항목을 받은 시점과 업무를 끝낸 시점도 구별한다. Subscriber가 원소를 받을 때마다 새 DB 작업을 만들고 곧바로 수요를 보충하면 원소 전달 규칙을 지키더라도 실행 중인 DB 작업은 계속 늘어날 수 있다. 자원 사용을 제한하려는 단계는 어떤 일을 마쳤을 때 다음 수요를 요청할지까지 정해야 한다.
큐 두 칸으로 보는 작은 예
다음은 Python 3.11 이상에서 실행한다. 생산자는 maxsize=2인 큐에 값을 하나씩 넣고 각 put의 완료를 기다린다. 큐가 차면 await queue.put()이 생산자 코루틴을 보류하고 이벤트 루프는 소비자 같은 다른 작업을 실행한다.[각주 10]
import asyncio
async def produce(queue):
for item in range(6):
await queue.put(item)
await queue.put(None)
async def consume(queue):
values = []
while True:
item = await queue.get()
if item is None:
return values
await asyncio.sleep(0.01)
values.append(item)
async def main():
queue = asyncio.Queue(maxsize=2)
async with asyncio.TaskGroup() as group:
group.create_task(produce(queue))
consumer = group.create_task(consume(queue))
assert consumer.result() == list(range(6))
print(consumer.result()) # [0, 1, 2, 3, 4, 5]
if __name__ == "__main__":
asyncio.run(main())
입력은 0부터 5까지 여섯 값이고 None은 종료 표지다. 단일 소비자는 앞선 값을 순서대로 처리한 뒤 표지를 받고 끝난다. 부모는 TaskGroup을 나갈 때 두 작업의 종료를 기다리므로 이 흐름에서는 완료 항목 수를 따로 세는 join과 task_done을 쓰지 않았다.
소비자가 일반 예외로 실패하면 TaskGroup은 생산자도 취소한다. 두 작업을 함께 중단할 때는 부모 작업을 취소하는 등 수명을 함께 끝내는 경로를 쓴다. 개별 소비자의 CancelledError는 다른 자식을 취소하는 일반 실패와 다르게 처리되기 때문이다.[각주 11]
이 예제는 한 프로세스에서 큐 대기를 전달한다. Reactive Streams는 별도의 인터페이스와 신호 규칙으로 수요를 전달한다. 같은 문제를 다루되 각각의 API와 종료 계약을 따라 읽으면 된다.
큐가 찼을 때 바로 거절하는 동작도 비교해보자. 다음 Python 3 예제는 put_nowait을 쓰므로 두 번째 항목을 기다려 넣는 대신 QueueFull을 받아 접수하지 않았다고 알린다. 마지막에 꺼내는 값은 먼저 넣은 first다.[각주 12]
import asyncio
queue = asyncio.Queue(maxsize=1)
queue.put_nowait("first")
try:
queue.put_nowait("second")
except asyncio.QueueFull:
print("second: not accepted")
print(queue.get_nowait()) # first
같은 한 칸짜리 큐에 await put으로 넣으면 다음처럼 기다린다. Python 3.11 이상에서 실행하며, put_started로 두 번째 항목을 넣기 시작한 시점을 맞춘다. 첫 항목이 큐에 남아 있는 동안에는 두 번째 put이 완료되지 않는다.
import asyncio
async def main():
queue = asyncio.Queue(maxsize=1)
await queue.put("first")
put_started = asyncio.Event()
async def put_second():
put_started.set()
await queue.put("second")
async with asyncio.TaskGroup() as group:
pending = group.create_task(put_second())
await put_started.wait()
print(pending.done()) # False
first = await queue.get()
await pending
print(first, await queue.get()) # first second
if __name__ == "__main__":
asyncio.run(main())
첫 get이 항목을 꺼내면 생산자가 두 번째 항목을 넣을 공간이 생긴다. 여기서는 첫 항목을 꺼내 변수에 담기만 했고 별도의 업무 처리를 시작하지도 않았다. 큐의 빈칸을 기다리는 것과 소비자의 처리 완료를 기다리는 것이 어느 지점에서 갈리는지 볼 수 있다.
버퍼가 차면 무엇을 할 것인가
장치 | 직접 정하는 것 | 함께 정할 내용
버퍼 상한 | 기다리는 데이터를 얼마나 둘지 | 가득 찼을 때 기다릴지, 거절할지, 버릴지
레이트 리미터 | 일정 시간에 허용할 양 | 실제 소비 속도가 줄었을 때의 대응
동시 실행 상한 | 동시에 처리할 작업 수 | 실행을 기다리는 요청을 어디에 얼마나 둘지
드롭·접수 거절 | 받지 않거나 버릴 항목 | 누락과 재시도를 호출자에게 어떻게 알릴지
큐의 두 칸은 대기 중인 항목 수를 제한한다. 소비자가 이미 꺼내 처리하는 항목과 생산자가 다음에 넣으려고 들고 있는 항목은 그 바깥에 있다. 위 소비자의 values도 결과를 모으는 별도 목록으로 여기서는 여섯 값만 저장한다. 항목 수의 상한과 프로세스 전체 메모리의 바이트 상한은 이렇게 다른 범위를 가리킨다.
센서 측정이나 마우스 이동처럼 원래의 발생원을 늦출 수 없는 입력도 있다. 이때는 스트림 경계에서 잠시 보관하거나 값을 합치거나 일부를 버리는 정책을 선택한다. Reactive Streams 명세도 원래의 생산 속도를 바꿀 수 없는 소스에서는 수요 상한을 지키기 위해 버퍼링이나 드롭이 필요할 수 있다고 설명한다.[각주 13]
처리 능력을 넘는 입력이 계속 들어오면 유한한 저장소는 언젠가 찬다. 더 오래 보관할 장치를 붙이는 일과 지속적으로 감당할 처리량을 늘리는 일은 구별해야 하며 모든 입력을 보존해야 한다면 그만큼 생산을 늦추거나 처리 능력을 확보해야 한다.
TCP에서는 수신자가 윈도를 알리고 SEDA의 애플리케이션은 단계별 큐를 제어하며 스트림의 Subscriber는 받을 원소 수를 요청한다. 구현을 읽을 때 데이터가 가는 방향과 제약이 돌아오는 방향을 함께 따라가면 어느 생산자가 무엇을 기다리는지 확인할 수 있다.
