[Spring Boot] ActiveMQ 연동하기

ActiveMQ란?

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

 ActiveMQ : https://activemq.apache.org

 

1. Docker를 사용하여 ActiveMQ 설치

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

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

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

 

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

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

ActiveMQ 웹 괸리 콘솔

 

2. 의존성 추가

Apache ActiveMQ를 사용하기 위해 'Spring Boot Starter ActiveMQ'를 추가하세요.

Maven Repository

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

 

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'

    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.activemqsample.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. ActiveMQ Service 추가

Queue로 메시지를 발행할 때는 JmsTemplate 클래스에 convertAndSend 메서드를 사용하고, Queue에서 메시지를 구독할 때는 하는 @JmsListener 어노테이션을 사용한 메서드를 추가하세요.
MessageService.java

package com.example.activemq.service;


import com.example.activemqsample.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);
  }

  /**
   * Queue에서 메시지를 구독
   *
   * @param messageDto 구독한 메시지를 담고 있는 MessageDto 객체
   */
  @JmsListener(destination = "${activemq.queue.name}")
  public void reciveMessage(MessageDto messageDto) {
    log.info("Received message: {}", messageDto.toString());
  }
}

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

package com.example.activemqsample.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;
}

 

5. ActiveMQ Controller 추가

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

package com.example.activemq.contoller;

import com.example.activemqsample.dto.MessageDto;
import com.example.activemqsample.service.MessageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
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 객체로 응답을 반환
   */
  @RequestMapping(value = "/send/message", method = RequestMethod.POST)
  public ResponseEntity<?> sendMessage(@RequestBody MessageDto messageDto) {
    messageService.sendMessage(messageDto);
    return ResponseEntity.ok("Message sent to ActiveMQ!");
  }
}

 

6. 메시지 발행 및 구독 확인

message.http 추가

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

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

콘솔 로그 확인

 

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