Pub-Sub — 실시간 메시징의 동작 원리와 한계

# 터미널 1 — 구독자
127.0.0.1:6379> SUBSCRIBE chat:room:1
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "chat:room:1"
3) (integer) 1    # 현재 구독 중인 채널 수

# 터미널 2 — 발행자
127.0.0.1:6379> PUBLISH chat:room:1 "안녕하세요!"
(integer) 1       # 메시지를 받은 구독자 수

# 터미널 1에서 수신
1) "message"
2) "chat:room:1"
3) "안녕하세요!"

127.0.0.1:6379> SUBSCRIBE chat:room:1 chat:room:2 notifications

127.0.0.1:6379> UNSUBSCRIBE chat:room:1
# 특정 채널만 해제

127.0.0.1:6379> UNSUBSCRIBE
# 모든 채널 해제

# chat:으로 시작하는 모든 채널 구독
127.0.0.1:6379> PSUBSCRIBE chat:*

# 메시지 수신 시 어떤 패턴/채널에서 왔는지 표시
1) "pmessage"
2) "chat:*"         # 매칭된 패턴
3) "chat:room:42"   # 실제 채널
4) "메시지 내용"

# * — 모든 문자 매칭
PSUBSCRIBE news:*          # news:sports, news:tech 등

# ? — 한 문자 매칭
PSUBSCRIBE chat:room:?     # chat:room:1, chat:room:2 (한 자리만)

# [abc] — 문자 클래스
PSUBSCRIBE log:[ew]*       # log:error, log:warn 등

허용: SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, PING, RESET
금지: GET, SET, HSET 등 모든 일반 명령

// Spring에서의 구현 — 별도 리스너 컨테이너
@Bean
public RedisMessageListenerContainer messageListenerContainer(
        RedisConnectionFactory connectionFactory) {
    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);  // 별도 연결 사용

    container.addMessageListener(chatMessageListener,
        new ChannelTopic("chat:room:1"));
    container.addMessageListener(notificationListener,
        new PatternTopic("notification:*"));

    return container;
}

1. 구독자가 없을 때
   PUBLISH chat:empty "아무도 없는 채널" → 메시지 사라짐

2. 구독자의 네트워크가 순간 끊겼을 때
   → 끊긴 동안의 메시지를 복구할 방법 없음

3. 구독자가 느려서 출력 버퍼가 가득 찰 때
   → Redis가 해당 클라이언트 연결을 끊어버림

# client-output-buffer-limit pubsub <hard> <soft> <seconds>
# hard: 즉시 연결 끊김, soft: seconds 동안 지속되면 끊김
client-output-buffer-limit pubsub 32mb 8mb 60
# 32MB 초과 시 즉시 끊김, 8MB 초과 상태가 60초 지속되면 끊김

# 기본적으로 비활성화 (CPU 오버헤드 때문)
127.0.0.1:6379> CONFIG SET notify-keyspace-events "KEA"
# K: Keyspace 이벤트 (__keyspace@<db>__)
# E: Keyevent 이벤트 (__keyevent@<db>__)
# A: 모든 이벤트 (g$lszhted의 별칭)

g — DEL, EXPIRE 등 일반 명령
$ — String 명령
l — List 명령
s — Set 명령
h — Hash 명령
z — Sorted Set 명령
x — 만료 이벤트
e — 퇴거(eviction) 이벤트
t — Stream 명령

# 설정
127.0.0.1:6379> CONFIG SET notify-keyspace-events "Ex"

# 구독
127.0.0.1:6379> SUBSCRIBE __keyevent@0__:expired

# 다른 터미널에서 TTL 설정
127.0.0.1:6379> SET session:abc "data" EX 5

# 5초 후 구독자에서 수신
1) "message"
2) "__keyevent@0__:expired"
3) "session:abc"

# Keyspace: "키 X에 무슨 일이 일어났나"
SUBSCRIBE __keyspace@0__:user:1001
# 수신: "set", "expire", "del" 등 (이벤트 이름)

# Keyevent: "어떤 이벤트가 어떤 키에서 일어났나"
SUBSCRIBE __keyevent@0__:set
# 수신: "user:1001", "user:1002" 등 (키 이름)

@Component
public class SessionExpirationListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String expiredKey = new String(message.getBody());
        if (expiredKey.startsWith("session:")) {
            String sessionId = expiredKey.substring("session:".length());
            // 세션 만료 후처리 로직
            auditService.logSessionExpired(sessionId);
            notificationService.notifyUserLogout(sessionId);
        }
    }
}

@Bean
public RedisMessageListenerContainer expirationListenerContainer(
        RedisConnectionFactory factory) {
    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(factory);
    container.addMessageListener(sessionExpirationListener,
        new PatternTopic("__keyevent@*__:expired"));
    return container;
}

// 캐시 갱신 시 모든 인스턴스에 알림
public void updateProduct(Long productId, ProductDto dto) {
    productRepository.update(productId, dto);
    localCache.invalidate("product:" + productId);

    // 다른 서버의 로컬 캐시도 무효화
    redisTemplate.convertAndSend("cache:invalidate",
        "product:" + productId);
}

// 리스너에서 로컬 캐시 무효화
@Override
public void onMessage(Message message, byte[] pattern) {
    String key = new String(message.getBody());
    localCache.invalidate(key);
}

// 알림 발송
public void sendNotification(Long userId, String content) {
    Notification noti = new Notification(userId, content);
    redisTemplate.convertAndSend("notification:user:" + userId,
        objectMapper.writeValueAsString(noti));
}

127.0.0.1:6379> SSUBSCRIBE chat:room:1    # Sharded 구독
127.0.0.1:6379> SPUBLISH chat:room:1 "hello"  # Sharded 발행