tooolbox/blog/Redis Pub/Sub 메시지 유실 문제 해결: 신뢰성 있는 메시징 구현
troubleshooting2025-02-25·9 min·172/348

Redis Pub/Sub 메시지 유실 문제 해결: 신뢰성 있는 메시징 구현

Redis Pub/Sub의 메시지 유실 문제를 분석하고 신뢰성 있는 메시징 시스템을 구현하는 방법을 설명합니다.

Redis Pub/Sub 메시지 유실 문제 해결

Introduction

Redis Pub/Sub는 간단하고 빠른 메시징 시스템을 구현할 수 있는 강력한 도구이지만, fire-and-forget 특성으로 인해 메시지 유실 문제가 빈번하게 발생합니다. 프로덕션 환경에서 메시지 유실은 심각한 데이터 불일치와 비즈니스 로직 오류를 초래할 수 있습니다. 이 글에서는 메시지 유실의 원인을 분석하고 해결 방법을 다루겠습니다.

Environment

# Redis Pub/Sub 테스트 환경
redis-cli

# 모니터링 도구
redis-cli MONITOR
// Node.js Redis 클라이언트 설정
const Redis = require('ioredis');

const publisher = new Redis({
  host: 'localhost',
  port: 6379,
  retryStrategy: (times) => {
    return Math.min(times * 50, 2000);
  }
});

const subscriber = new Redis({
  host: 'localhost',
  port: 6379
});

Problem

Pub/Sub 메시지 유실 상황:

// 발행자 (publisher.js)
async function publishMessages() {
  for (let i = 0; i < 10000; i++) {
    await publisher.publish('orders', JSON.stringify({
      orderId: i,
      amount: Math.random() * 1000,
      timestamp: Date.now()
    }));
  }
  console.log('Published 10000 messages');
}

// 구독자 (subscriber.js)
let receivedCount = 0;

subscriber.subscribe('orders', (err) => {
  if (err) {
    console.error('Subscribe error:', err);
  }
});

subscriber.on('message', (channel, message) => {
  receivedCount++;
  // 처리 지연 시뮬레이션
  setTimeout(() => {
    console.log(`Received: ${receivedCount}`);
  }, 10);
});

// 실행 결과
// Published 10000 messages
// Received: 8547  // 1453개 메시지 유실!
# 구독자 연결 끊김 시 메시지 유실 확인
# 구독자 재시작 후
$ node subscriber.js
# 메시지 수신 안됨 (재시작 전 발행된 메시지 유실)

Analysis

메시지 유실의 원인을 분석했습니다:

// Redis Pub/Sub 동작 원리 확인
// 1. 메시지는 연결된 모든 구독자에게만 전달됨
// 2. 구독자가 연결되지 않은 상태에서 발행된 메시지는 버려짐
// 3. 메모리 기반이므로 오프라인 큐 기능 없음

// 구독자 상태 확인
const clientInfo = await publisher.client('INFO', 'clients');
console.log(clientInfo);

// 연결된 구독자 수 확인
const subscribedClients = await publisher.pubsub('NUMSUB', 'orders');
console.log('Subscribers:', subscribedClients);
# Redis 메모리 사용량 확인
redis-cli INFO memory
# used_memory:1048576
# used_memory_human:1.00M

# Pub/Sub 채널 정보
redis-cli PUBSUB CHANNELS
# 1) "orders"
# 2) "notifications"

Solution

1단계: Redis Streams를 활용한 신뢰성 있는 메시징

// streams-publisher.js
const Redis = require('ioredis');

const redis = new Redis();

async function publishToStream(message) {
  const id = await redis.xadd(
    'orders-stream',
    '*',  // 자동 ID 생성
    'orderId', message.orderId,
    'amount', message.amount.toString(),
    'timestamp', message.timestamp.toString()
  );
  
  console.log(`Published message with ID: ${id}`);
  return id;
}

// 메시지 발행
async function publishMessages() {
  for (let i = 0; i < 10000; i++) {
    await publishToStream({
      orderId: i,
      amount: Math.random() * 1000,
      timestamp: Date.now()
    });
  }
}
// streams-consumer.js
const Redis = require('ioredis');

const redis = new Redis();
const CONSUMER_GROUP = 'order-processors';
const CONSUMER_NAME = 'processor-1';

async function createConsumerGroup() {
  try {
    await redis.xgroup(
      'CREATE', 'orders-stream', CONSUMER_GROUP, '0', 'MKSTREAM'
    );
    console.log('Consumer group created');
  } catch (err) {
    if (err.message.includes('BUSYGROUP')) {
      console.log('Consumer group already exists');
    } else {
      throw err;
    }
  }
}

async function processMessages() {
  await createConsumerGroup();
  
  while (true) {
    try {
      const messages = await redis.xreadgroup(
        'GROUP', CONSUMER_GROUP, CONSUMER_NAME,
        'COUNT', 10,
        'BLOCK', 5000,
        'STREAMS', 'orders-stream', '>'
      );
      
      if (messages) {
        for (const [stream, entries] of messages) {
          for (const [id, fields] of entries) {
            await processMessage(id, fields);
            
            // 메시지 처리 완료 확인
            await redis.xack('orders-stream', CONSUMER_GROUP, id);
          }
        }
      }
    } catch (err) {
      console.error('Consumer error:', err);
      await new Promise(resolve => setTimeout(resolve, 1000));
    }
  }
}

async function processMessage(id, fields) {
  const message = {};
  for (let i = 0; i < fields.length; i += 2) {
    message[fields[i]] = fields[i + 1];
  }
  
  console.log(`Processing message ${id}:`, message);
  
  // 비즈니스 로직 처리
  // 데이터베이스 저장 등
}

2단계: Pending 메시지 복구

// 미처리 메시지 확인 및 복구
async function recoverPendingMessages() {
  const pending = await redis.xpending(
    'orders-stream', CONSUMER_GROUP, 'COUNT', 100
  );
  
  for (const [id, consumer, idleTime, deliveryCount] of pending) {
    if (idleTime > 60000) {  // 60초 이상 미처리
      console.log(`Recovering message ${id}, idle: ${idleTime}ms`);
      
      // 다른 컨슈머에게 메시지 전달
      const claimResult = await redis.xclaim(
        'orders-stream',
        CONSUMER_GROUP,
        'backup-processor',
        60000,  // 최소 유휴 시간
        id
      );
      
      if (claimResult.length > 0) {
        console.log(`Claimed message ${id}`);
      }
    }
  }
}

// 주기적으로 미처리 메시지 확인
setInterval(recoverPendingMessages, 30000);

3단계: 하이브리드 아키텍처

// 하이브리드 메시징 시스템
class HybridMessageSystem {
  constructor() {
    this.redis = new Redis();
    this.fallbackQueue = [];  // 로컬 큐 (Redis 장애 시 대체)
  }
  
  async publish(message) {
    try {
      // Redis Streams에 발행
      const id = await this.redis.xadd(
        'messages-stream',
        '*',
        'data', JSON.stringify(message),
        'timestamp', Date.now().toString()
      );
      
      return { success: true, id };
    } catch (err) {
      console.error('Redis publish failed, using fallback:', err);
      
      // Redis 실패 시 로컬 큐에 저장
      this.fallbackQueue.push({
        message,
        timestamp: Date.now(),
        retries: 0
      });
      
      return { success: false, fallback: true };
    }
  }
  
  async retryFailedMessages() {
    const pending = [...this.fallbackQueue];
    this.fallbackQueue = [];
    
    for (const item of pending) {
      if (item.retries < 3) {
        const result = await this.publish(item.message);
        if (!result.success) {
          item.retries++;
          this.fallbackQueue.push(item);
        }
      }
    }
  }
}

Lessons Learned

  1. Redis Streams 도입: Pub/Sub 대신 Redis Streams를 사용하면 메시지 유실을 방지할 수 있습니다
  2. Consumer Group 활용: 컨슈머 그룹을 사용하면 메시지 처리를 분산하고 미처리 메시지를 추적할 수 있습니다
  3. ACK 메커니즘: 메시지 처리 완료를 명시적으로 확인하는 ACK 메커니즘이 필수적입니다
  4. 하이브리드 접근: Redis 장애에 대비한 폴백 메커니즘을 구현해야 합니다
  5. 모니터링: 처리 지연과 미처리 메시지 수를 지속적으로 모니터링해야 합니다

이 블로그는 외부 스폰서십, 제휴 마케팅 또는 광고 수익을 받지 않습니다.