RabbitMQ로 구현하는 비동기 메시지 큐(입문)

메시지 큐 : 시스템 간 데이터를 비동기적으로 전달하기 위한 중간 저장소. 

- 생산자(Producer)가 메시지를 보내고, 소비자(Consumer)가 받아 처리한다.

- 비유: 우체국에서 편지를 보내는 것처럼, 발신자가 수신자를 기다리지 않고 우체통(큐)에 넣으면 배달원(소비자)가 가져가 처리하는 것.

 

동작원리

- 생산자 : 데이터를 메시지로 만들어 큐에 전송

- 큐 : 메시지를 순서대로 저장, 소비자가 준비될 때까지 보관

- 소비자 : 큐에서 메시지를 꺼내 처리

- 비동기성 : 생산자와 소비자가 서로를 기다리지 않음

- 메시지보장 : 설정에 따라 메시지가 손실되지 않도록 보장 가능  

 

장점

- 비동기 : 생산자와 소비자가 독립적으로 동작

- 부하분산 : 처리 속도 차이 완충

- 확장성 : 다중 소비자로 작업 분산

예로는 주문처리, 알림전송, 로그 수집 등으로 사용가능하다.

 

단점

- 복잡성 : 직접 호출보다 디버깅이 어렵다

- 지연 : 큐를 거치며 약간의 처리 지연 발생 가능

- 설정 필요 : 메시지 손실 방지, 순서 보장 등 추가 설정 요구 

 

--

 

RabbitMq : AMQP (Advanced Message Queuing Protocol)를 구현한 오픈소스 메시지 브로커이다.

 

특징

-  큐, 교환기, 바인딩으로 유연한 라우팅

큐 : 메시지가 저장되는 공간

교환기 : 메시지를 받아 규칙에 따라 큐로 분배

바인딩 : 교환기와 큐를 연결하는 규칙 (분류 기준)

 

- 높은 처리량과 안전성

- Spring AMQP로 비교적 쉬운 개발 가능 (@RabbitListener, RabbitTemplate)

 

 

 

예제

# gradle 추가
// RabbitMQ 연동
implementation 'org.springframework.boot:spring-boot-starter-amqp'


# yml 추가
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

 

# RabbitMq 설치
brew install rabbitmq

# RabbitMq 실행
rabbitmq-server

 

실행 후 브라우저로 접속 및 로그인을 진행한다.

브라우저 주소 : http://localhost:15672 

(디폴트) 로그인 / 비밀번호 : guest / guest 

 

 

rabbitMq 에 보내고 싶은 객체를 생성한다.

@Data
public class TransferRequest {
    private String fromAccount;
    private String toAccount;
    private BigDecimal amount;
}

 

RabbitMq 설정파일을 생성한다.

@Configuration
public class RabbitMQConfig {

    public static final String TRANSFER_QUEUE = "transfer_queue";

    @Bean // 큐 정의
    public Queue transferQueue() {
        // durable: true로 영속성 보장
        return new Queue(TRANSFER_QUEUE, true);
    }

    @Bean // JSON 메시지 컨버터 설정
    public Jackson2JsonMessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean // RabbitTemplate에 컨버터 적용
    public RabbitTemplate rabbitTemplate(org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(messageConverter());
        return template;
    }
}

 

큐에서 메시지를 수신 및 처리해주는 리스너를 생성한다.

@Slf4j
@ComponentScan
public class TransferListener {

    // 큐에서 메시지 수신 및 처리
    @RabbitListener(queues = "transfer_queue")
    public void processTransfer(TransferRequest request) {
        log.info("이체 처리 시작: {} -> {}, 금액: {}",
            request.getFromAccount(), request.getToAccount(), request.getAmount());


        // 실체 이체 로직 (stub)
        try {
            Thread.sleep(2000);
            log.info("이체 완료: {}", request);
        } catch (InterruptedException ex) {
            log.error("이체 처리 오류 발생: {}", ex.getMessage());
        }
    }

}

 

queue 이름은 설정한 이름으로 생성된다.

 

서비스로직을 생성한다.

@Slf4j
@Service
@RequiredArgsConstructor
public class RabbitTransferService {

    private final RabbitTemplate rabbitTemplate;

    // 이체 요청을 큐에 전송
    public void sendTransferRequest(String fromAccount, String toAccount, BigDecimal amount) {
        TransferRequest request = new TransferRequest();
        request.setFromAccount(fromAccount);
        request.setToAccount(toAccount);
        request.setAmount(amount);

        log.info("큐에 이체 요청 전송: {}", request);
        rabbitTemplate.convertAndSend(RabbitMQConfig.TRANSFER_QUEUE, request);
    }

}

 

컨트롤러를 생성한다.

@Slf4j
@RequiredArgsConstructor
@RestController
public class RabbitTransferController {

    private final RabbitTransferService transferService;

    @PostMapping("/rabbit/transfer")
    public String performTransfer(@RequestParam String from,
                                  @RequestParam String to,
                                  @RequestParam BigDecimal amount) {
        transferService.sendTransferRequest(from, to, amount);
        return "이체 요청 큐 등록 완료";
    }

}

 

 

--

정상적으로 프로젝트 실행 후 포스트맨이나 터미널로 요청한다.

curl -X POST "http://localhost:8080/rabbit/transfer?from=A&to=B&amount=1000"

 

RabbitMq 사이트에서는 아래처럼 나온다.

- Ready 에만 쌓여있으면 리스너에 문제가 있는 것이다. 설정 파일이 잘 되었는지 확인해본다.

 

 

 

정상처리 완료