[SpringBoot] MQTT 기반 단말기 통신 API 서버 구축하기

🌟 INDEX

• [SpringBoot] MQTT 기반 단말기 통신 API 서버 구축하기
• 개발 환경

---

[SpringBoot] MQTT 기반 단말기 통신 API 서버 구축하기

이 글에서는 Java Spring Boot를 사용하여 MQTT 기반의 단말기 통신 API 서버를 구축하는 방법을 소개합니다. MQTT 브로커를 통해 단말기와의 실시간 데이터 송수신 기능을 구현하고, 수집된 정보를 데이터베이스에 저장하는 과정을 다룹니다. 

MqttConfig.java

전반적으로 mqtt Broker에서 발행 및 구독되는 message를 읽고, 처리하는 기능들을 MqttConfig 에서 다루고 있습니다.

(추후에 리펙토링이 필요할 것으로 보입니다.)

MqttConfig.java 파일을 생성하고, @Log4j2, @Configuration 어노테이션을 사용해 주었습니다.

 

@Log4j2

클래스에 이 어노테이션을 붙이면 Log4j 2 의 로거 인스턴스가 생성되어 로그를 기록할 수 있습니다.

코드 내에서 로그를 쉽게 작성할 수 있도록 도와주며, 다양한 로그레벨을 지원합니다.

 

@Configuration

해당 클래스가 하나 이상의 Spring Bean을 정의하는 구성 클래스를 나타냅니다.

이 클래스는 Spring 의 IoC 컨테이너에 의해 관리되며, 애플리케이션의 설정을 Java 코드로 정의할 수 있게 해 주며, @Bean 어노테이션과 함께 사용하여 스프링 컨테이너에 등록할 객체를 정의하고 이를 통해 XML 기반 설정을 대체할 수 있습니다.

---

build.gradle 에 MQTT 라이브러리 임포트하기

Spring Integration MQTT를 사용하기 위해 mqtt 의존성을 추가해 줍니다.

dependencies {
    implementation 'org.springframework.integration:spring-integration-mqtt'
}

 

Maven, Gradle 무관하며, 사용하는 설정에 맞춰 의존성을 추가해 주면 됩니다.

application.yml 에 MQTT URL 정보 설정하기

MQTT 브로커 URL 정보를 설정 파일에 추가합니다. application.properties 등 사용하는 설정에 맞게 설정해 줍니다

spring.mqtt.url=tcp://localhost:1883
spring.mqtt.clientId=myClientId
spring.mqtt.username=myUsername // 필요한 경우
spring.mqtt.password=myPassword // 필요한 경우
spring.mqtt.subtopic: LIYO/V1/liyo/+/OUTBOX
pubTopicAddDeviceId: LIYO // 필요한 경우
spring.mqtt.qos: 1

 

URL 외에도 subtopic과 qos 정보도 추가해 주었습니다.

 

**Topic

MQTT에서 토픽은 메시지가 전송되는 경로를 나타내는 문자열입니다.

클라이언트는 특정 토픽을 구독(subscribe)하거나, 해당 토픽으로 메시지를 발행(publish)할 수 있습니다.

토픽은 계층구조로 구성되어 있어 메시지를 조직적으로 관리하고 필터링할 수 있습니다.

 

와일드카드

+ : 한 레벨을 대체합니다. (ex: LIYO/+/liyo)

# : 여러 레벨을 대체합니다. (ex: LIYO/#)

 

QoS 레벨

MQTT의 QoS(Quality of Service)는 메시지가 전송되는 방식과 신뢰성을 정의합니다.

 

QoS 0 : 메시지가 한 번만 전송되며 가장 빠르지만 신뢰성이 가장 낮으며 실시간 데이터 스트리밍 등에 사용됩니다.

QoS 1 : 메시지가 최소한 한 번 이상 전송되며 수신자는 메시지를 반드시 받아야 하며, 중복 수신이 발생할 수 있습니다. 상태 업데이트와 같은 중요한 메시지에 주로 사용됩니다.

QoS 2 : 메시지가 정확히 한 번만 전송되며 가장 높은 신뢰성을 제공하며 중복이 발생하지 않습니다. 전송 과정이 복잡하여 성능이 떨어질 수 있습니다. 금융 거래와 같이 중복이 절대 허용되지 않는 경우에 주로 사용합니다.

 

package com.ebab.beacnd.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import lombok.extern.log4j.Log4j2;

@Log4j2
@Configuration
public class MqttConfig {

    @Value("${spring.mqtt.brokerUrl}")
    private String brokerUrl;

    @Value("${spring.mqtt.qos}")
    private int qos;

    @Value("${spring.mqtt.subTopic}")
    private String subTopic;

    @Value("${spring.mqtt.username}")
    private String username;

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

    @Value("${spring.mqtt.pubTopicAddDeviceId}")
    private String pubTopicAddDeviceId;

 

필드에 설정파일에서 값을 주입받아 해당 값들을 사용할 수 있도록 해줍니다.

---

MQTT 클라이언트 설정

MQTT_CLIENT_ID 와 MQTT_CLIENT_ID2 를 static final 변수로 생성해 주었습니다.

클라이언트 ID가 동일하면 MQTT 브로거가 연결을 끊으므로 발행과 구독 클라이언트 ID를 따로 사용합니다

---

메시지 발행 빈(Bean) 정의

MQTT 클라이언트 생성 - mqttClientFactory

추가되는 코드들은 mqttConfig 코드의 하단에 추가됩니다. 전체 이미지는 생략하겠습니다.

// 메시지 발행 (Publish)
@Bean
public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions options = new MqttConnectOptions();
    options.setServerURIs(new String[] { brokerUrl });
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    factory.setConnectionOptions(options);
    return factory;
}

MqttPahoClientFactory를 설정하여 MQTT 클라이언트를 생성합니다.

MqttConnectOptions를 사용하여 브로커의 URL, 사용자 이름 및 비밀번호를 설정합니다.

 

패스워드는 toCharArray() 메서드를 사용하여 문자열을 문자 배열로 변환하여 보안을 강화해야 합니다.

 

문자열(String)은 불변 객체로, 메모리에 문자열이 남아있을 수 있어 암호화 같은 민감한 정보가 메모리에서 쉽게 접근될 수 있습니다. char[] 배열 객체 사용으로 사용 후 즉시 내용을 지워 메모리에서 비밀번호가 남아 있지 않도록 합니다.

---

MQTT 메시지 발행 채널 설정 - mqttOutbound

@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
    MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(MQTT_CLIENT_ID2, mqttClientFactory());
    messageHandler.setAsync(true);
    messageHandler.setDefaultTopic(pubTopicAddDeviceId);
    messageHandler.setCompletionTimeout(1000);
    return messageHandler;
}

발행 채널을 설정하는 빈입니다. MqttPahoMessageHandler를 사용하여 메시지를 비동기로 발행합니다. 

기본 발행 토픽은 미리 설정해 둔 pubTopicAddDevice 로 설정하였습니다.

---

MQTT 메시지 발행 채널 - mqttOutboundChannel

@Bean
public MessageChannel mqttOutboundChannel() {
    return new DirectChannel();
}

메시지를 발행하는 데 사용되는 채널입니다.

---

MQTT 메시지 전송 인터페이스 설정 - MyGateway

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MyGateway {
    void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic);
}

 

메시지를 MQTT로 전송하기 위한 인터페이스입니다.

@MessageGatewqy 어노테이션을 통해 Spring Integration의 메시지 게이트웨이로 설정됩니다.

 

---

메시지 구독 빈(Bean) 정의

메시지 구독 설정 - mqttInputChannel

@Bean
public MessageChannel mqttInputChannel() {
    return new DirectChannel();
}

MQTT 메시지를 수신할 채널을 정의합니다. DirectChannel은 메시지를 수신하는 데 사용되는 동기적 채널입니다.

---

MQTT 메시지 생산자 설정 - inboundChannel

@Bean
public MessageProducer inboundChannel() {
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(brokerUrl, MQTT_CLIENT_ID, subTopic);
    adapter.setCompletionTimeout(5000);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(qos);
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
}

MQTT 메시지를 수신하는 어댑터를 설정합니다.

• brokerUrl: MQTT 브로커의 URL.
• MQTT_CLIENT_ID: 클라이언트 ID.
• subTopic: 구독할 주제.
• setCompletionTimeout(5000): 메시지 전송 완료 대기 시간 설정 (5초).
• setConverter: 메시지 변환기를 설정합니다.
• setQos(qos): QoS(서비스 품질) 수준 설정.
• setOutputChannel(mqttInputChannel()): 수신한 메시지를 mqttInputChannel로 보냅니다.

---

MQTT 통신 이력 DB 저장 - mqttServiceProcess

public void mqttServiceProcess(MqttDto mdto, String topic, String inOugGb, String payload) {
    mdto.setTopic(topic);
    mdto.setDirection_cd(inOugGb);
    mdto.setPayload(payload);
    mqttService.insertACND0910(mdto);
}

MQTT 메시지를 처리하고, 관련 정보를 데이터베이스에 저장합니다.

관련 데이터를 담을 mdto를 생성해 주었습니다.

토픽 연결 확인

본문내용 삽입

public boolean isTopicConnection(String topic) {
    boolean result = false;
    return result;
}

특정 토픽의 연결 상태를 확인하는 메서드입니다. 현재는 항상 false를 반환합니다.

문자열 반환

public String booleanString(String bool) {
    String returnStr = "N";
    if ("true".equals(bool)) {
        returnStr = "Y";
    }
    return returnStr;
}

문자열 "ture"를 확인하여 "Y" 또는 "N"을 반환합니다. 주로 Boolean 값을 문자열로 변환하는 용도로 사용합니다. 

---

MQTT 요청 처리 - callMqttservice

본문내용 삽입

// MQTT 요청
public String callMqttService(SubPayLoad subPayLoad, String inboxGb) {
    String resMessage = "";
    try {
        log.info(":::::::::::::::::::::::MQTT 전송 전 getDeviceId {}", subPayLoad.getDeviceId());
        // data에 request data 조회하여 가져옴
        
        log.info("resMessage: {}", resMessage);
        myGateway.sendToMqtt(resMessage, pubTopicAddDeviceId);

    } catch (Exception e) {
        log.info(":::::::::::::::::::::::callMqttService Exception : {}", e.toString());
        resMessage = null;
        return resMessage;
    }

    return resMessage;
}

해당 프로젝트에서는 발행은 따로 하지 않고, 구독한 데이터를 DB에 저장하기 까지만 할 것이기 때문에 해당 내용을 담은 Mqtt 요청을 처리하고 메시지를 전송하는 메서드입니다.

 

여기까지 MQTT 메시지를 주고받기 위한 기본 설정이 완료되었습니다.

---

MQTT 메시지 구독 후 처리 - inboundMessageHandler

MQTT 메시지 수신 및 로그 출력

String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
log.info(":::::::::::::::::::::::Topic:{}", topic);
log.info(":::::::::::::::::::::::Payload:{}", message.getPayload());

MQTT 메시지의 토픽과 Payload를 로깅해 줍니다. 브로커로부터 수신한 메시지의 내용을 확인할 수 있습니다.

---

Payload JSON 파싱

String payload = message.getPayload().toString();
ObjectMapper mapper = new ObjectMapper();
SubPayLoad subPayLoad = mapper.readValue(payload, SubPayLoad.class);

페이로드를 JSON 형식으로 간주하고 이를 SubPayload 객체로 변환해 줍니다.

ObjectMapper 를 사용해서 JSON 데이터를 매핑해 주었습니다.

---

메시지 타입에 따른 분기 처리

String messageType = subPayLoad.getMessageType();
switch (messageType) {
    case "EVENT_BOOTUP":
        // 디바이스 부팅 처리
        break;
    case "EVENT_SENSOR_REPORT":
        // 센싱 데이터 처리
        break;
    default:
        break;
}

 

subPayLoad에서 MessageType 을 추출하여 메시지 타입별 분기를 처리합니다.

 

저는 단말기의 부팅과, 센싱 정보를 수집하는 MQTT 서버이기 때문에, 디바이스 부팅, 센싱 데이터 처리를 메시지로 받고 있습니다.

 

해당 정보들을 통해 단말기 설정 정보, 현재 상태, 연결 이력, 센싱 데이터 이력 등을 저장하여 관리할 수 있습니다.

 

---

개발 환경

• Frontend
  • Framework: Vue 3
  • Build Tool: Vite
• Backend
  • Framework: Spring Boot 3.2.2
  • Language: Java
  • Gateway: Spring Cloud Gateway
• Database
  • DBMS: MS SQL Server
  • Container: Docker
• Development Tools
  • IDE: Visual Studio Code (VSCode)