[Spring Boot] ActiveMQ + STOMP 연동

ActiveMQ란?

• Apache Software Foundation에서 개발된 오픈 소스 메시지 브로커(Message Broker)로 메시징 시스템에서 메시지를 중개하고 전달하는 역할을 담당합니다.
• Java로 작성되었으며, Java Message Service (JMS)를 지원하여 다양한 애플리케이션 간에 비동기적인 통신을 가능하게 합니다.
• 큐(Queue)와 토픽(Topic)이라는 두 가지 메시지 전달 방식을 지원하며, 큐(Queue) 방식은 메시지를 수신 대기 중인 클라이언트 중 하나에게 전달하고, 토픽(Topic) 방식은 메시지를 여러 구독자에게 전달하는 방식입니다. 

 ActiveMQ : https://activemq.apache.org

STOMP란?

• Apache Software Foundation에서 개발된 오픈 소스 메시지 브로커(Message Broker)로 메시징 시스템에서 메시지를 중개하고 전달하는 역할을 담당합니다. 

STOMP: http://stomp.github.io

 

1. Docker를 사용하여 ActiveMQ 설치

Docker를 사용하여 ActiveMQ를 로컬에 설치하세요.

docker run -p 61616:61616 -p 61613:61613 -p 8161:8161 rmohr/activemq

Docker Hub : https://hub.docker.com/r/rmohr/activemq

 

포트 61616는 ActiveMQ 브로커(Broker) 연결, 포트 61613는 STOMP 연결, 포트 8161는ActiveMQ 웹 관리 콘솔에 사용됩니다.

• 웹 관리 콘솔 : http://localhost:8161
• 기본 계정 및 비밀번호 : admin/admin

 

ActiveMQ 웹 괸리 콘솔

 

2. 의존성 추가

Apache ActiveMQ를 사용하기 위해 'Spring Boot Starter ActiveMQ'를 추가하고, WebSocket(STOMP)을 사용하기 위해 'Spring Boot Starter WebSocket', 'Spring Boot Starter Reactor Netty'을 추가하세요.

Maven Repository

https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-activemq

https://mvnrepository.com/search?q=spring-boot-starter-reactor-netty
https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-reactor-netty

 

build.gradle

plugins {
    id 'java'
    id 'org.springframework.boot' version '3.1.0'
    id 'io.spring.dependency-management' version '1.1.0'
}

group = 'com.example'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = '17'

configurations {
    compileOnly {
        extendsFrom annotationProcessor
    }
}

repositories {
    mavenCentral()
}

dependencies {
    // Spring Boot
    implementation 'org.springframework.boot:spring-boot-starter-web'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'

    // ActiveMQ
    implementation 'org.springframework.boot:spring-boot-starter-activemq'

    // WebSocket
    implementation 'org.springframework.boot:spring-boot-starter-websocket'
    implementation 'org.springframework.boot:spring-boot-starter-reactor-netty'

    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    testCompileOnly 'org.projectlombok:lombok'
    testAnnotationProcessor 'org.projectlombok:lombok'
}

tasks.named('test') {
    useJUnitPlatform()
}

 

3. ActiveMQ 설정

ActiveMQ 연결을 설정하고, 메시지 발행과 구독을 위한 큐(Queue)명을 추가하세요.
ActiveMQConfig.java

package com.example.activemq.config;

import com.example.activemq.dto.MessageDto;
import jakarta.jms.Queue;
import java.util.HashMap;
import java.util.Map;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.MessageType;

@Configuration
public class ActiveMQConfig {

  @Value("${spring.activemq.broker-url}")
  private String activemqBrokerUrl;

  @Value("${spring.activemq.user}")
  private String activemqUsername;

  @Value("${spring.activemq.password}")
  private String activemqPassword;

  @Value("${activemq.queue.name}")
  private String queueName;

  /**
   * 지정된 큐 이름으로 Queue 빈을 생성
   *
   * @return Queue 빈 객체
   */
  @Bean
  public Queue queue() {
    return new ActiveMQQueue(queueName);
  }

  /**
   * ActiveMQ 연결을 위한 ActiveMQConnectionFactory 빈을 생성하여 반환
   *
   * @return ActiveMQConnectionFactory 객체
   */
  @Bean
  public ActiveMQConnectionFactory activeMQConnectionFactory() {
    ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
    activeMQConnectionFactory.setBrokerURL(activemqBrokerUrl);
    activeMQConnectionFactory.setUserName(activemqUsername);
    activeMQConnectionFactory.setPassword(activemqPassword);
    return activeMQConnectionFactory;
  }

  /**
   * JmsTemplate을 생성하여 반환
   *
   * @return JmsTemplate 객체
   */
  @Bean
  public JmsTemplate jmsTemplate() {
    JmsTemplate jmsTemplate = new JmsTemplate(activeMQConnectionFactory());
    jmsTemplate.setMessageConverter(jacksonJmsMessageConverter());
    jmsTemplate.setExplicitQosEnabled(true);    // 메시지 전송 시 QOS을 설정
    jmsTemplate.setDeliveryPersistent(false);   // 메시지의 영속성을 설정
    jmsTemplate.setReceiveTimeout(1000 * 3);    // 메시지를 수신하는 동안의 대기 시간을 설정(3초)
    jmsTemplate.setTimeToLive(1000 * 60 * 30);  // 메시지의 유효 기간을 설정(30분)
    return jmsTemplate;
  }

  /**
   * JmsListenerContainerFactory을 위한 빈을 생성
   *
   * @return JmsTemplate
   */
  @Bean
  public JmsListenerContainerFactory<?> jmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(activeMQConnectionFactory());
    factory.setMessageConverter(jacksonJmsMessageConverter());
    return factory;
  }

  /**
   * MessageConverter을 위한 빈을 생성
   *
   * @return MessageConverter
   */
  @Bean
  public MessageConverter jacksonJmsMessageConverter() {
    MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    converter.setTargetType(MessageType.TEXT);
    converter.setTypeIdPropertyName("_typeId");
    Map<String, Class<?>> typeIdMappings = new HashMap<>();
    typeIdMappings.put("message", MessageDto.class);
    converter.setTypeIdMappings(typeIdMappings);
    return converter;
  }
}

application.yml
ActiveMQ 연결 정보 및 Queue 정보를 추가하세요.

# Service Post 설정
server:
  port: 9091

# Log Level 설정
logging:
  level:
    root: info

# ActiveMQ 연결 정보
spring:
  activemq:
    broker-url: tcp://localhost:61616
    user: admin
    password: admin

# Activemq queue 정보
activemq:
  queue.name: sample-queue

 

4. WebSocket 설정

WebSocketMessageBrokerConfigurer 클래스를 상속받아 WebSocket 엔드포인트 및 메시지를 구독할 경로를 설정하세요.
WebSocketConfig.java

package com.example.activemq.config;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

@Slf4j
@RequiredArgsConstructor
@EnableWebSocketMessageBroker // WebSocket 메시지 브로커를 활성화
@Configuration
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

  @Value("${spring.activemq.stomp.host}")
  private String activemqStompHost;

  @Value("${spring.activemq.stomp.port}")
  private String activemqStompPort;

  @Value("${spring.activemq.user}")
  private String activemqUsername;

  @Value("${spring.activemq.password}")
  private String activemqPassword;

  /**
   * STOMP 관련 설정을 구성
   */
  @Override
  public void registerStompEndpoints(StompEndpointRegistry registry) {
    registry.addEndpoint("/ws") // WebSocket 엔드포인트 설정
        .setAllowedOriginPatterns("*")
        .withSockJS(); // withSockJS()를 사용하여 SockJS를 활성화
  }

  /**
   * 메시지 브로커 관련 설정을 구성
   */
  @Override
  public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.enableStompBrokerRelay("/queue") // 메시지를 구독할 경로를 설정
        // ActiveMQ 브로커와 연결을 위한 호스트, 가상 호스트 및 포트, 관리자 로그인 설정
        .setRelayHost(activemqStompHost)
        .setRelayPort(Integer.parseInt(activemqStompPort))
        .setSystemLogin(activemqUsername)
        .setSystemPasscode(activemqPassword)
        .setClientLogin(activemqUsername)
        .setClientPasscode(activemqPassword);
  }
}

application.yml
STOMP 연결 정보를 추가하세요.

# Service Post 설정
server:
  port: 9091

# Log Level 설정
logging:
  level:
    root: info

# ActiveMQ 연결 정보
spring:
  activemq:
    broker-url: tcp://localhost:61616
    stomp:
      host: localhost
      port: 61613
    user: admin
    password: admin

# Activemq queue 정보
activemq:
  queue.name: sample-queue

 

5. ActiveMQ Service 추가

Queue로 메시지를 발행할 때는 JmsTemplate 클래스에 convertAndSend 메서드를 사용하세요.
MessageService.java

package com.example.activemq.service;

import com.example.activemq.dto.MessageDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@RequiredArgsConstructor
@Service
public class MessageService {

  @Value("${activemq.queue.name}")
  private String queueName;

  private final JmsTemplate jmsTemplate;

  /**
   * Queue로 메시지를 발행
   *
   * @param messageDto 발행할 메시지의 DTO 객체
   */
  public void sendMessage(MessageDto messageDto) {
    log.info("message sent: {}", messageDto.toString());
    jmsTemplate.convertAndSend(queueName, messageDto);
  }
}

MessageDto.java
메시지를 보내기 위한 DTO 클래스를 추가하세요.

package com.example.activemq.dto;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.ToString;

@Getter
@Setter
@ToString
@AllArgsConstructor
@NoArgsConstructor
public class MessageDto {

  private String title;
  private String content;
}

 

6. ActiveMQ Controller 추가

클라이언트로부터 메시지를 받아 해당 메시지를 발행할 수 있도록 간단한 REST API 추가하세요.
MessageController.java

package com.example.activemq.contoller;

import com.example.activemq.dto.MessageDto;
import com.example.activemq.service.MessageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RequiredArgsConstructor
@RestController
public class MessageController {

  private final MessageService messageService;

  /**
   * Queue로 메시지를 발행
   *
   * @param messageDto 발행할 메시지의 DTO 객체
   * @return ResponseEntity 객체로 응답을 반환
   */
  @CrossOrigin
  @RequestMapping(value = "/send/message", method = RequestMethod.POST)
  public ResponseEntity<?> sendMessage(@RequestBody MessageDto messageDto) {
    messageService.sendMessage(messageDto);
    return ResponseEntity.ok("Message sent to ActiveMQ!");
  }
}

 

7. 메시지 발행 확인

message.http 추가

### 메시지 발행
POST http://localhost:9091/send/message
Content-Type: application/json

{
    "title" : "Message Title Test",
    "content" : "Message Content Test"
}

콘솔 로그 확인

 

8. 메시지 구독 확인

APIC을 사용하여 WebSocket을 통한 STOMP를 테스트할 수 있습니다.

apic : https://docs.apic.app/tester/test-websocket

 

WebSocket을 통한 STOMP 테스트

• New tab에서 ws를 선택
• Request URL에 서버 정보를 입력: http://localhost:9091/ws
• Connection type에서 Stomp을 선택
• Subscription URL 입력: /queue/sample-queue
• 오른 쪽 상단에 Connect을 클릭
• "Stomp connected." 문구가 나오면 정상적으로 구독 중

 

소스 코드는 Github Repository - https://github.com/tychejin1218/message-queue.git 에서 active-stomp-sample 프로젝트를 참조하세요.