Spring AMQP를 사용한 메시징

1. 개요

본 포스트에서는 Spring AMQP 프레임워크를 사용하여 AMQP릉 통한 메시지 기반 통신을 살펴보겠습니다.

 

2. 메시지 기반 통신

메시징(Messaging)은 애플리케이션 사이에 통신을 위한 기술입니다.

동기적인 요청/응답 기반 아키텍처 대신 비동기 메시지 전달에 의존합니다.

메시지 생산자(Producer)와 소비자(Consumer)는 메시지 브로커라고 하는 중간 메시징 계층에 의해 분리됩니다.

메시지 브로커는 메시지의 퍼시스턴트 스토리지, 메시지 필터링 및 메시지 변환과 같은 기능을 제공합니다.

 

Java로 개발된 애플리케이션 간의 메시징은 일반적으로 JMS(Java Message Service) API가 사용됩니다.

서로 다른 벤더와 플랫폼 간의 상호 운용성을 위해, JMS 클라이언트와 브로커를 사용할 수 없게되면 AMQP가 유용합니다.

 

3. AMQP - Advanced Message Queuing Protocol

AMQP는 비동기식 메시지 통신을 위한 개방형 표준 유선 스펙입니다.

메시지 구성 방법에 대한 설명을 제공합니다.

 

3-1. JMS와 다른 점

AMQP는 플랫폼 중립적 바이너리 프로토콜 표준이기 때문에 라이브러리는 다른 프로그래밍 언어로 작성될 수 있고, 또한 다른 환경에서 실행될 수 있습니다.

 

하나의 JMS 브로커에서 다른 브로커로 이동할 때와 같이 벤더 기반의 프로토콜 락인(lock-in)이 없습니다.

널리 사용되는 AMQP 브로커는 RabbitMQ, OpenMQ 또는 StormMQ가 있습니다.

 

3-2. AMQP 엔티티

간단히 보면, AMQP는 Exchange(교환), Queues, 그리고 바인딩으로 구성됩니다.

 

Exchange는 우체국이나 사서함과 같이 클라이언트가 AMQP exchange에 메시지를 포스트합니다. 

• Direct Exchange : 라우팅 키가 완전히 일치하는 큐로 라우팅합니다.
• Fanout Exchange : 바인딩된 모든 큐로 라우팅합니다.
• Topic Exchange : 라우팅 키를 패턴과 일치하는 여러 큐로 메시지를 라우팅합니다.
• Headers Exchange : 메시지 헤더를 기반으로 라우팅합니다.

큐들은 라우팅 키를 사용하여 exchange에 바인딩됩니다.

메시지는 라우팅 키로 exchange에 보내집니다. 그런 다음 exchange는 카피를 큐에 배포합니다.

 

3-3. Spring AMQP

Spring AMQP는 spring-amqp와 spring-rabbit의 두 가지 모듈로 구성됩니다.

아래와 같은 추상화를 제공합니다. 

• AMQP 엔티티 : 메시지, 큐, 바인딩 및 Exchange 클래스를 사용하여 엔티티를 만듭니다.
• Connection 관리 : CachingConnectionFactory를 사용하여 RabbitMQ 브로커에 연결합니다.
• 메시지 퍼블리싱 : RabbitTemplate을 사용하여 메시지를 보냅니다.
• 메시지 소비 : @RabbitListener를 사용하여 큐에서 메시지를 읽습니다.

 

4. Rabbitmq Broker 설정

RabbitMQ 브로커가 필요한데, 이를 수행하는 가장 간단한 방법은 Docker를 사용하여 RabbitMQ 이미지를 가져와서 실행하는 것입니다.

docker run -d -p 5672:5672 -p 15672:15672 --name my-rabbit rabbitmq:3-management

애플리케이션이 RabbitMQ에 바인딩할 수 있도록 5672 포트를 노출합니다.

그리고 관리 UI를 통해 RabbitMQ 브로커가 수행하는 작업을 볼 수 있도록 15672 포트를 노출합니다.

http://localhost:15672 or the HTTP API: http://localhost:15672/api/index.html.

 

5. Spring Amqp 애플리케이션 생성

Spring AMQP를 사용해서 "Hello, world!" 메시지를 보내고 받는 간단한 애플리케이션을 만들어봅니다.

5-1. 의존성

spring-amqp 및 spring-rabbit 모듈을 추가하기 위해서 spring-boot-starter-amqp 를 추가합니다.

// https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp
implementation group: 'org.springframework.boot', name: 'spring-boot-starter-amqp', version: '2.6.1'

mvnrepository.com 에서 찾을 수 있습니다.

 

5-2. Rabbitmq 브로커에 연결하기

Spring Boot의 auto-configuration을 사용하여 ConnectionFactory, RabbitTemplate 및 RabbitAdmin 빈을 생성합니다.

결과적으로 기본 username과 password로 "guest"를 사용하여 5672 포트에서 RabbitMQ 브로커에 연결합니다.

@SpringBootApplication 어노테이션이 이를 처리해줍니다.

 

5-3. 큐 만들기

큐를 만들기 위해 단순히 큐 타입의 빈을 정의합니다.

RabbitAdmin은 이것을 찾아 "myQueue"의 라우팅 키를 사용하여 기본 exchange(교환)에 바인딩합니다.

@Bean
public Queue myQueue() {
    return new Queue("myQueue", false);
}

RabbitMQ가 중지되면 큐와 큐에 있는 모든 메시지가 제거되도록 큐를 non-durable로 설정합니다.

하지만 애플리케이션이 다시 시작되도 큐에는 영향을 미치지 않습니다.

 

5-4. 메시지 보내기

RabbitTemplate을 사용해서 "Hello, world!"메시지를 보낼 수 있습니다.

rabbitTemplate.convertAndSend("myQueue", "Hello, world!");

5-5. 메시지 소비하기

@RabbitListener로 메서드에 어노테이션을 달아 메시지 소비자를 구현합니다.

@RabbitListener(queues = "myQueue")
public void listen(String in) {
    System.out.println("Message read from myQueue : " + in);
}

6. 애플리케이션 실행해보기

먼저 RabbitMQ 브로커를 시작합니다.

docker run -d -p 5672:5672 -p 15672:15672 --name my-rabbit rabbitmq:3-management

 

그런 다음 HelloWorldMessge.java 애플리케이션을 실행합니다.

mvn spring-boot:run -Dstart-class=com.baeldung.springamqp.simple.HelloWorldMessageApp

→ 메이븐 빌드

 

애플리케이션이 실행되는 동안 다음과 같은 사항을 확인할 수 있습니다. :

• 애플리케이션은 라우팅 키로 "myQueue"를 사용하여 기본 exchange(교환)에 메시지를 보냅니다.
• 그러면 "myQueue" 큐는 메시지를 수신합니다.
• 마지막으로, listen 메소드는 "myQueue" 로부터 메시지를 소비하고 콘솔에 출력합니다.

RabbitMQ 관리 페이지(http://localhost:15672)를 사용하여 메시지가 전송 및 사용되었는지 확인할 수 있습니다.

 

출처:

https://www.baeldung.com/spring-amqp