Redis Streams를 이용한 컨슈머 그룹: 신뢰할 수 있는 이벤트 처리

1. 개요

Redis의 Pub/Sub에는 큰 단점이 있습니다. 게시(publish) 시점에 컨슈머가 꺼져있으면 메시지가 손실됩니다. Redis Streams는 이 문제를 해결합니다. 메시지는 stream에 지속적으로 저장되며, 각 컨슈머 그룹은 처리 완료된 메시지를 기억합니다.

이 글은 Spring Boot 3과 Spring Data Redis를 사용하여 주문 이벤트 처리 파이프라인을 구축하는 방법을 안내합니다. 파이프라인은 다음 단계들로 구성됩니다: 이벤트 기록, 컨슈머 그룹을 통한 읽기, 수동 ACK, 미처리 메시지 재수신 및 실패한 메시지를 dead-letter stream으로 이동.

2. Redis Streams와 컨슈머 그룹은 어떻게 작동하는가

Java를 작성하기 전에 redis-cli에서 기본 명령어를 시도해봐야 합니다:

XADD orders * orderId 1001 amount 250000
XGROUP CREATE orders shipping-group 0
XREADGROUP GROUP shipping-group worker-1 COUNT 1 STREAMS orders >
XPENDING orders shipping-group
XACK orders shipping-group 1728640000000-0

각 명령어의 의미:

• XADD는 stream orders에 항목을 추가합니다. * 문자는 Redis에게 <millis>-<seq> 형식의 ID를 자동으로 생성하도록 지시합니다.
• XGROUP CREATE ... 0은 group shipping-group을 만들고 stream의 처음부터 읽기를 시작합니다. $를 사용하면 그룹은 새로운 메시지만 읽습니다.
• XREADGROUP ... >는 그룹 내에서 아직 누구에게도 할당되지 않은 메시지를 가져옵니다. 메시지는 worker-1에게 할당되고 동시에 그룹의 **Pending Entries List (PEL)**에 기록됩니다.
• XPENDING은 할당되었지만 아직 ACK되지 않은 메시지의 개수를 보여줍니다.
• XACK은 메시지를 PEL에서 제거합니다. 이 시점부터 그룹은 메시지가 처리되었다고 봅니다.

ACK 이전의 XPENDING 출력은 다음과 같습니다:

1) (integer) 1
2) "1728640000000-0"
3) "1728640000000-0"
4) 1) 1) "worker-1"
      2) "1"

PEL은 신뢰성의 핵심입니다. 컨슈머가 메시지를 받은 후 ACK하기 전에 충돌한다고 가정합시다. 메시지는 여전히 PEL에 남아있으며, 다른 컨슈머가 그것을 받도록(claim) 할 수 있습니다. 이것이 at-least-once 의미론입니다.

3. 설치

Spring Data Redis 의존성을 추가합니다. 이 starter는 기본적으로 Lettuce를 클라이언트로 사용합니다:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

연결 설정과 stream 이름을 구성합니다:

spring:
  data:
    redis:
      host: localhost
      port: 6379

app:
  stream:
    key: orders
    group: shipping-group
    consumer: ${HOSTNAME:worker-local}

컨슈머 이름은 HOSTNAME에서 가져오므로 각 인스턴스(pod)는 고유한 이름을 가집니다. 이것은 매우 중요합니다. 두 인스턴스가 같은 이름을 사용하면 Redis는 그들을 하나의 컨슈머로 취급하고 PEL이 섞입니다.

4. Stream에 이벤트 기록

@Service
public class OrderEventPublisher {

    private final StringRedisTemplate redisTemplate;
    private final String streamKey;

    public OrderEventPublisher(StringRedisTemplate redisTemplate,
                               @Value("${app.stream.key}") String streamKey) {
        this.redisTemplate = redisTemplate;
        this.streamKey = streamKey;
    }

    public RecordId publishOrderCreated(String orderId, long amount) {
        MapRecord<String, String, String> record = StreamRecords.newRecord()
                .in(streamKey)
                .ofMap(Map.of(
                        "type", "ORDER_CREATED",
                        "orderId", orderId,
                        "amount", String.valueOf(amount)));

        return redisTemplate.opsForStream().add(record);
    }
}

StreamRecords.newRecord().in(...).ofMap(...)는 문자열 필드를 가진 MapRecord를 생성합니다. opsForStream().add() 명령어는 XADD orders * ...와 동등하며 Redis가 생성한 RecordId를 반환합니다. 예: 1728640000000-0.

StringRedisTemplate을 사용하면 Redis의 데이터를 redis-cli로 읽을 수 있습니다. JDK serialization의 바이트 문자열을 다룰 필요가 없습니다.

5. 안전한 컨슈머 그룹 생성

읽기 전에 그룹이 존재해야 합니다. 애플리케이션이 stream에 아직 메시지가 없을 때 시작될 수 있으므로 MKSTREAM 파라미터가 필요합니다. StreamOperations의 createGroup() 메서드는 이 파라미터를 활성화하지 않습니다. 따라서 connection 레벨로 내려가서 호출합니다:

@Component
public class StreamGroupInitializer {

    private final StringRedisTemplate redisTemplate;

    public StreamGroupInitializer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void ensureGroup(String streamKey, String group) {
        try {
            redisTemplate.execute((RedisCallback<String>) connection ->
                    connection.streamCommands().xGroupCreate(
                            streamKey.getBytes(StandardCharsets.UTF_8),
                            group,
                            ReadOffset.from("0"),
                            true)); // MKSTREAM
        } catch (DataAccessException e) {
            String msg = NestedExceptionUtils.getMostSpecificCause(e).getMessage();
            if (msg == null || !msg.contains("BUSYGROUP")) {
                throw e;
            }
            // 그룹이 이미 존재함: 애플리케이션 재시작 시 정상
        }
    }
}

마지막 파라미터 true는 MKSTREAM을 활성화합니다. Redis는 stream이 없으면 빈 stream을 생성합니다. 애플리케이션 재시작 시 Redis는 BUSYGROUP Consumer Group name already exists 오류를 반환합니다. 이 특정 오류는 무시하고 다른 오류는 다시 throw합니다.

6. StreamMessageListenerContainer로 메시지 읽기

6.1. 수동 ACK를 가진 Listener

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

    private final StringRedisTemplate redisTemplate;
    private final ShippingService shippingService;
    private final String streamKey;
    private final String group;

    public OrderEventListener(StringRedisTemplate redisTemplate,
                              ShippingService shippingService,
                              @Value("${app.stream.key}") String streamKey,
                              @Value("${app.stream.group}") String group) {
        this.redisTemplate = redisTemplate;
        this.shippingService = shippingService;
        this.streamKey = streamKey;
        this.group = group;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> record) {
        Map<String, String> body = record.getValue();
        shippingService.createShipment(body.get("orderId"));

        redisTemplate.opsForStream().acknowledge(streamKey, group, record.getId());
    }
}

ACK 명령어는 createShipment이 성공한 후에만 실행됩니다. 메서드가 예외를 throw하면 메시지는 여전히 PEL에 남아있어 나중에 재처리됩니다. 처리하기 전에 ACK하지 마세요. 그렇게 하면 at-least-once가 at-most-once로 변합니다.

ShippingService 이 글의 간단한 서비스입니다. 배송 전표를 생성한 orderId들을 기록합니다:

@Service
public class ShippingService {

    private final Set<String> shippedOrders = ConcurrentHashMap.newKeySet();

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // 이미 처리됨, 무시 (idempotent)
        }
        // 배송 회사 API 호출...
    }

    public Set<String> shippedOrders() {
        return Set.copyOf(shippedOrders);
    }
}

6.2. Container 설정

@Configuration
public class StreamConfig {

    @Bean(destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>>
    orderStreamContainer(RedisConnectionFactory connectionFactory,
                         StreamGroupInitializer initializer,
                         OrderEventListener listener,
                         @Value("${app.stream.key}") String streamKey,
                         @Value("${app.stream.group}") String group,
                         @Value("${app.stream.consumer}") String consumerName) {

        initializer.ensureGroup(streamKey, group);

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

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

        var request = StreamMessageListenerContainer.StreamReadRequest
                .builder(StreamOffset.create(streamKey, ReadOffset.lastConsumed()))
                .consumer(Consumer.from(group, consumerName))
                .autoAcknowledge(false)
                .cancelOnError(t -> false)
                .build();

        container.register(request, listener);
        container.start();
        return container;
    }
}

설정의 주요 포인트:

• ReadOffset.lastConsumed()는 XREADGROUP의 >에 해당하며, 이는 그룹의 새로운 메시지만 가져오는 것을 의미합니다.
• pollTimeout은 각 읽기의 block 시간입니다. batchSize는 한 번의 읽기에서 가져올 최대 메시지 수입니다.
• autoAcknowledge(false)는 자동 ACK를 비활성화하므로 listener에서 직접 ACK합니다.
• cancelOnError(t -> false)는 가장 중요한 라인입니다. 기본적으로 listener가 예외를 throw하면 container는 subscription을 취소합니다. 컨슈머는 명확한 경고 없이 조용히 읽기를 중단합니다.

참고: 중요한 비즈니스 로직에는 receiveAutoAck(...)을 사용하지 마세요. 이 모드는 읽기 직후 ACK하므로 listener 실행 전에 ACK합니다. 애플리케이션이 중간에 충돌하면 메시지는 영구적으로 손실됩니다.

7. 미처리 메시지 및 dead-letter 처리

Listener는 새로운 메시지만 읽습니다(>). 오류가 있는 메시지나 죽은 컨슈머의 메시지는 아무도 처리하지 않으면 PEL에 영구적으로 남아있습니다. 정기적으로 실행되는 job이 필요합니다:

@Component
public class PendingOrderReclaimer {

    private static final Duration MIN_IDLE = Duration.ofMinutes(1);
    private static final long MAX_DELIVERIES = 5;

    private final StringRedisTemplate redisTemplate;
    private final ShippingService shippingService;
    private final String streamKey;
    private final String group;
    private final String consumerName;

    public PendingOrderReclaimer(StringRedisTemplate redisTemplate,
                                 ShippingService shippingService,
                                 @Value("${app.stream.key}") String streamKey,
                                 @Value("${app.stream.group}") String group,
                                 @Value("${app.stream.consumer}") String consumerName) {
        this.redisTemplate = redisTemplate;
        this.shippingService = shippingService;
        this.streamKey = streamKey;
        this.group = group;
        this.consumerName = consumerName;
    }

    @Scheduled(fixedDelay = 30_000)
    public void reclaim() {
        StreamOperations<String, String, String> ops = redisTemplate.opsForStream();
        PendingMessages pending = ops.pending(streamKey, group, Range.unbounded(), 50);

        for (PendingMessage pm : pending) {
            if (pm.getElapsedTimeSinceLastDelivery().compareTo(MIN_IDLE) < 0) {
                continue;
            }
            List<MapRecord<String, String, String>> claimed =
                    ops.claim(streamKey, group, consumerName, MIN_IDLE, pm.getId());

            for (MapRecord<String, String, String> record : claimed) {
                if (pm.getTotalDeliveryCount() >= MAX_DELIVERIES) {
                    moveToDeadLetter(record);
                } else {
                    shippingService.createShipment(record.getValue().get("orderId"));
                }
                ops.acknowledge(streamKey, group, record.getId());
            }
        }
    }

    private void moveToDeadLetter(MapRecord<String, String, String> record) {
        Map<String, String> body = new HashMap<>(record.getValue());
        body.put("originalId", record.getId().getValue());
        redisTemplate.opsForStream().add(
                StreamRecords.newRecord().in(streamKey + ":dlq").ofMap(body));
    }
}

Job의 처리 흐름:

1. pending(...)은 XPENDING orders shipping-group - + 50과 동등합니다. 각 PendingMessage는 어느 컨슈머가 메시지를 보유하고 있는지, idle 시간 및 할당 횟수를 나타냅니다.
2. claim(...)은 XCLAIM과 동등합니다. 이 명령어는 소유권을 현재 컨슈머로 이전합니다. Redis는 메시지가 최소 MIN_IDLE 동안 idle 상태인 경우에만 이전하므로 두 인스턴스가 동시에 job을 실행해도 같은 메시지를 중복으로 claim하지 않습니다. 각 claim도 delivery count를 증가시킵니다.
3. MAX_DELIVERIES 횟수를 초과한 메시지는 orders:dlq stream으로 복사된 후 ACK됩니다. 영구적으로 실패하는 메시지(poison message)는 무한 재시도를 피합니다.

createShipment이 예외를 throw하면 ACK 명령어가 실행되지 않습니다. 메시지는 PEL에 남아있고 다음 주기에 재시도됩니다. @Configuration 클래스에 @EnableScheduling을 활성화해야 합니다.

참고: Redis 6.2부터는 XAUTOCLAIM 명령어가 있습니다. 이 명령어는 XPENDING과 XCLAIM 단계를 결합하며, PEL이 클 때 더 효율적입니다. 현재 사용 중인 Spring Data Redis 버전이 편리한 API를 아직 제공하지 않으면 redisTemplate.execute(...)를 통해 호출할 수 있습니다.

8. Testcontainers로 테스트 작성

테스트용 의존성을 추가합니다:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-testcontainers</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

아래 테스트는 이벤트를 게시합니다. 그 다음 배송 전표가 생성되었고 PEL이 비어있는지 확인합니다:

@SpringBootTest
@Testcontainers
class OrderStreamIntegrationTest {

    @Container
    @ServiceConnection(name = "redis")
    static GenericContainer<?> redis =
            new GenericContainer<>("redis:7-alpine").withExposedPorts(6379);

    @Autowired OrderEventPublisher publisher;
    @Autowired ShippingService shippingService;
    @Autowired StringRedisTemplate redisTemplate;

    @Test
    void whenOrderPublished_thenShipmentCreatedAndMessageAcked() {
        publisher.publishOrderCreated("1001", 250_000);

        await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
            assertThat(shippingService.shippedOrders()).contains("1001");

            PendingMessagesSummary summary = redisTemplate.opsForStream()
                    .pending("orders", "shipping-group");
            assertThat(summary.getTotalPendingMessages()).isZero();
        });
    }
}

@ServiceConnection(name = "redis")는 Spring Boot가 컨테이너에서 host와 port를 자동으로 가져오게 합니다. @DynamicPropertySource가 필요 없습니다. await()는 listener가 별도 thread에서 실행되기 때문에 필요합니다. getTotalPendingMessages() assertion이 0인 것은 메시지가 읽혔을 뿐만 아니라 ACK되었음을 증명합니다.

9. 흔한 오류와 언제 사용하면 안 되는지

멱등성이 없는 처리. At-least-once는 메시지가 두 번 올 수 있다는 뜻입니다. 예를 들어: 컨슈머가 처리를 끝낸 후 ACK하기 전에 충돌합니다. 따라서 비즈니스 로직은 반복 실행을 견디어야 합니다. 일반적인 방법은 orderId가 이미 처리되었는지 확인하거나 데이터베이스의 unique constraint를 사용하는 것입니다.

Stream이 무한정 커집니다. ACK는 stream에서 항목을 삭제하지 않습니다. 기록할 때 크기를 제한해야 합니다:

redisTemplate.opsForStream().add(record, XAddOptions.maxlen(100_000).approximateTrimming(true));

이 옵션은 XADD orders MAXLEN ~ 100000 * ...와 동등합니다. ~ 문자는 Redis가 내부 노드에 따라 자르도록 허용하여 정확한 자르기보다 훨씬 빠릅니다.

cancelOnError 잊기. 첫 예외가 발생하면 컨슈머가 조용히 중단됩니다. 증상은 메시지가 stream에 쌓이지만 로그에 명확한 오류가 없는 것입니다.

“유령” 컨슈머. 배포할 때마다 새로운 컨슈머 이름(예: pod 이름)으로 실행하면 이전 이름은 그룹에 남아있습니다. XGROUP DELCONSUMER를 사용하여 정기적으로 정리하세요. 다만 해당 컨슈머의 모든 pending 메시지를 claim한 후에만 정리합니다.

Redis Streams를 사용하면 안 되는 경우:

• 여러 날 동안 이벤트를 저장하고 대량의 메시지를 재생해야 합니다. Redis는 모든 데이터를 RAM에 유지하므로 비용이 매우 높습니다.
• 데이터 손실을 절대 허용할 수 없습니다. AOF everysec 또는 비동기 failover replica를 사용해도 최근에 기록한 일부 메시지가 손실될 수 있습니다.
• 처리량이 매우 크고 수백 개의 컨슈머로 분할 및 확장이 필요합니다. 이 경우 Kafka가 더 적합합니다.

10. 결론

Redis Streams를 컨슈머 그룹과 결합하면 이미 가지고 있는 Redis에서 바로 실행되는 가볍고 신뢰할 수 있는 이벤트 큐를 얻습니다. 네 가지 핵심 사항을 기억해야 합니다:

• 처리가 완료된 후에만 수동으로 ACK합니다.
• cancelOnError(t -> false)를 설정하여 컨슈머가 조용히 멈추지 않도록 합니다.
• 미처리 메시지를 claim하는 job과 poison message용 dead-letter stream을 가집니다.
• 처리 로직을 멱등성 있게 작성합니다.

MAXLEN을 추가하여 메모리를 제어하고 Testcontainers로 통합 테스트하면, 대부분의 Spring Boot 애플리케이션에 충분히 견고한 at-least-once 이벤트 처리 파이프라인을 갖게 됩니다.