[MQTT] Spring Boot 에서 MQTT(mosquitto) 사용하기

프론트에서 특정 api를 호출하면 비즈니스 로직을 수행하고 mqtt를 사용해서 윈도우 PC나 키오스크에 데이터를 전달해야 했다. 작은 양의 데이터 였고, 이전에 비슷한 프로젝트에서 mosquitto를 사용해서 구현한 레퍼런스가 있어 MQTT로 이를 구현했다.

그렇기 때문에 여기서 Spring Boot 웹 서버는 publisher라고 할 수 있고, 윈도우 PC는 subscriber, mosquitto는 broker 역할을 한다

 

MQTT 특징

• 경량 프로토콜 
  • 최소화 된 헤더
  • 2바이트 정도의 적은 용량 메세지 -> 제한된 대역폭 환경에서도 사용 가능 
• Publish/Subscribe 모델
  • Publisher (발행 주체) <-> Broker (중간서버) <-> Subscriber (구독 주체)
  • Publisher는 특정 topic에 메시지를 발행함
  • Subscriber는 원하는 주제에 대해 구독을 요청
  • Broker는 해당 topic의 메세지가 발행되면 구독자에게 전달
  • Publisher와 Subscriber는 서로 몰라도 된다

 

QoS (Quality of Service) Level

mqtt 프로토콜 뿐만 아니라 다양한 네트워크 및 통신 영역에서 사용되는 용어로서 자세한 설명과 사진은 아래 커뮤니티 사이트에서 확인 가능하다.

 

What is MQTT Quality of Service (QoS) 0,1, & 2? – MQTT Essentials: Part 6

 

QoS 0

• 메세지는 최대 한번 전송하며 클라이언트에게 전달이 되었는지 여부는 확인하지 않음
• 가장 낮은 수준의 보장 서비스
• 속도가 빠르다
• find and forget이라고 하는데 그 이유는 후속 조치나 확인 없이 메세지가 전송되는 기본 TCP 프로토콜과 유사한 기능을 가지고 있기 때문

Qos 1

• 중간 수준의 서비스
• 적어도 한번 이상 클라이언트 측으로 전달되도록 보장
• 전달 여부를 확인해서 메세지가 정상적으로 전달되지 못하면 메세지를 다시 보내기 위해 시도한다
• 메세지는 중복되어 보내질 수 있으며 한 번만 도착한다는 보장은 없다
• 과정
  • QoS 1로 메세지가 published
  • Publisher는 수신자로 부터 PUBACK 패킷을 수신할 때 까지 메시지의 사본을 보관
  • Publisher가 만약 적절한 시간 내에 PUBACK 패킷을 수신하지 못하면 메세지를 재전송한다
  • 메시지를 전송받은 수신자인 MQTT 브로커는 모든 구독 클라이언트에게 메시지를 배포
  • 브로커는 메세지 수신 확인을 위해 Publisher에게 PUBACK 패킷으로 응답

QoS 2

• 가장 높은 수준의 보장 서비스
• 각 메세지가 의도한 클라이언트에게 한 번만 정확히 수신되도록 함
• 발신자와 수신자(브로커)간에 4부분으로 구성된 핸드세이크를 사용한다.
• 정확히 메세지를 수신하기 위한 과정 때문에 속도가 다른 레벨에 비해 느리다
• 과정
  • broker가 publisher로 부터 PUBLISH 패킷을 받는다
  • broker는 게시 메세지를 처리하고 publisher에게 게시 패킷을 승인하는 PUBREC 패킷으로 응답한다.
  • publisher가 broker로 부터 PUBREC 패킷을 받지 못하면 수신자가 확인을 받을 때까지 DUP 플래그가 포함된 PUBLISH 패킷을 재전송한다
  • publisher는 broker로 부터 PUBREC 패킷을 받으면 publisher는 초기 PUBLISH패킷을 안전하게 삭제한다
  • publisher는 brocker로 부터 PUBREC패킷을 받아 저장하고 PUBREL 패킷으로 응답한다
  • broker는 PUBREL을 받으면 저장된 모든 상태를 삭제하고 PUBCOMP 패킷 응답을 보낸다

 

토픽이란

• MQTT 메세지가 전달되는 경로
• home/factory1/temperature 와 같은 형식의 계층 구조이다
• 메세지를 어떤 용도로 사용할지에 따라 분류해주는 역할
• 클라이언트는 특정 토픽을 구독하고 해당 토픽에서 발행되는 메세지만 수신한다

=> 메시지를 특정 주제나 범주로 구분하는 주소 역할

 

페이로드

• 메시지의 실제 내용
• 데이터를 담고 있는 부분
• home/factory1/temperature 토픽에 {"temperature": 30.1} 과 같은 데이터를 페이로드에 포함시킬 수 있음
• 특정 토픽에서 발행된 메시지가 어떤 정보를 포함하는지 결정

=> MQTT 메시지가 전달하고자 하는 실제 데이터에 해당함

 

클라이언트 ID

• 고유 ID를 설정해야 한다
• 중복된 클라이언트는 id 연결 충돌을 일으켜 나중에 접속하는 client는 disconnect 될 수 있다

 

주요 MQTT

• Mosquitto
• EMQX
• RabbitMQ

등등

 

Spring Boot 구현

의존성 추가

dependencies {
    implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'
}

 

MQTT Configuration 설정

@Configuration
public class MqttConfig {
    @Value("${mqtt.server}")
    private String BROKER_URL;

    @Value("${mqtt.client-id}")
    private String CLIENT_ID;

    @Value("${mqtt.user}")
    private String userName;

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

    @Bean
    public MqttClient mqttClient() throws MqttException {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setKeepAliveInterval(30);
        options.setConnectionTimeout(10);
        options.setUserName(userName);
        options.setPassword(password.toCharArray());

        MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID, new MemoryPersistence());
        client.connect(options);
        client.setCallback(new MqttCallback());

        return client;
    }
}

 

• 여기서 말하는 client란 MQTT server에 접속한 클라이언트를 의미 (spring boot 웹 서버)
• username과 password가 있는 경우 MqttConnectOptions에 추가해 줘야 한다.

 

MqttCallback

@Configuration
public class MqttConfig {
    @Value("${mqtt.server}")
    private String BROKER_URL;

    @Value("${mqtt.client-id}")
    private String CLIENT_ID;

    @Value("${mqtt.user}")
    private String userName;

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

    @Bean
    public MqttClient mqttClient() throws MqttException {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setKeepAliveInterval(30);
        options.setConnectionTimeout(10);
        options.setUserName(userName);
        options.setPassword(password.toCharArray());

        MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID, new MemoryPersistence());
        client.connect(options);
        client.setCallback(new MqttCallback());

        return client;
    }
}

 

MqttPublisherService

@Slf4j
@Service
@RequiredArgsConstructor
public class MqttPublisherService {
    private final MqttClient mqttClient;

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

    public void publish(String topic, String payload) throws MqttException {
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        mqttClient.publish(topic, message);
        log.info("MQTT message published to topic {} :{}", topic, payload);
    }
}

 

• 현재 client인 spring boot 웹 서버는 publisher로서 메세지를 brocker server에 publish한다

mqttPublisherService.publish(MQTT_TOPIC, ttMsg.toString());

• 실제 사용시 이렇게 작성될 수 있다

 

구동 확인 하기 & MQTT Explorer

https://mqtt-explorer.com/

MQTT Explorer

• 위 사이트에서 MQTT Explorer를 다운로드 받아 mqtt 서버에 publish 되는 페이로드와 토픽을 확인할 수 있다

sudo apt update
sudo apt install mosquitto-clients
mosquitto_sub -h {mqtt ip} -p {port} -t {topic}

 

• MQTT Explorer를 사용하지 않고 간단하게 로컬에서 확인할 수 있는 방법
• broker 서버에 해당 topic을 subscribe한다