Implementing a Queue With Redis Lists and Streams

목표

Redis 리스트와 스트림으로 각각 큐를 만들어, 리스트 큐가 메시지를 잃는 지점을 재현하고 스트림의 컨슈머 그룹과 PEL 로 그것을 막는다.

왜 중요한가

LPUSH / BRPOP 두 줄이면 큐가 됩니다. 그래서 많은 팀이 여기서 멈추고, 배포 때마다 처리 중이던 작업이 조용히 사라지는 것을 몇 달 뒤에 발견합니다. BRPOP 이 반환한 순간 메시지는 이미 Redis 에서 지워졌기 때문입니다. 이 실습은 그 유실을 먼저 눈으로 만든 뒤 두 가지 해법을 순서대로 붙입니다. LMOVE 로 처리 중 리스트를 두는 손수 만든 방법과, 처음부터 이 문제를 위해 설계된 스트림의 컨슈머 그룹입니다. 이 둘의 차이를 알면 SQS 의 가시성 타임아웃이나 Kafka 의 오프셋 커밋을 처음 볼 때도 무슨 문제를 푸는 장치인지 바로 보입니다.

단계

1. redis-cli PING 결과를 /root/q/ping.txt 에 저장한다. PONG 이 들어 있어야 한다.
2. /root/q/produce.py 로 q:jobs 에 job-1 부터 job-5 까지 5건을 넣는다. LLEN q:jobs 가 5 이다.
3. /root/q/consume.py 로 5건을 모두 꺼내 /root/q/order.out 에 한 줄씩 적는다. 첫 줄이 job-1, 마지막 줄이 job-5 여야 한다.
4. /root/q/safe_consume.py 는 LMOVE 로 q:jobs 에서 q:jobs:processing 으로 원자적으로 옮긴 뒤 처리하고, 성공하면 처리 중 리스트에서 지운다. 처리 도중 죽는 경우를 흉내 내어 q:jobs:processing 에 1건이 남게 만든다.
5. /root/q/stream_add.py 로 스트림 q:orders 에 5건을 XADD 한다. MAXLEN ~ 1000 상한을 건다. XLEN q:orders 가 5 이다.
6. 컨슈머 그룹 g1 을 만들고 /root/q/stream_consume.py 로 5건을 읽어 그중 4건만 XACK 한다.
7. XPENDING q:orders g1 결과를 /root/q/pending.txt 에 저장한다. 미확인 메시지가 정확히 1건이어야 한다.
8. /root/q/compare.md 에 마크다운 표를 쓴다. 첫 열의 행 제목은 소비 후 보존, 다중 소비자 그룹, 실패 회수, 메모리 네 개이고 리스트와 스트림 열이 있어야 한다.

참고

• LMOVE q:jobs q:jobs:processing RIGHT LEFT 는 꺼내기와 옮기기를 한 번에 합니다.
• 그룹 생성: XGROUP CREATE q:orders g1 0
• 미확인 조회: XPENDING q:orders g1
• 흔한 실수 1: LPUSH 로 넣고 LPOP 으로 꺼내는 것 — 그러면 LIFO 가 됩니다.
• 흔한 실수 2: 스트림에 MAXLEN 을 안 걸어 메모리가 무한히 자라는 것.

Redis 연결 확인하기

redis-cli PING 결과를 /root/q/ping.txt 에 저장한다. PONG 이 들어 있어야 한다.

redis-cli 로 응답을 확인하고 결과를 파일로 남깁니다. 127.0.0.1:6379 에 이미 떠 있습니다.

리스트로 큐에 넣기

/root/q/produce.py 로 q:jobs 에 job-1 부터 job-5 까지 5건을 넣는다. LLEN q:jobs 가 5 이다.

한쪽 끝으로 넣고 반대쪽 끝으로 꺼내야 FIFO 가 됩니다. 어느 쪽으로 넣을지가 중요합니다.

소비 순서가 FIFO 임을 증명하기

/root/q/consume.py 로 5건을 모두 꺼내 /root/q/order.out 에 한 줄씩 적는다. 첫 줄이 job-1, 마지막 줄이 job-5 여야 한다.

넣은 순서와 꺼낸 순서를 각각 파일로 남겨 비교하세요. 순서가 뒤집혔다면 넣는 방향과 꺼내는 방향이 같은 쪽입니다.

유실을 막는 처리중 리스트 도입하기

/root/q/safe_consume.py 는 LMOVE 로 q:jobs 에서 q:jobs:processing 으로 원자적으로 옮긴 뒤 처리하고, 성공하면 처리 중 리스트에서 지운다. 처리 도중 죽는 경우를 흉내 내어 q:jobs:processing 에 1건이 남게 만든다.

꺼내기와 옮기기를 한 명령으로 해야 원자적입니다. 두 명령으로 나누면 그 사이에 죽을 수 있습니다.

스트림에 메시지 추가하기

/root/q/stream_add.py 로 스트림 q:orders 에 5건을 XADD 한다. MAXLEN ~ 1000 상한을 건다. XLEN q:orders 가 5 이다.

필드-값 쌍으로 저장됩니다. 무한히 자라지 않도록 상한도 함께 지정하세요.

컨슈머 그룹으로 소비하고 확인응답 보내기

컨슈머 그룹 g1 을 만들고 /root/q/stream_consume.py 로 5건을 읽어 그중 4건만 XACK 한다.

그룹을 먼저 만들어야 읽을 수 있습니다. 읽기만 하면 PEL 에 남고, 확인응답을 보내야 지워집니다.

확인응답 없는 메시지 확인하기

XPENDING q:orders g1 결과를 /root/q/pending.txt 에 저장한다. 미확인 메시지가 정확히 1건이어야 한다.

일부러 하나만 확인응답을 보내지 않으면 됩니다. 미확인 메시지 수를 조회하는 명령이 있습니다.

리스트와 스트림 비교표 쓰기

/root/q/compare.md 에 마크다운 표를 쓴다. 첫 열의 행 제목은 소비 후 보존, 다중 소비자 그룹, 실패 회수, 메모리 네 개이고 리스트와 스트림 열이 있어야 한다.

소비 후 보존, 다중 그룹, 실패 회수, 메모리 네 축으로 정리합니다. 표 형식과 행 제목이 채점 기준입니다.