9편. Redis Pub/Sub 실습: 실시간 메시지 전달

반응형

 

9편. Redis Pub/Sub 실습: 실시간 메시지 전달

 

📚 목차
1. 실시간 알림 발행하기
2. 캐시 무효화 메시지 발행하기
3. 채팅 메시지 브로드캐스트하기
4. 관리자 공지 발행하기

 

 

 

 

1. 실시간 알림 발행하기

실시간 알림은 Pub/Sub을 설명하기 가장 좋은 예제입니다.

예를 들어 다음 상황을 생각해 볼 수 있습니다.

- 게시글에 좋아요가 눌림
- 댓글이 작성됨
- 주문 상태가 변경됨
- 관리자 메시지가 도착함

이런 이벤트는 DB에 저장할 수도 있지만, 현재 접속 중인 사용자에게 즉시 알려야 할 때는 Pub/Sub이 유용합니다.

전체 흐름은 다음과 같습니다.

NotificationService
  ↓
Redis PUBLISH channel:notification
  ↓
Notification Subscriber
  ↓
WebSocket / SSE / Console Log

 

✔️ 실시간 알림 발행

// src/services/notification-pubsub.service.ts

/**
 * 실시간 알림을 공용 알림 채널에 발행합니다.
 *
 * 1. 알림 메시지를 JSON 문자열로 변환합니다.
 * 2. 알림 채널을 구독 중인 모든 구독자에게 문자열을 발행합니다.
 * 3. 메시지를 전달받은 구독자 수를 반환합니다.
 *
 * @returns 메시지를 받은 subscriber 수
 */
async publishNotification(message: RealtimeNotificationMessage): Promise<number> {
  const channel = RedisKey.channel.notification();

  const payload = JSON.stringify(message);

  // 직렬화한 알림을 현재 연결된 구독자에게 전달합니다.
  // 메시지를 발행하고 이를 전달받은 구독자 수를 반환합니다.
  return redis.publish(channel, payload);
}

/**
 * 게시글 좋아요 정보를 실시간 알림으로 발행합니다.
 *
 * 1. 수신자와 게시글 정보를 좋아요 알림 메시지로 구성합니다.
 * 2. 현재 시각을 알림 생성 시각으로 기록합니다.
 * 3. 공용 알림 발행 메서드에 메시지를 전달합니다.
 */
async publishPostLikedNotification(input: {
  receiverUserId: number;
  postId: number;
  likedByUserName: string;
}): Promise<number> {
  return this.publishNotification({
    type: 'POST_LIKED',
    userId: input.receiverUserId,
    title: '게시글 좋아요 알림',
    message: `${input.likedByUserName}님이 ${input.postId}번 게시글을 좋아합니다.`,
    createdAt: new Date().toISOString(),
  });
}

 

알림 발행의 핵심 코드는 다음입니다.

const channel = RedisKey.channel.notification();
const payload = JSON.stringify(message);
return redis.publish(channel, payload);

 

Redis Pub/Sub 메시지는 문자열로 전달됩니다.

따라서 객체 메시지는 JSON.stringify()로 문자열 변환 후 publish합니다.

channel:notification
  ↓
{
  "type": "POST_LIKED",
  "userId": 1,
  "title": "게시글 좋아요 알림",
  "message": "Kim님이 10번 게시글을 좋아합니다.",
  "createdAt": "2026-07-27T00:00:00.000Z"
}

redis.publish()의 반환값은 메시지를 받은 subscriber 수입니다.
예를 들어 channel:notification을 구독 중인 subscriber가 2개라면 반환값은 2입니다.

 

✔️ 실시간 알림 채널 구독

/**
 * 실시간 알림 채널 구독을 시작합니다.
 *
 * 1. 일반 명령용 연결과 분리된 구독 전용 클라이언트를 생성합니다.
 * 2. 알림 채널에서 받은 JSON 문자열을 알림 객체로 변환합니다.
 * 3. 변환한 알림을 콜백에 전달하고 구독 종료 함수를 반환합니다.
 *
 * 참고:
 * Pub/Sub 모드의 연결은 일반 Redis 명령 처리에 함께 사용하지 않습니다.
 *
 * @param onMessage 알림 메시지를 받았을 때 실행할 콜백
 * @returns 구독 종료 함수
 */
async subscribeNotification(
  onMessage: (message: RealtimeNotificationMessage) => void | Promise<void>,
): Promise<() => Promise<void>> {
  const channel = RedisKey.channel.notification();

  const subscriber = redis.duplicate();
  await subscriber.connect();

  // 알림 채널에서 새 메시지를 실시간으로 수신합니다.
  // 구독이 유지되는 동안 메시지를 받을 때마다 등록한 콜백을 실행합니다.
  await subscriber.subscribe(channel, async (rawMessage) => {
    try {
      const message = JSON.parse(rawMessage) as RealtimeNotificationMessage;
      await onMessage(message);
    } catch (error) {
      console.error('[NotificationPubSub] Invalid message:', rawMessage, error);
    }
  });

  return async () => {
    // 더 이상 알림을 받지 않도록 해당 채널의 구독을 해제합니다.
    // 구독 해제가 완료되면 구독자 클라이언트 연결을 종료할 수 있습니다.
    await subscriber.unsubscribe(channel);
    await subscriber.quit();
  };
}

 

구독 코드는 다음 부분이 중요합니다.

const subscriber = redis.duplicate();
await subscriber.connect();

 

Pub/Sub에서는 구독용 Redis 연결을 일반 Redis 명령용 연결과 분리하는 것이 좋습니다.
일반 Redis client는 다음 작업을 처리합니다.

GET
SET
HGETALL
INCR
PUBLISH

하지만 Subscriber client는 다음처럼 계속 채널을 구독합니다.

SUBSCRIBE channel:notification

 

그래서 보통 다음처럼 나눕니다.

일반 Redis client    → GET, SET, PUBLISH
Subscriber client   → SUBSCRIBE

 

✔️ 실행 예시

// 예시 실행 코드

const service = new NotificationPubSubService();

const unsubscribe = await service.subscribeNotification((message) => {
  console.log('[알림 수신]', message);
});

await service.publishPostLikedNotification({
  receiverUserId: 1,
  postId: 10,
  likedByUserName: 'Kim',
});

// 실습 종료 시
await unsubscribe();

 

2. 캐시 무효화 메시지 발행하기

Pub/Sub은 실시간 알림뿐 아니라 서버 간 캐시 무효화 전파에도 자주 사용됩니다.

 

예를 들어 서버가 여러 대 있다고 가정해 보겠습니다.

API Server A
API Server B
API Server C

각 서버는 사용자 정보를 Redis 또는 메모리 캐시에 가지고 있을 수 있습니다.

이때 Server A에서 사용자 정보가 변경되면, 다른 서버들도 기존 캐시를 지워야 합니다.

Server A에서 사용자 정보 수정
  ↓
DB update
  ↓
channel:cache-invalidation publish
  ↓
Server B, Server C subscriber가 메시지 수신
  ↓
각 서버에서 해당 캐시 삭제

 

✔️ 캐시 무효화 메시지 발행

/**
 * 캐시 무효화 메시지를 공용 채널에 발행합니다.
 *
 * 1. 무효화 메시지를 JSON 문자열로 변환합니다.
 * 2. 캐시 무효화 채널을 구독 중인 모든 서버에 발행합니다.
 * 3. 메시지를 전달받은 구독자 수를 반환합니다.
 *
 * @returns 메시지를 받은 subscriber 수
 */
async publishCacheInvalidation(message: CacheInvalidationMessage): Promise<number> {
  const channel = RedisKey.channel.cacheInvalidation();

  // 캐시 무효화 정보를 현재 연결된 구독 서버에 전달합니다.
  // 메시지를 발행하고 이를 전달받은 구독자 수를 반환합니다.
  return redis.publish(channel, JSON.stringify(message));
}

/**
 * 사용자 캐시 무효화 메시지를 구성해 발행합니다.
 *
 * 1. 사용자 ID로 무효화할 캐시 키를 생성합니다.
 * 2. 사용자 변경 사유와 현재 시각을 메시지에 기록합니다.
 * 3. 공용 캐시 무효화 발행 메서드에 전달합니다.
 */
async publishUserCacheInvalidation(userId: number): Promise<number> {
  const key = RedisKey.cache.user(userId);

  return this.publishCacheInvalidation({
    type: 'USER_CACHE_INVALIDATED',
    key,
    reason: `User ${userId} updated`,
    createdAt: new Date().toISOString(),
  });
}

 

캐시 무효화 메시지는 다음 구조를 가집니다.

export type CacheInvalidationMessage = {
  type: 'USER_CACHE_INVALIDATED' | 'POST_CACHE_INVALIDATED' | 'CUSTOM_KEY_INVALIDATED';
  key: string;
  reason: string;
  createdAt: string;
};

여기서 가장 중요한 필드는 key입니다.

예를 들어 사용자 캐시를 삭제하려면 다음 key가 메시지에 포함됩니다.

cache:user:1

 

✔️ 캐시 무효화 메시지 구독

/**
 * 캐시 무효화 메시지 구독을 시작합니다.
 *
 * 1. 일반 명령용 연결과 분리된 구독 전용 클라이언트를 생성합니다.
 * 2. 수신한 JSON 문자열을 캐시 무효화 메시지로 변환합니다.
 * 3. 메시지에 포함된 키를 삭제하고 선택적 콜백을 실행합니다.
 * 4. 구독 해제와 연결 종료를 수행하는 함수를 반환합니다.
 *
 * 참고:
 * 각 API 서버는 애플리케이션 시작 시 구독을 시작해 캐시 상태를 동기화할 수 있습니다.
 */
async subscribeCacheInvalidation(
  onInvalidated?: (message: CacheInvalidationMessage) => void | Promise<void>,
): Promise<() => Promise<void>> {
  const channel = RedisKey.channel.cacheInvalidation();

  const subscriber = redis.duplicate();
  await subscriber.connect();

  // 캐시 무효화 채널에서 새 이벤트를 실시간으로 수신합니다.
  // 구독이 유지되는 동안 이벤트를 받을 때마다 등록한 콜백을 실행합니다.
  await subscriber.subscribe(channel, async (rawMessage) => {
    try {
      const message = JSON.parse(rawMessage) as CacheInvalidationMessage;

      // 현재 서버에 저장된 무효화 대상 캐시를 삭제합니다.
      // 키가 없어도 오류 없이 처리하며 삭제한 키의 수를 반환합니다.
      await redis.del(message.key);

      if (onInvalidated) {
        await onInvalidated(message);
      }
    } catch (error) {
      console.error('[CacheInvalidationPubSub] Invalid message:', rawMessage, error);
    }
  });

  return async () => {
    // 더 이상 캐시 무효화 이벤트를 받지 않도록 채널 구독을 해제합니다.
    // 구독 해제가 완료되면 구독자 클라이언트 연결을 종료할 수 있습니다.
    await subscriber.unsubscribe(channel);
    await subscriber.quit();
  };
}

 

✔️ 캐시 무효화에 Pub/Sub을 쓰는 이유

캐시 무효화는 단일 서버에서는 단순합니다.

DB update
  ↓
Redis DEL

하지만 서버가 여러 대이면 문제가 생깁니다.

Server A는 캐시를 지웠지만
Server B, C는 아직 오래된 캐시를 들고 있을 수 있음

 

이때 Pub/Sub을 사용하면 캐시 삭제 이벤트를 여러 서버에 동시에 전파할 수 있습니다.

Server A
  ↓ publish
channel:cache-invalidation
  ↓ subscribe
Server B
Server C

 

단, Pub/Sub 메시지는 저장되지 않습니다.

따라서 캐시 무효화 메시지를 놓치면 일부 서버에 오래된 로컬 캐시가 남을 수 있습니다.
이 문제를 줄이려면 로컬 캐시에도 짧은 TTL을 두거나, 중요한 데이터는 Redis/DB를 기준으로 다시 검증하는 전략이 필요합니다.

반응형

 

3. 채팅 메시지 브로드캐스트하기

Pub/Sub은 채팅 메시지 브로드캐스트에도 사용할 수 있습니다.

채팅방 사용자가 여러 WebSocket 서버에 나뉘어 접속해 있다고 가정해 보겠습니다.

User A → WebSocket Server 1
User B → WebSocket Server 2
User C → WebSocket Server 3

 

User A가 메시지를 보내면 Server 1만 알고 있습니다.
이 메시지를 Server 2, Server 3에도 전달해야 다른 사용자들이 같은 채팅방 메시지를 받을 수 있습니다.

User A 메시지 전송
  ↓
WebSocket Server 1
  ↓
Redis PUBLISH channel:chat:{roomId}
  ↓
WebSocket Server 2, 3 수신
  ↓
User B, User C에게 전달

 

✔️ 사용하는 Redis Channel

RedisKey.channel.chat(roomId)

# 실제 채널 이름은 다음과 같습니다.
channel:chat:room-1
channel:chat:room-2

 

✔️ 채팅방 메세지 발행

// src/services/chat-pubsub.service.ts

/**
 * 채팅 메시지를 해당 채팅방 채널에 발행합니다.
 *
 * 1. 채팅방 ID로 발행할 채널을 결정합니다.
 * 2. 채팅 메시지를 JSON 문자열로 변환해 발행합니다.
 * 3. 메시지를 전달받은 구독자 수를 반환합니다.
 *
 * @returns 메시지를 받은 subscriber 수
 */
async publishChatMessage(message: ChatMessage): Promise<number> {
  const channel = RedisKey.channel.chat(message.roomId);

  // 채팅 메시지를 해당 채팅방에 접속한 구독자에게 전달합니다.
  // 메시지를 발행하고 이를 전달받은 구독자 수를 반환합니다.
  return redis.publish(channel, JSON.stringify(message));
}

/**
 * 입력값으로 채팅 메시지를 구성해 발행합니다.
 *
 * 1. 채팅방과 발신자 정보를 채팅 메시지로 구성합니다.
 * 2. 현재 시각을 메시지 생성 시각으로 기록합니다.
 * 3. 공용 채팅 메시지 발행 메서드에 전달합니다.
 *
 * 참고:
 * 채팅 이력이 필요하면 이 메서드 호출 전후에 DB 저장을 별도로 수행합니다.
 */
async sendMessage(input: {
  roomId: string;
  senderUserId: number;
  senderName: string;
  message: string;
}): Promise<number> {
  return this.publishChatMessage({
    roomId: input.roomId,
    senderUserId: input.senderUserId,
    senderName: input.senderName,
    message: input.message,
    createdAt: new Date().toISOString(),
  });
}

 

채팅방 채널은 roomId 기준으로 분리합니다.

const channel = RedisKey.channel.chat(message.roomId);

예를 들어 roomId가 room-1이면 다음 채널을 사용합니다.

channel:chat:room-1

채팅 메시지 발행 코드는 다음입니다.

return redis.publish(channel, JSON.stringify(message));

 

✔️ 채팅방 메세지 구독

/**
 * 특정 채팅방의 메시지 구독을 시작합니다.
 *
 * 1. 일반 명령용 연결과 분리된 구독 전용 클라이언트를 생성합니다.
 * 2. 채팅방 채널에서 받은 JSON 문자열을 채팅 메시지로 변환합니다.
 * 3. 변환한 메시지를 콜백에 전달하고 구독 종료 함수를 반환합니다.
 *
 * 참고:
 * WebSocket 서버는 콜백에서 해당 채팅방에 접속한 클라이언트에게 메시지를 전달할 수 있습니다.
 */
async subscribeChatRoom(
  roomId: string,
  onMessage: (message: ChatMessage) => void | Promise<void>,
): Promise<() => Promise<void>> {
  const channel = RedisKey.channel.chat(roomId);

  const subscriber = redis.duplicate();
  await subscriber.connect();

  // 지정한 채팅방 채널에서 새 메시지를 실시간으로 수신합니다.
  // 구독이 유지되는 동안 메시지를 받을 때마다 등록한 콜백을 실행합니다.
  await subscriber.subscribe(channel, async (rawMessage) => {
    try {
      const message = JSON.parse(rawMessage) as ChatMessage;
      await onMessage(message);
    } catch (error) {
      console.error('[ChatPubSub] Invalid message:', rawMessage, error);
    }
  });

  return async () => {
    // 더 이상 채팅 메시지를 받지 않도록 해당 채널의 구독을 해제합니다.
    // 구독 해제가 완료되면 구독자 클라이언트 연결을 종료할 수 있습니다.
    await subscriber.unsubscribe(channel);
    await subscriber.quit();
  };
}

구독자는 같은 채널을 구독합니다.

await subscriber.subscribe(channel, async (rawMessage) => {
  const message = JSON.parse(rawMessage) as ChatMessage;
  await onMessage(message);
});

이 구조를 사용하면 여러 WebSocket 서버가 같은 Redis 채널을 구독할 수 있습니다.

WebSocket Server 1 subscribe channel:chat:room-1
WebSocket Server 2 subscribe channel:chat:room-1
WebSocket Server 3 subscribe channel:chat:room-1

어떤 서버에서 메시지를 publish하더라도 모든 서버가 메시지를 받을 수 있습니다.

 

✔️ 채팅에서 Pub/Sub 사용 시 주의할 점

ub/Sub은 메시지를 저장하지 않습니다.

따라서 다음 기능은 Pub/Sub만으로 처리하면 안 됩니다.

- 이전 채팅 내역 조회
- 안 읽은 메시지 계산
- 메시지 재전송
- 장애 발생 후 메시지 복구

 

따라서 보통 다음처럼 역할을 나눕니다.

DB 또는 Stream
  → 채팅 메시지 저장

Redis Pub/Sub
  → 현재 접속 중인 사용자에게 실시간 전달

즉 Pub/Sub은 채팅 저장소가 아니라 실시간 전달 경로입니다.

 

4. 관리자 공지 발행하기

관리자 공지는 Pub/Sub의 특징을 보여주기 좋은 예제입니다.
예를 들어 관리자가 전체 사용자에게 다음 메시지를 보내야 한다고 가정합니다.

"오늘 23시에 서버 점검이 있습니다."

이 메시지는 모든 서버 또는 모든 클라이언트에게 즉시 전달되어야 합니다.

Admin API
  ↓
Redis PUBLISH channel:admin-notice
  ↓
모든 Subscriber 수신
  ↓
접속 중인 사용자에게 공지 표시

 

✔️ 사용하는 Channel 예시

RedisKey.channel.adminNotice()

channel:admin-notice

 

✔️ 공지 발행

// src/services/admin-notice-pubsub.service.ts

/**
 * 관리자 공지를 공용 공지 채널에 발행합니다.
 *
 * 1. 관리자 공지를 JSON 문자열로 변환합니다.
 * 2. 공지 채널을 구독 중인 모든 구독자에게 발행합니다.
 * 3. 메시지를 전달받은 구독자 수를 반환합니다.
 *
 * @returns 메시지를 받은 subscriber 수
 */
async publishAdminNotice(message: AdminNoticeMessage): Promise<number> {
  const channel = RedisKey.channel.adminNotice();

  // 관리자 공지를 현재 연결된 모든 공지 구독자에게 전달합니다.
  // 메시지를 발행하고 이를 전달받은 구독자 수를 반환합니다.
  return redis.publish(channel, JSON.stringify(message));
}

/**
 * 입력값으로 일반 공지를 구성해 발행합니다.
 *
 * 1. 공지 식별자와 표시 내용을 메시지로 구성합니다.
 * 2. 중요도를 INFO로, 생성 시각을 현재 시각으로 설정합니다.
 * 3. 공용 관리자 공지 발행 메서드에 전달합니다.
 */
async publishInfoNotice(input: {
  noticeId: string;
  title: string;
  content: string;
}): Promise<number> {
  return this.publishAdminNotice({
    noticeId: input.noticeId,
    title: input.title,
    content: input.content,
    level: 'INFO',
    createdAt: new Date().toISOString(),
  });
}

 

관리자 공지 메시지는 다음 구조를 가집니다.

export type AdminNoticeMessage = {
  noticeId: string;
  title: string;
  content: string;
  level: 'INFO' | 'WARNING' | 'URGENT';
  createdAt: string;
};

 

공지 레벨은 다음처럼 나누었습니다.

INFO     → 일반 공지
WARNING  → 주의 공지
URGENT   → 긴급 공지

 

공지 발행 코드는 단순합니다.

const channel = RedisKey.channel.adminNotice();

return redis.publish(channel, JSON.stringify(message));

channel:admin-notice를 구독 중인 서버는 모두 같은 메시지를 받습니다.

 

✔️ 관리자 공지 구독

/**
 * 관리자 공지 채널 구독을 시작합니다.
 *
 * 1. 일반 명령용 연결과 분리된 구독 전용 클라이언트를 생성합니다.
 * 2. 공지 채널에서 받은 JSON 문자열을 공지 메시지로 변환합니다.
 * 3. 변환한 공지를 콜백에 전달하고 구독 종료 함수를 반환합니다.
 */
async subscribeAdminNotice(
  onMessage: (message: AdminNoticeMessage) => void | Promise<void>,
): Promise<() => Promise<void>> {
  const channel = RedisKey.channel.adminNotice();

  const subscriber = redis.duplicate();
  await subscriber.connect();

  // 관리자 공지 채널에서 새 공지를 실시간으로 수신합니다.
  // 구독이 유지되는 동안 공지를 받을 때마다 등록한 콜백을 실행합니다.
  await subscriber.subscribe(channel, async (rawMessage) => {
    try {
      const message = JSON.parse(rawMessage) as AdminNoticeMessage;
      await onMessage(message);
    } catch (error) {
      console.error('[AdminNoticePubSub] Invalid message:', rawMessage, error);
    }
  });

  return async () => {
    // 더 이상 관리자 공지를 받지 않도록 해당 채널의 구독을 해제합니다.
    // 구독 해제가 완료되면 구독자 클라이언트 연결을 종료할 수 있습니다.
    await subscriber.unsubscribe(channel);
    await subscriber.quit();
  };
}

 

✔️ 관리자 공지에서 Pub/Sub을 쓰는 이유

관리자 공지는 보통 다음 요구사항을 가집니다.

- 접속 중인 사용자에게 즉시 전달
- 여러 서버에 동시에 전파
- 메시지 구조가 단순함
- 유실되어도 이후 공지 목록 API로 보완 가능

이런 경우 Pub/Sub이 적합합니다.

다만 공지 자체를 보관해야 한다면 Pub/Sub만 사용하면 안 됩니다.

DB
  → 공지 이력 저장

Redis Pub/Sub
  → 현재 접속 중인 사용자에게 실시간 전달

 

---

※ 게시된 글 및 이미지 중 일부는 AI 도구의 도움을 받아 생성되거나 다듬어졌습니다.

반응형