[Server] Mac 환경 로컬로 Mosquitto, Spring Integration MQTT 사용

프로젝트 개요

앱 <-> 서버 <-> 디바이스로 구성된 프로젝트를 수행하던 중 다수의 사용자가 앱에서 발행하는 이벤트를 디바이스에서 빠르고 가볍게 받을 수 있도록 하기 위해 MQTT를 도입하였다.

 

 

 

• Spring 서버에 MQTT 브로커 설치
  • Mosquitto, EMQX, HiveMQ 등 MQTT 브로커를 서버에 설치.
  • 포트(보통 1883, 또는 TLS일 경우 8883) 열기.
  • MQTT 브로커가 외부(디바이스)에서 접근 가능해야 하므로 방화벽/보안 그룹 설정 필요.
• 디바이스에서 MQTT 클라이언트 구현
  • 디바이스에서 MQTT 클라이언트를 실행시켜서 서버의 브로커에 연결.
  • 예: mqtt://your.server.ip:1883
  • 토픽을 구독하여 서버에서 publish하는 메시지를 받을 수 있게 함.
• Spring 서버에서 MQTT 메시지 발행
  • 디바이스가 구독 중인 토픽으로 메시지를 발행 (publish).
  • Spring에서는 Eclipse Paho 클라이언트를 이용하거나, MQTT 지원 라이브러리(e.g., spring-integration-mqtt)를 이용.

 

1. Mac 환경에서 Mosquitto 설치

설치

brew install mosquitto

 

 

실행

brew services start mosquitto

 

conf로 실행

/opt/homebrew/opt/mosquitto/sbin/mosquitto -c /opt/homebrew/etc/mosquitto/mosquitto.conf

 

 

2. Spring Integration MQTT 

 

https://docs.spring.io/spring-integration/reference/mqtt.html#mqtt-outbound

MQTT Support :: Spring Integration

MqttConnectOptions

https://eclipse.dev/paho/files/javadoc/org/eclipse/paho/client/mqttv3/MqttConnectOptions.html

MqttConnectOptions

 

 

 

build.gradle

// Spring Integration
implementation 'org.springframework.boot:spring-boot-starter-integration'

// Spring MQTT
implementation "org.springframework.integration:spring-integration-mqtt:6.4.3"

 

configuration

@Configuration
public class MqttOutboundConfig {

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

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

    @Value("${mqtt.default-topic}")
    private String defaultTopic;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{brokerUrl});
        options.setCleanSession(true);
        factory.setConnectionOptions(options);

        return factory;
    }

    @Bean
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler(clientId, mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(defaultTopic);
        return messageHandler;
    }

    @Bean
    public IntegrationFlow mqttOutboundFlow() {
        return IntegrationFlow.from("mqttOutboundChannel")
                .handle(mqttOutbound())
                .get();
    }

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


}

1. 변수

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

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

@Value("${mqtt.default-topic}")
private String defaultTopic;

brokerUrl : MQTT 브로커가 실행되고 있는 주소/포트를 의미한다.

clientId : MQTT 브로커에 접속할 경우 ClientId를 기입해야 한다. 중복된 Id로는 접속이 불가하다.

 

2. mqttClientFactory

 @Bean
public MqttPahoClientFactory mqttClientFactory() {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    MqttConnectOptions options = new MqttConnectOptions();
    options.setServerURIs(new String[]{brokerUrl});
    options.setCleanSession(true);
    factory.setConnectionOptions(options);

    return factory;
}

 

 

Factory : 객체를 만들어 주는 역할을 하는 일종의 생산자 클래스.

Spring Integration에서는 MQTT를 다룰때 내부적으로 Paho 라이브러리를 사용하므로 Paho Client를 생성하려면 설정이 필요함.

MqttConnectOptions -> 설정. 설정을 담아놓는 생산자 -> Factory

 

구체적인 옵션에 관한건 위의 링크를 참조하면 된다.

 

3. mqttOutbound(Handler)

@Bean
public MessageHandler mqttOutbound() {
    MqttPahoMessageHandler messageHandler =
            new MqttPahoMessageHandler(clientId, mqttClientFactory());
    messageHandler.setAsync(true);
    messageHandler.setDefaultTopic(defaultTopic);
    return messageHandler;
}

MQTT 메시지를 Mosquitto에 보내는 역할을 하는 핸들러.

Integration Flow에서 .handle(mqttOutbound())로 사용된다.

 

4. mqttOutboundFlow(Integration Flow)

@Bean
public IntegrationFlow mqttOutboundFlow() {
    return IntegrationFlow.from("mqttOutboundChannel")
            .handle(mqttOutbound())
            .get();
}

mqttOutboundChannel에 대한 IntegrationFlow를 관리한다.

 

service

@Service
@RequiredArgsConstructor
public class MqttPublisherService {

    private final MessageChannel mqttOutboundChannel;

    public void publish(String topic, String payload) {
        Message<String> message = MessageBuilder.withPayload(payload)
                .setHeader(MqttHeaders.TOPIC, topic)
                .build();
        mqttOutboundChannel.send(message);
    }
}

 

 

mqttOutboundChannel은 Spring Integration 내부에서 사용되는 MessageChannel의 이름이다. Spring이 이 이름으로 메시지 라우팅을 할 수 있도록 한다.

 

QOS

MQTT에서 메시지 전송 품질 수준을 정하는 설정으로

0 : At most once (최대 한번) -> 실패해도 재전송 X

1 : At least once (최소한번) -> 성공할 때까지 재전송 (중복가능)

2 : Exactly once (정확히 한번) -> 가장 안전하지만 느림.

 

public void publish(String topic, String payload) {
    Message<String> message = MessageBuilder.withPayload(payload)
            .setHeader(MqttHeaders.TOPIC, topic)
            .setHeader(MqttHeaders.QOS, 0)
            .build();
    mqttOutboundChannel.send(message);
}

Spring 사용시 이 부분에 들어가게 된다.