Pub/Sub과 Redis Streams: 차이점과 Spring Boot에서의 사용법

1. 개요

Redis는 Pub/Sub과 Streams라는 두 가지 메시징 메커니즘을 제공합니다. Redis를 처음 사용하는 분들은 이 두 가지를 헷갈리기 쉬워서 잘못된 도구를 선택하고 시스템 재시작 시 데이터를 잃는 문제가 발생합니다.

이 글에서는 Pub/Sub과 Streams의 핵심 차이를 설명하고, 두 가지 모두 작동하는 Java 코드 예제를 제공하며, 어느 것을 언제 사용해야 하는지 명확히 설명합니다. 이 글을 읽고 나면, 실제 메시징 문제에 맞는 올바른 모델을 자신 있게 선택할 수 있을 것입니다.

2. Redis Pub/Sub이란?

Pub/Sub(Publish/Subscribe)은 “fire-and-forget” 모델로 작동합니다. Publisher가 메시지를 채널로 보내면, Redis는 그 채널을 구독 중인 모든 subscriber에게 즉시 전달합니다. 구독 중인 subscriber가 없으면, 메시지는 영구적으로 사라집니다.

2.1. Maven 의존성

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

2.2. Publisher

@Service
public class OrderEventPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public void publish(String orderId, String status) {
        String message = "orderId=%s;status=%s".formatted(orderId, status);
        redisTemplate.convertAndSend("order-events", message);
    }
}

convertAndSend는 order-events 채널로 메시지를 보냅니다. 아직 구독자가 없다면, 명령은 성공하지만 메시지는 저장되지 않습니다.

2.3. Subscriber

@Component
public class OrderEventListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody());
        System.out.println("이벤트 수신: " + body);
    }
}

@Configuration
public class RedisPubSubConfig {

    @Bean
    public RedisMessageListenerContainer container(
            RedisConnectionFactory connectionFactory,
            OrderEventListener listener) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listener, new ChannelTopic("order-events"));
        return container;
    }
}

RedisMessageListenerContainer는 order-events 채널을 수신하기 위한 별도의 연결(blocking 연결 사용)을 관리합니다. Publisher가 메시지를 보내면 onMessage가 즉시 호출됩니다.

이벤트를 publish할 때의 출력:

이벤트 수신: orderId=ORD-001;status=CONFIRMED

주의: OrderEventListener가 아직 시작되지 않았다면(container가 bind되지 않음), 이전에 publish된 메시지는 완전히 손실되며 다시 재생할 수 없습니다.

3. Redis Streams란?

Stream은 순서가 있는 append-only 로그 데이터 구조로, 각 메시지는 <timestamp>-<sequence> 형태의 고유한 ID를 가집니다. Pub/Sub과 달리, Streams는 메시지를 Redis에 저장하며, 직접 삭제(XDEL) 또는 자르기(XTRIM)할 때까지 유지됩니다. Consumer는 처음부터 다시 읽거나, Kafka와 유사한 acknowledge 메커니즘이 있는 consumer group으로 읽을 수 있습니다.

3.1. Stream에 메시지 추가

@Service
public class OrderStreamPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public RecordId publish(String orderId, String status) {
        Map<String, String> payload = Map.of(
                "orderId", orderId,
                "status", status
        );
        return redisTemplate.opsForStream()
                .add("order-stream", payload);
    }
}

opsForStream().add는 order-stream으로 record를 쓰고 RecordId를 반환합니다 — record의 자동 생성된 primary key처럼 동작합니다. 이 record는 trim되거나 삭제될 때까지 Redis에 존재하며, 아무도 읽지 않았어도 유지됩니다.

3.2. Consumer Group으로 Stream 읽기

@Component
public class OrderStreamConsumer {

    private final StringRedisTemplate redisTemplate;

    public OrderStreamConsumer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        createGroupIfAbsent();
    }

    private void createGroupIfAbsent() {
        try {
            redisTemplate.opsForStream()
                    .createGroup("order-stream", "order-processors");
        } catch (Exception e) {
            // 그룹이 이미 존재합니다. 무시합니다.
        }
    }

    public void pollOnce() {
        List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream()
                .read(Consumer.from("order-processors", "worker-1"),
                        StreamReadOptions.empty().count(10),
                        StreamOffset.create("order-stream", ReadOffset.lastConsumed()));

        for (MapRecord<String, Object, Object> record : records) {
            System.out.println("record " + record.getId() + " 처리: " + record.getValue());
            redisTemplate.opsForStream().acknowledge("order-stream", "order-processors", record.getId());
        }
    }
}

createGroup은 consumer group order-processors를 생성합니다. Kafka의 consumer group과 유사합니다.
Consumer.from("order-processors", "worker-1")은 해당 그룹에 속한 worker-1이라는 이름의 consumer를 선언합니다. 같은 그룹의 여러 worker는 record를 나눠서 처리하므로 중복으로 읽지 않습니다.
ReadOffset.lastConsumed()는 그룹의 consumer에게 아직 전달되지 않은 새로운 record를 읽는다는 의미입니다.
acknowledge는 처리 완료를 확인하면 Redis가 그룹의 “pending list”에서 record를 제거합니다.

출력:

record 1696500000000-0 처리: {orderId=ORD-001, status=CONFIRMED}

주의: acknowledge를 호출하지 않으면 record는 pending list에 남아 있습니다. 다른 consumer 또는 이 consumer 자신(재시작 후)이 XCLAIM/readPending을 사용하여 다시 처리할 수 있습니다 — 이것이 Pub/Sub에는 없는 “at-least-once delivery” 메커니즘입니다.

4. Pub/Sub과 Streams 직접 비교

기준	Pub/Sub	Streams
메시지 저장소	아니요	예, trim/삭제될 때까지
이전 메시지 재생	아니요	예 (`XRANGE`, 사용자 정의 offset)
Consumer 오프라인 시 손실	오프라인 중 전송된 모든 메시지 손실	손실 없음, 온라인 복귀 후 다시 읽기
여러 consumer의 부하 분산	각 subscriber가 자신의 사본 수신(broadcast)	Consumer group이 작업을 분배, 각 record를 1개 consumer만 처리
Acknowledge / 재시도	없음	있음 (`XACK`, `XCLAIM`)
복잡도	매우 간단함	더 복잡함, Kafka와 유사
적합한 사용 사례	실시간 알림, 보증 불필요	Event sourcing, job queue, 처리 보증 필요

4.1. Pub/Sub 사용 시기

실시간으로 broadcast해야 하고 구독자가 없을 때 메시지 손실을 받아들일 수 있을 때 Pub/Sub을 사용합니다. 전형적인 예: 채팅의 “입력 중…” 상태 업데이트, 여러 애플리케이션 인스턴스 간의 cache 동기화(cache invalidation), 또는 몇 개를 놓쳐도 중요하지 않은 실시간 알림입니다.

@Service
public class CacheInvalidationPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public void invalidate(String cacheKey) {
        redisTemplate.convertAndSend("cache-invalidation", cacheKey);
    }
}

각 애플리케이션 인스턴스는 cache-invalidation 채널을 구독하여 로컬 cache에서 제거할 key를 받습니다. 한 인스턴스가 재시작 중일 때 이벤트를 놓쳐도, 잠시 동안만 오래된 cache를 사용할 뿐 심각한 오류가 아닙니다.

4.2. Streams 사용 시기

메시지가 완전히 처리되어야 하는 중요한 비즈니스 데이터인 경우 Streams을 사용합니다: 주문, 결제 거래, 감사 로그 등입니다. Consumer가 도중에 충돌해도 재시작 후 해당 record를 다시 처리할 수 있도록 보장해야 합니다.

@Service
public class PaymentEventPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public RecordId publish(String paymentId, BigDecimal amount) {
        Map<String, String> payload = Map.of(
                "paymentId", paymentId,
                "amount", amount.toString()
        );
        return redisTemplate.opsForStream().add("payment-stream", payload);
    }
}

각 payment event는 payment-stream에 기록됩니다. PaymentWorker가 배포를 위해 꺼져 있어도, 이벤트는 Redis에 대기하고 있으며, worker가 재시작되면 계속 읽을 수 있습니다.

5. Stream Consumer 테스트 작성

Streams는 상태가 영구적(persistent)이므로 publish – consume – acknowledge 흐름을 철저히 테스트해야 합니다. Testcontainers를 사용하여 실제 Redis를 실행하고 잘못된 모의 동작을 피하세요.

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>

@Testcontainers
class OrderStreamConsumerTest {

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

    StringRedisTemplate redisTemplate;

    @BeforeEach
    void setUp() {
        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration(
                redis.getHost(), redis.getFirstMappedPort());
        LettuceConnectionFactory factory = new LettuceConnectionFactory(config);
        factory.afterPropertiesSet();
        redisTemplate = new StringRedisTemplate(factory);
    }

    @Test
    void shouldConsumeAndAcknowledgeRecord() {
        OrderStreamPublisher publisher = new OrderStreamPublisher(redisTemplate);
        OrderStreamConsumer consumer = new OrderStreamConsumer(redisTemplate);

        publisher.publish("ORD-002", "SHIPPED");
        consumer.pollOnce();

        PendingMessages pending = redisTemplate.opsForStream()
                .pending("order-stream", Consumer.from("order-processors", "worker-1"),
                        Range.unbounded());

        assertThat(pending.getTotalPendingMessages()).isZero();
    }
}

이 테스트는 record를 publish하고, pollOnce()를 호출하여 consume 및 acknowledge한 후, pending list가 비어 있는지 확인합니다 — 즉, record가 처리 완료로 확인되었고 그룹에서 더 이상 대기 중이 아니라는 의미입니다.

6. 흔한 오류

중요한 데이터에 Pub/Sub 사용: 많은 초보자들이 주문 이벤트를 보내기 위해 Pub/Sub을 사용한 후, consumer 재시작으로 이벤트가 손실될 때 놀랍니다. Pub/Sub은 message queue가 아닙니다.
Stream을 trim하지 않음: Stream은 오래된 데이터를 자동으로 삭제하지 않습니다. 지속적으로 publish하면서 add할 때 XTRIM을 호출하거나 MAXLEN을 설정하지 않으면, Redis 메모리가 시간에 따라 계속 커집니다.

redisTemplate.opsForStream().add(
        StreamRecords.newRecord()
                .in("order-stream")
                .ofMap(payload),
        RedisStreamCommands.XAddOptions.maxlen(10_000)
);

읽기 전에 consumer group 생성 잊음: group이 없으면 XREADGROUP 명령이 오류를 발생합니다. 항상 “BUSYGROUP” 오류(group이 이미 있음)를 포착하는 try-catch로 createGroup을 호출하세요.
Pub/Sub을 위한 connection timeout 설정 안 함: subscriber는 blocking 연결을 사용합니다. 네트워크가 끊어졌는데 reconnect 메커니즘이 없으면, 그 시점부터 발견 및 재연결할 때까지 모든 메시지를 잃습니다.

7. 결론

Pub/Sub과 Streams는 Redis 내에 있지만 두 가지 다른 문제를 해결합니다. Pub/Sub은 간단하고 빠르지만 어떤 보증도 하지 않으므로 놓쳐도 되는 실시간 알림에 적합합니다. Streams는 더 복잡하지만 견고하며 consumer group과 acknowledge를 가지고 있어서, 완전히 처리되어야 하는 비즈니스 데이터에 적합하며 경량 message queue 같습니다.

시스템이 더 높은 신뢰성(exactly-once, 대규모 파티셔닝, 대규모 병렬 여러 consumer group)이 필요하다면 Redis Streams에 과도하게 의존하기보다는 Kafka 또는 RabbitMQ를 고려하세요.