Redis Streams로 이벤트 스트림 처리하기

Radix Tree
├── 1710000000000-*    → Listpack [msg1, msg2, ..., msgN]
├── 1710000001000-*    → Listpack [msg1, msg2, ...]
└── ...

# 스트림에 이벤트 추가 (ID는 Redis가 자동 생성)
> XADD orders * orderId ORD-001 product "MacBook" qty 1
"1710000000000-0"

> XADD orders * orderId ORD-002 product "iPad" qty 2
"1710000000001-0"

# 스트림 길이 확인
> XLEN orders
(integer) 2

# 범위 조회 (전체)
> XRANGE orders - +
1) 1) "1710000000000-0"
   2) 1) "orderId" 2) "ORD-001" 3) "product" 4) "MacBook" 5) "qty" 6) "1"
2) 1) "1710000000001-0"
   2) 1) "orderId" 2) "ORD-002" 3) "product" 4) "iPad" 5) "qty" 6) "2"

# 특정 ID 이후만 조회
> XRANGE orders 1710000000001-0 +
# → ORD-002만 반환

# 역순 조회 (최신 1개)
> XREVRANGE orders + - COUNT 1

# Consumer Group 생성 (0 = 스트림 처음부터)
> XGROUP CREATE orders order-processors 0 MKSTREAM

# 컨슈머 worker-1이 새 메시지 읽기
> XREADGROUP GROUP order-processors worker-1 COUNT 10 BLOCK 5000 STREAMS orders >

# 처리 완료 후 ack
> XACK orders order-processors 1710000000000-0

[Producer] ──XADD──► [Stream] ──XREADGROUP──► [Consumer]
                                                  │
                                              처리 중...
                                                  │
                                           ┌──────┴──────┐
                                           │    정상      │   장애
                                           │   XACK ✓    │   PEL에 남음
                                           └─────────────┘

# 그룹 전체 pending 요약
> XPENDING orders order-processors
1) (integer) 3              # pending 메시지 수
2) "1710000000000-0"         # 가장 오래된 pending ID
3) "1710000000002-0"         # 가장 새로운 pending ID
4) 1) 1) "worker-1"         # 컨슈머별 pending 수
      2) "2"
   2) 1) "worker-2"
      2) "1"

# 특정 컨슈머의 pending 상세 (최대 10개)
> XPENDING orders order-processors - + 10 worker-1
1) 1) "1710000000000-0"      # 메시지 ID
   2) "worker-1"             # 소유 컨슈머
   3) (integer) 300000       # idle 시간 (ms) — 300초 동안 ack 안 됨
   4) (integer) 1            # 전달 횟수

# 수동 재할당: 5분(300000ms) 이상 idle인 메시지를 worker-2로
> XCLAIM orders order-processors worker-2 300000 1710000000000-0

# 자동 재할당 (Redis 6.2+): 5분 이상 idle인 것을 자동으로
> XAUTOCLAIM orders order-processors worker-2 300000 0-0 COUNT 10

@Service
@RequiredArgsConstructor
public class OrderEventConsumer {

    private final StringRedisTemplate redis;
    private final OrderService orderService;

    /**
     * 멱등 키: orderId
     * TTL: 24시간 (재처리 윈도우 내 중복 차단)
     */
    public void process(MapRecord<String, String, String> record) {
        String orderId = record.getValue().get("orderId");
        String idempotencyKey = "processed:order:" + orderId;

        // 1) 이미 처리했으면 skip + ack
        Boolean alreadyProcessed = redis.opsForValue()
            .setIfAbsent(idempotencyKey, "1", Duration.ofHours(24));
        if (Boolean.FALSE.equals(alreadyProcessed)) {
            log.info("중복 메시지 skip: orderId={}", orderId);
            ack(record);
            return;
        }

        try {
            // 2) 비즈니스 로직
            orderService.fulfill(orderId);
            // 3) 성공 시 ack
            ack(record);
        } catch (Exception e) {
            // 실패 시 멱등 키 삭제 → 재처리 허용
            redis.delete(idempotencyKey);
            log.error("처리 실패, 재처리 대상: orderId={}", orderId, e);
        }
    }

    private void ack(MapRecord<String, String, String> record) {
        redis.opsForStream().acknowledge("orders", "order-processors", record.getId());
    }
}

@Entity
@Table(uniqueConstraints = @UniqueConstraint(columnNames = "orderId"))
public class ProcessedEvent {
    @Id @GeneratedValue
    private Long id;

    @Column(nullable = false)
    private String orderId;

    private Instant processedAt;
}

// 사용: save 시 DataIntegrityViolationException → 중복으로 판단
@Transactional
public void processWithDbGuard(String orderId) {
    try {
        processedEventRepo.save(new ProcessedEvent(orderId, Instant.now()));
        orderService.fulfill(orderId);
    } catch (DataIntegrityViolationException e) {
        log.info("DB 유니크 제약으로 중복 차단: orderId={}", orderId);
    }
}

1710000000000-0
│              │
└── 밀리초 타임스탬프  └── 시퀀스 (같은 ms 내 구분)

# ❌ 위험: 과거 타임스탬프 ID를 삽입하려고 하면 에러
> XADD orders 1700000000000-0 orderId ORD-OLD
(error) ERR The ID specified ... is equal or smaller than ...

# 최대 10만 건 유지 (정확)
> XTRIM orders MAXLEN 100000

# 근사 트림 (더 빠름, ~100000 근처로)
> XTRIM orders MAXLEN ~ 100000

# ID 기반 트림: 특정 시점 이전 삭제 (Redis 6.2+)
> XTRIM orders MINID 1710000000000-0

# XADD 시 동시 트림 (쓰기+정리 원자적)
> XADD orders MAXLEN ~ 100000 * orderId ORD-003 product "AirPods" qty 1

<!-- build.gradle / pom.xml -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

@Configuration
public class RedisStreamConfig {

    @Bean
    public Subscription orderStreamSubscription(
            RedisConnectionFactory factory,
            OrderStreamListener listener) {

        var options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
            .builder()
            .pollTimeout(Duration.ofSeconds(2))
            .batchSize(10)
            .build();

        var container = StreamMessageListenerContainer.create(factory, options);

        // Consumer Group이 없으면 생성
        try {
            factory.getConnection().streamCommands()
                .xGroupCreate("orders".getBytes(), "order-processors",
                    ReadOffset.from("0"), true);
        } catch (RedisSystemException e) {
            // BUSYGROUP: 이미 존재
        }

        var subscription = container.receive(
            Consumer.from("order-processors", "worker-" + UUID.randomUUID()),
            StreamOffset.create("orders", ReadOffset.lastConsumed()),
            listener
        );

        container.start();
        return subscription;
    }
}

@Component
@Slf4j
public class OrderStreamListener
        implements StreamListener<String, MapRecord<String, String, String>> {

    @Autowired private StringRedisTemplate redis;
    @Autowired private OrderService orderService;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String orderId = message.getValue().get("orderId");
        log.info("수신: streamId={}, orderId={}", message.getId(), orderId);

        try {
            orderService.fulfill(orderId);
            redis.opsForStream()
                .acknowledge("orders", "order-processors", message.getId());
            log.info("처리 완료 + ACK: orderId={}", orderId);
        } catch (Exception e) {
            log.error("처리 실패 (PEL에 남음): orderId={}", orderId, e);
            // ack하지 않으면 PEL에 남아 재처리 대상
        }
    }
}

@Service
@RequiredArgsConstructor
public class OrderEventProducer {

    private final StringRedisTemplate redis;

    public RecordId publish(String orderId, String product, int qty) {
        var record = StreamRecords.newRecord()
            .ofMap(Map.of(
                "orderId", orderId,
                "product", product,
                "qty", String.valueOf(qty),
                "timestamp", Instant.now().toString()
            ))
            .withStreamKey("orders");

        RecordId recordId = redis.opsForStream().add(record);
        log.info("이벤트 발행: streamId={}, orderId={}", recordId, orderId);
        return recordId;
    }
}

# 스트림 상세 정보
> XINFO STREAM orders FULL

# 그룹별 상태
> XINFO GROUPS orders

# 컨슈머별 상태
> XINFO CONSUMERS orders order-processors

# 죽은 컨슈머 삭제
> XGROUP DELCONSUMER orders order-processors dead-worker-1

# 그룹 오프셋 리셋 (주의: 전체 재처리)
> XGROUP SETID orders order-processors 0

# 1) 그룹이 존재하는지 확인
> XINFO GROUPS orders

# 2) 컨슈머가 살아있는지 확인
> XINFO CONSUMERS orders order-processors
# → idle 시간이 매우 크면 죽은 컨슈머

# 3) 그룹의 last-delivered-id 확인
> XINFO GROUPS orders
# → last-delivered-id가 스트림 끝보다 훨씬 뒤면 정상
# → 같으면 소비자가 메시지를 가져가지 않는 것