Redis Pub/Sub 구현 (구독자모드의 제약 사항)

Redis Pub/Sub을 사용하기로 하고 구현에 나섰다. 우선, 프로젝트 내에서 이미 Redis가 구현이 되어서 동작하고있는 상황이었기에, 간단하게 Pub/Sub 기능만 추가해주면 되는 상황이었다. 그러면 처음으로 Redis Pub/Sub을 공부하며 구현하고, 마주한 문제에 대해 정리해보려 한다.

 

 

Redis Pub/Sub

: Redis에서 지원하는 하나의 메세지를 여러 수신자에게 동시에 전송하는 실시간 통신으로 활용되어 서로 다른 서비스 간 메세지를 쉽게 주고받을 수 있다. 메세지를 발행하는 Publisher와, 해당 메세지를 구독하여 수신하는 Subscriber로 구성된다. Publisher는 특정 채널(Channel)에 메세지를 발행하고, Subscriber는 해당 채널을 구독하여 실시간으로 메세지를 수신한다.

 

 

Publish 메서드

async publish(channel: string, message: any): Promise<number> {
  return this.client.publish(channel, JSON.stringify(message));
}

 

• 역할 : 지정된 Redis 채널에 메세지를 발행한다. 발행된 메세지는 해당 채널을 구독 중인 모든 클라이언트에게 전송된다.
• 반환 값 : 메세지를 수신한 구독자 수를 반환한다.

Subscribe 메서드

async subscribe(channel: string, callback: (message: string) => void): Promise<void> {
  this.client.subscribe(channel);
  this.client.on('message', (subscribedChannel, message) => {
    if (subscribedChannel === channel) {
      callback(message);
    }
  });
}

• 역할 : 지정된 채널을 구독하고, 해당 채널로 발행된 메세지를 실시간으로 수신한다.
• Callback : 수신된 메세지를 처리하는 함수

 

 

이렇게 단순히 기존 redis 파일에 publish, subscribe 함수를 추가해준 후 마주하게 된 문제

Redis Pub/Sub 구독자모드 문제

 

현재 Redis 연결이 '구독자 모드(subscriber mode)'에 있다고한다. 그래서 구독자 명령이 아닌 publish 같은 다른 일반적인 명령을 사용할 수 없다는 뜻..

그래서 우선 구독자 모드의 제약 사항에 대해 살펴보았다.

 

구독자 모드의 제약 사항

: Redis에서는 구독자 모드에 들어간 클라이언트가 구독자 명령어(SUBSCRIBE, UNSUBSCRIBE 등)만 처리할 수 있다. 따라서 구독모드에서는 일반적인 Redis 명령어(ex. PUBLISH, SET, GET 등)을 사용할 수 없다.

 -> 하나의 클라이언트로 발행과 구독을 동시에 처리하려고 하면 충돌 발생

    ex) 구독 모드에 있는 클라이언트가 PUBLISH 명령을 실행하려고하는 경우

 -> 발행자와 구독자 클라이언트를 분리해야 한다는 결론에 도달

 

해결 : 발행자와 구독자 클라이언트 분리

1. 구독자 전용 클라이언트 : 채널을 구독하고, 해당 채널로 발행된 메세지를 실시간으로 수신
2. 발행자 전용 클라이언트 : 메세지를 채널에 발행

 

• 실제 구현 코드

@Injectable()
export class RedisService implements OnModuleInit, OnModuleDestroy {
  private client: Redis; // 발행자용 Redis 클라이언트
  private pubSubClient: Redis; // 구독자용 Redis 클라이언트
  private subscribeChannels: Set<string> = new Set(); // 구독 중인 채널 목록

  constructor(private configService: ConfigService) {}

  async onModuleInit() {
    // Redis 클라이언트 설정
    const redisHost = this.configService.get<string>('REDIS_HOST');
    const redisPort = Number(this.configService.get<string>('REDIS_PORT')) || 6379;

    // 발행자 클라이언트
    this.client = new Redis({
      host: redisHost,
      port: redisPort,
    });

    // 구독자 클라이언트
    this.pubSubClient = new Redis({
      host: redisHost,
      port: redisPort,
    });

    // Redis 연결 성공 및 오류 처리
    this.client.on('connect', () => console.log('Connected to Redis (Publisher)'));
    this.pubSubClient.on('connect', () => console.log('Connected to Redis (Subscriber)'));
    this.client.on('error', (err) => console.error('Redis error', err));
    this.pubSubClient.on('error', (err) => console.error('Redis error (Subscriber)', err));
  }

  // 메시지 발행 메서드
  async publish(channel: string, message: any): Promise<number> {
    return this.client.publish(channel, JSON.stringify(message));
  }

  // 메시지 구독 메서드
  async subscribe(channel: string, callback: (message: string) => void): Promise<void> {
    if (!this.subscribeChannels.has(channel)) {
      this.pubSubClient.subscribe(channel);
      this.pubSubClient.on('message', (subscribedChannel, message) => {
        if (subscribedChannel === channel) {
          callback(message);
        }
      });
      this.subscribeChannels.add(channel);
    }
  }

  async onModuleDestroy() {
    await this.client.quit();
    await this.pubSubClient.quit();
  }
}

Redis Pub/Sub에 대한 기본적인 개념뿐 아니라 구독자 모드의 제약사항으로 인해 발행자와 구독자를 별도의 클라이언트로 분리하는 것이 필수적이라는 것을 알 수 있었다. 발행과 구독 개념이 재미있다고 느껴졌고, 안정적인 실시간 메세지 처리가 가능해진 Redis Pub/Sub을  다양한 곳에서 어떻게 활용하면 좋을지 기대된다.