[RabbitMQ] rabbitMQ 메시지가 왔다가 안왔다가 왔다가 안왔다가... ♾️

프로젝트 중 제일 어려웠던 rabbitMQ에 대해 작성해보려고 한다. 

1. RabbitMQ 왜 썼지? 

프로젝트를 진행하면서 RabbitMQ를 사용하게 되었다. 우리는 redis가 1분마다 업데이트될 때마다 조건탐지와 기업별 알림의 조건들을 실제 차트 데이터와 비교하였고 조건이 충족하면 알림을 보내주는 서비스였기 때문에 두 서비스가 비동기적으로 동작해야한다고 생각했다. kafka는 대규모 처리에 적합하기 때문에 우리 서비스에는 오버스펙이라 느껴져 rabbitMQ를 선택하게 되었다. 

 

2. RabbitMQ가 뭔데? 

수업시간에 배웠던 내용으로 간략히 설명하자면 메시지 브로커의 한 종류이다. 여러 서비스가 직접 서로 소통하지 안하고 중간에 메시지를 전달해주는 중개자로 rabbitMQ는 우체국이라고 생각하면 된다. 즉, 1:1로 데이터를 전달하고 싶을 때 사용하면 된다. RabbitMQ를 설정할 때 보면 Exchange, Queue, Routing Key를 지정해줘야하는데 이들 각각은

Exchange - 메시지를 보내는 곳, 우체국

Queue - 메시지를 보관하는 우체통

Routing Key - 주소 라고 생각하면 된다.

바인딩을 통해 이 메시지는 이 주소를 통해 이 우체통으로 보내주세요. 라고 지정해주면 된다. 

 

3. 트러블 슈팅

3.1 publish 로그는 찍히지만 메시지 수신을 받지 못하고 있을 때

문제 발생

말 그대로 메시지 발신한 것은 맞는 것 같은데 수신이 오지 않는 상황이었다. rabbitMQ management로 확인해보았을 때도 분명 메시지가 전달되어 튀는 것을 확인할 수 있었다. 

 

원인 분석

내가 rabbitMQ를 구현하기 전 인프라를 담당한 언니가 rabbitMQ를 어떻게 사용하는지 간단한 테스트 코드를 짜줬었다. 아래는 해당 관련 코드의 일부이다. 

@Configuration
@EnableRabbit
public class RabbitMQConfig {

    // ======================================================
    // 📦 ALERT 모듈 (테스트용 Exchange / Queue / RoutingKey)
    // ======================================================
    public static final String ALERT_EXCHANGE = "alert.exchange";
    public static final String ALERT_QUEUE = "alert.queue";
    public static final String ALERT_ROUTING_KEY = "alert.key";

    @Bean
    public DirectExchange alertExchange() {
        return new DirectExchange(ALERT_EXCHANGE);
    }

    @Bean
    public Queue alertQueue() {
        return new Queue(ALERT_QUEUE, true);
    }

    @Bean
    public Binding alertBinding(Queue alertQueue, DirectExchange alertExchange) {
        return BindingBuilder.bind(alertQueue)
                .to(alertExchange)
                .with(ALERT_ROUTING_KEY);
    }
}

@Service
@RequiredArgsConstructor
public class AlertSenderService {

    private final RabbitTemplate rabbitTemplate;

    public void sendTestMessage(String message) {
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.ALERT_EXCHANGE,
                RabbitMQConfig.ALERT_ROUTING_KEY,
                message
        );
        System.out.println("📤 Sent message to RabbitMQ: " + message);
    }
}

@Service
public class AlertReceiverService {

    @RabbitListener(queues = "alert.queue")
    public void receive(String message) {
        System.out.println("📩 Received message from RabbitMQ: " + message);
    }
}

 

그리고 내가 짠 코드는 아래와 같다.

public class AlertEventPublisher {

    private final RabbitTemplate rabbitTemplate;

    public void publish(AlertConditionManager manager) {
        var alert = manager.getAlert();
        var event = new AlertEvent(
                alert.getUserId(),
                alert.getStockCode(),
                manager.getAlertCondition().getIndicator(),
                LocalDateTime.now()
        );

        log.info("""
        📤 [Publish 시도]
        Exchange  = {}
        RoutingKey= {}
        Connection= {}
        Event     = {}
        """,
                RabbitMQConfig.ALERT_EXCHANGE,
                RabbitMQConfig.ALERT_ROUTING_KEY,
                rabbitTemplate.getConnectionFactory(),
                event);

        rabbitTemplate.convertAndSend(
                RabbitMQConfig.ALERT_EXCHANGE,
                RabbitMQConfig.ALERT_ROUTING_KEY,
                event
        );
    }
}

@RabbitListener(queues = RabbitMQConfig.ALERT_QUEUE)
public class AlertNotifier {
    private final PushService pushService;

    @RabbitHandler
    public void handleAlert(AlertEvent event) {
        log.info("📥 MQ 이벤트 수신: {}", event);
        pushService.send(event);
    }
}

 

위 코드들 중 주요깊게 봐야할 부분은 바로 이 부분이다. 

public static final String ALERT_EXCHANGE = "alert.exchange";
public static final String ALERT_QUEUE = "alert.queue";
public static final String ALERT_ROUTING_KEY = "alert.key";

지금 한 큐 안에 AlertEvent와 String, 두 타입 모두 들어가게 된다. RabbitMQ는 우체국 같은 아이라고 비유했다. 1:1 소통, 그런데 이렇게 다른 타입을 넣어버리면 내부적으로 충돌이 생길 수 밖에 없다. MessageConverter를 지정하여 json으로 직렬화/역직렬화 할 수 있게 할 수도 있지만, 굉장히 위험하다. 

 

해결

그래서 나는 test용 큐와 바인딩을 추가하여 관리하도록 해주었다. 

public static final String ALERT_EXCHANGE = "alert.exchange";
public static final String ALERT_QUEUE = "alert.queue";
public static final String ALERT_ROUTING_KEY = "alert.key";

public static final String ALERT_TEST_QUEUE = "alert.test.queue";
public static final String ALERT_TEST_ROUTING_KEY = "alert.test.key";

 

이렇게 해결을 한 뒤에 이제 잘 되나 싶었더니 또 다른 문제가 발생했다. 

 

 

3.2 메시지가 왔다가 안왔다가.... 두번째 메시지는 왔는데 첫번째 메시지가 소실됐다... 

문제1

27분에 실행한 로그와 

28분에 실행한 로그이다. 

고작 1분차이인데 메시지가 응답을 받았다 말았다 한다. 

 

문제2

RSI_OVER인 조건과

RSI_UNDER인 조건 분명 2개가 publish 되었다. 

근데 이벤트 수신은 RSI_UNDER만 되버린 결과이다. 

 

원인 분석

RabbitListner가 아직 컨테이너에 완전히 등록되기 전에 메시지가 발행된 것 같았다. RabbitMQ는 리스너 등록 전에 발행된 메시지도 큐에 쌓이지만, 만약 그시점에 큐가 존재하지 않았고 durable인 큐가 아니라면 메시지는 그대로 사라진다고 한다. 

@RabbitListener(queues = "alert.queue", autoStartup = "true")
public class AlertNotifier {
    private final PushService pushService;

    @RabbitHandler
    public void handleAlert(AlertEvent event) {
        log.info("📥 MQ 이벤트 수신: {}", event);
        pushService.send(event);
    }
}

그래서 autoStartup을 true로 해주고 실행했다. 그랬더니 잘 들어오는 것이었다. 그런 줄만 알았다......

근데 아니었다...... autoStartup은 default 값이 true이고 마침 딱 잘 돌아간 것 뿐이었다 . . . 

 

그래서 강사님께 상황 설명을 드렸더니 rabbitMQ는 원래 메시지 소실이 잦은 아이라고 한다. 그렇기 때문에 최소화할 수 있는 방법들이 여러가지가 있고 그 중 하나가 클러스터링이라고 하셨다. 

 

 

알림은 무조건 보내져야하기 때문에 클러스터링과 미러링을 도입하기로 하였다. 

해결 - 클러스터링 및 미러링

클러스터링은 가용할 수 있는 노드의 개수를 늘리는 것이고 메시지를 유실하지 않기 위해서는 미러 큐/쿼럼 큐를 적용해야한다. 

※ 미러 큐는 3.8 이후 deprecated를 지정했으나 EC2 환경이 넉넉하지 않았고 알림 서비스 특성상 메시지 처리량이 크지 않아서 Mirror Queue로도 충분하다고 판단해 미러 큐를 적용했다.

 

rabbitMQ 또한, ec2내부 도커 위에 이미 설치되어 있었기 때문에 일단 해당 ec2로 접속하여 주었다. 아래 순서대로 코드를 실행하여 주었고 

mkdir -p ~/rabbitmq-cluster && cd ~/rabbitmq-cluster

nano docker-compose.yml

# 클러스터 실행
docker compose up -d

# 컨테이너 확인
docker ps

# 클러스터 상태확인
docker exec -it rabbit1 rabbitmqctl cluster_status

# 미러큐 적용
docker exec -it rabbit1 rabbitmqctl set_policy ha-all \
  ".*" \
  '{"ha-mode":"all","ha-sync-mode":"automatic"}' \
  --apply-to queues

docker-compose.yml 파일은 아래와 같다. 

services:
  rabbit1:
    image: rabbitmq:3.12-management
    container_name: rabbit1
    hostname: rabbit1
    environment:
      RABBITMQ_ERLANG_COOKIE: "<SECRET_COOKIE>"
      RABBITMQ_DEFAULT_USER: "<ADMIN_USER>"
      RABBITMQ_DEFAULT_PASS: "<ADMIN_PASS>"
    ports:
      - "15672:15672"
      - "5672:5672"
    networks:
      - rabbitnet

  rabbit2:
    image: rabbitmq:3.12-management
    container_name: rabbit2
    hostname: rabbit2
    environment:
      RABBITMQ_ERLANG_COOKIE: "<SECRET_COOKIE>"
      RABBITMQ_DEFAULT_USER: "<ADMIN_USER>"
      RABBITMQ_DEFAULT_PASS: "<ADMIN_PASS>"
    ports:
      - "15673:15672"
      - "5673:5672"
    networks:
      - rabbitnet
    depends_on:
      - rabbit1
    command: >
      bash -c "
        rabbitmq-server -detached;
        sleep 15;
        rabbitmqctl stop_app;
        rabbitmqctl reset;
        rabbitmqctl join_cluster rabbit@rabbit1;
        rabbitmqctl start_app;
        tail -f /dev/null
      "

  rabbit3:
    image: rabbitmq:3.12-management
    container_name: rabbit3
    hostname: rabbit3
    environment:
      RABBITMQ_ERLANG_COOKIE: "<SECRET_COOKIE>"
      RABBITMQ_DEFAULT_USER: "<ADMIN_USER>"
      RABBITMQ_DEFAULT_PASS: "<ADMIN_PASS>"
    ports:
      - "15674:15672"
      - "5674:5672"
    networks:
      - rabbitnet
    depends_on:
      - rabbit1
    command: >
      bash -c "
        rabbitmq-server -detached;
        sleep 15;
        rabbitmqctl stop_app;
        rabbitmqctl reset;
        rabbitmqctl join_cluster rabbit@rabbit1;
        rabbitmqctl start_app;
        tail -f /dev/null
      "

networks:
  rabbitnet:
    driver: bridge

 

그리고 큐의 durable 옵션을 true로 둬서 서버가 재시작 되더라도 큐가 유지될 수 있도록 하고 메시지 내용까지 복구하고자 한다면 메시지 자체에 persistent를 설정해야하는데 Spring AMQP에서 기본적으로 유지, 복구될 수 있도록 설정해주기 때문에 명시하고 싶은 것이 아니라면 따로 설정해줘야할 필요는 없다.

 

혹시 명시적으로 설정하고 싶다면 아래와 같은 방식으로 하면 된다 ! 

@Bean
public Queue alertCompanyQueue() {
    return new Queue(ALERT_COMPANY_QUEUE, true);
}

rabbitTemplate.convertAndSend(
    RabbitMQConfig.ALERT_EXCHANGE,
    RabbitMQConfig.ALERT_COMPANY_ROUTING_KEY,
    event,
    message -> {
        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        return message;
    }
);

 

 

 

이번에 처음으로 메시지 브로커라는 것을 사용해보는 것이라 이리저리 헤매고 이렇게 블로그를 작성할 때가 되니 좀 더 보완해야할 부분이 많다고 느껴진다. 다음에 또 사용하게 된다면 더욱 잘할 수 있겠다는 자신감이 든다 !!!!