[AWS] AmazonMQ (feat. ActiveMQ)

Notice

티스토리 블로그 시작!

Recent Posts

Recent Comments

관리 메뉴

안녕세계

[AWS] AmazonMQ (feat. ActiveMQ) 본문

[AWS] AmazonMQ (feat. ActiveMQ)

Junhong Kim 2023. 10. 31. 22:58

728x90

AmazonMQ

AmazonMQ는 AWS에서 제공하는 관리형 메시징 서비스입니다. 현재 AmazonMQ는 Apache ActiveMQ와 RabbitMQ 브로커 엔진을 지원합니다. 본 포스팅에서는 AmazonMQ 브로커 엔진 중 ActiveMQ에 대해 알아봅니다.

ActiveMQ

지원되는 와이어 레벨 프로토콜

AMQP
MQTT
MQTT over WebSocket
OpenWire
STOMP
STOMP over WebSocker

브로커 엔진

ActiveMQ용 AmazonMQ 브로커의 배포 모드는 단일 인스턴스 브로커나 활성/대기 브로커로 생성할 수 있습니다.

단일 인스턴스 브로커

단일 인스턴스 브로커는 하나의 가용 영역(AZ)에 있는 하나의 브로커로 구성됩니다.

브로커는 "애플리케이션(Client) <-> Amazon EBS 또는 Amazon EFS" 볼륨과 통신합니다.

데이터를 여러 가용 영역(AZ)에 중복 저장하여 Amazon EFS는 최고 수준의 내구성과 가용성을 제공합니다.

데이터를 단일 AZ 내 여러 서버에 복제하여 Amazon EBS는 짧은 대기 시간과 높은 처리량에 최적화된 블록 수준 스토리지를 제공합니다.

활성/대기 브로커

활성/대기(active/standby) 브로커는 두 개의 서로 다른 가용 영역에 있는 두 개의 브로커가 중복 페어로 구성됩니다. 활성/대기 브로커는 애플리케이션(Client)와 Amazon EFS가 동기식으로 통신합니다.

일반적으로 한 번에 하나의 브로커 인스턴스만 활성 상태이고 다른 브로커 인스턴스는 대기 상태입니다. 브로커 인스턴스 중 하나가 제대로 작동하지 않으면 AmazonMQ가 해당 인스턴스를 서비스 중지하고 대기 중인 인스턴스가 활성화되면서 통신이 재개됩니다.

활성/대기 브로커는 AmazonMQ에서 2개의 ActiveMQ 웹 콘솔 URL을 제공하지만 한 번에 하나의 URL만 활성화 됩니다. 와이어 레벨 프로토콜에 대해서도 2개의 엔드포인트를 제공하지만 한 번에 각 페어의 한 엔드포인트만 활성화 됩니다. 와이어 레벨 프로토콜 엔드포인트의 경우 Failorver Transport를 사용하여 활성 인스턴스가 다운됐을 때 대기 중인 인스턴스를 사용할 수 있습니다.

Client <-> Broker(active/standby) <-> EFS

ActiveMQ 메시지 전달 방식

Queue 방식

N개의 Consumer가 있을 때, 하나의 메시지가 Round Robin으로 하나의 Consumer에게만 전달됩니다.

Pub/Sub 방식

Topic에 메시지가 발행(Publish)될 때 해당 Topic을 구독(Subscribe)하고 있는 모든 Consumer에게 메시지가 전달됩니다.

Spring과 ActiveMQ 연동하기

AWS 공식 문서에 ActiveMQ를 사용하는 JMS 사용 예제가 있습니다. 본 문서에서는 OpenWire를 사용하는 방법을 알아봅니다.

gradle 설정

api("org.springframework.boot:spring-boot-starter-activemq")

ActiveMQ 설정

@Configuration
class ActiveMQConfig {

    @Bean
    fun jmsTemplate(): JmsTemplate {
        val jmsTemplate = JmsTemplate()
        jmsTemplate.isPubSubDomain = true // Queue 방식대신 PubSub 방식 사용
        jmsTemplate.messageConverter = jacksonJmsMessageConverter()
        jmsTemplate.isExplicitQosEnabled = true // QOS 설정
        jmsTemplate.setDeliveryPersistent(true) // 메시지 영속성 설정
        jmsTemplate.receiveTimeout = 1_000 * 3 // 메시지 수신하는 동안 대기 시간
        jmsTemplate.timeToLive = 1_000 * 60 // 메시지 유효 기간
        jmsTemplate.connectionFactory = ActiveMQConnectionFactory(
            "{id}",
            "{password}",
            "ssl://{broker-url}:61617"
        )
        return jmsTemplate
    }

    @Bean
    fun jacksonJmsMessageConverter(): MessageConverter {
        val converter = MappingJackson2MessageConverter()
        converter.setTargetType(MessageType.TEXT)
        converter.setTypeIdPropertyName("_typeId")
        val typeIdMappings = HashMap<String, Class<*>>()
        typeIdMappings["message"] = MessageDto::class.java
        converter.setTypeIdMappings(typeIdMappings)
        return converter
    }
}

메시지 서비스

@Service
class ActiveMqService(
    private val jmsTemplate: JmsTemplate
) {
    fun send(message: MessageDto) {
        // 토픽을 'hello/world' 로 만들었다면 'hello.world'로 입력
        jmsTemplate.convertAndSend("hello.world", message)
    }

    @JmsListener(destination = "hello.world")
    fun receive(message: MessageDto) {
        println("Received message : {}", messageDto.toString());
    }
}

메시지 발행

@RestController
@RequestMapping("/api/active-mq")
class ActiveMqController(
    private val activeMqService: ActiveMqService
) {
    @GetMapping("/send/{txt}")
    fun publish(@PathVariable("txt") txt: String): String {
        val messageDto = LedMessageDto(txt)
        activeMqService.send(messageDto)
        return "OK"
    }
}