MQTT를 사용한 프로젝트 진행해보기 1

MQTT를 현재 진행중인 프로젝트에 적용시켜보려고한다.

먼저 Homebrew에서 mosquitto를 설치를 한다.

https://formulae.brew.sh/formula/mosquitto

mosquitto

Homebrew가 있다면 간편하게 설치가 가능하다.

brew install mosquitto

다음으로 MQTT가 정상작동하는지 확인해 보았다.

mosquitto

이미 1883port를 사용중이라 하니 1883포트를 사용중인 프로세스를 먼저 찾아보겠다.

이미 사용중인데.. 그냥 Mosquitto서비스를 재시작해보겠다..

아놔..

포트를 바꾸고해보겠다..

문제를 해결했다. 하나의 터미널에서 사용하려고 한 것이 문제였다.ㅠㅠ..

터미널을 열고 아래의 명령어로 MQTT브로커에 메세지를 전송해 보았다.

메세지가 잘 전송되었다.🤗

 

다음으로 현재 진행중인 Spring Boot프로젝트에 MQTT를 설정하고 가상의 센서 데이터를 처리해보겠다.

먼저 MQTT라이브러리를 추가해 주겠다.

진행중인 프로젝트에는 build.gradle을 사용하기 때문에 아래와 같은 형식으로 라이브러리를 추가해주었다.

implementation 'org.springframework.integration:spring-integration-mqtt:5.5.3'

추가해준 다음 MQTT설정을 추가해 주겠다.

SpringBoot에서 MQTT브로커에 연결하려면 몇가지 설정이 필요한데, application.yml파일이나 별도의 설정 클래스를 설정 할 수 있는데 지금은 MqttConfig.java라는 설정 클래스를 만들어 진행해보겠다.

package com.project.config;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.internal.wire.MqttConnect;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;

@Configuration //Spring 설정 클래스 명시
public class MqttConfig {

    /*
        MqttConnectOptions빈을 생성
        MQTT브로커에 연결할 때 사용할 옵션을 설정
     */
    @Bean
    public MqttConnectOptions mqttConnectOptions(){
        MqttConnectOptions option = new MqttConnectOptions();
        /*
            MQTT브로커의 URI를 설정. 로컬 호스트의 1883포트사용
            여러개의 URI설정도 가능하다.
         */
        option.setServerURIs(new String[]{"tcp://localhost:1883"});
        return option;
    }
    /*
        MqttPahoClientFactory빈 생성
        MQTT클라이언트 인스턴스를 생성할 때 사용
     */
    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory(){
        //DefaultMqttPahoClientFactory클래스는 MqttPahoClientFactory인터페이스를 구현
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        //생성한 MqttConnectOptions빈을 사용하여 연결 옵션을 설정
        factory.setConnectionOptions(mqttConnectOptions());
        return factory;
    }
    
    //tcp://localhost:1883를 쓰는 MQTT브로커 URL로 "someClientId"는 MQTT클라이언트 ID
    @Bean
    public MqttClient mqttClient()throws MqttException{
        return new MqttClient("tcp://localhost:1883", "someClientId");
    }
    
    @Bean
    public MqttPahoMessageDrivenChannelAdapter messageDrivenChannelAdapter(MqttPahoClientFactory mqttPahoClientFactory){
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("someClientId", mqttPahoClientFactory, "someTopic");
        adapter.setCompletionTimeout(5000);  // 타임아웃 설정
        adapter.setConverter(new DefaultPahoMessageConverter());  // 메시지 컨버터 설정
        adapter.setQos(1);  // Quality of Service 설정
        adapter.setOutputChannelName("mqttInputChannel");//출력 채널 설정
        return adapter;
    }
}

MqttConnectOptions는 MQTT브로커에 연결할 때 사용할 옵션을 설정하는 클래스이다. setServietURIs는 연결할 MQTT브로커의 주소를 설정한다 이때 여러 주소를 배열로 전달이 가능하다.

MqttPahoClientFactory는 MQTT클라이언트 객체를 생성하는 팩토리이며 setConnectionOptions는 MQTT브로커에 연결할 때 사용할 옵션을 설정한다.

mqttClient 메서드는 mqttClient빈을 생성하는 코드이다. Paho MQTT라이브러리의 MqttClient클래스의 인스턴스를 생성한다.

 

MqttPahoMessageDrivenChannelAdapter메서드는 MQTT메세지를 구독하고 처리하는 역할을 담당한다.

Quality of Service설정은 해당 레벨을 설정하는 것인데 위에는 레벨 1로 설정해두었다. 이는 메세지가 최소 한 번은 도착하도록 하는 설정이다.

 

다음으로 센서 데이터를 생성하는 로직을 추가해보겠다.