[Monitoring System] 2. MQTT 통신

728x90

: Message Queuing Telemetry Transport

가장 먼저 데이터 수집을 위해 MQTT 통신이 무엇인지부터 알아보도록 하겠다.

원리

Publish - Subscribe 모델
- 메세지 발신자와 수신자를 구분함
- 브로커 라는 제 3의 구성요소가 둘 사이의 통신 처리
브로커의 publisher - subcriber 분리
- 공간 분리
  - 발신자 / 수신자가 서로의 IP / 포트번호 / 네트워크 위치 모름
- 시간 분리
  - 동시에 일생되거나 네트워크 통해 연결되지 않음
- 동기화 분리
  - 서로를 중단시키지 않고 메세지 송수신 가능. 구독자가 데이터 올 때까지 기다릴 필요 없다.

요약하면, 메세지 발신자가 주제 + 데이터를 브로커에게 보내면, 브로커는 그 주제를 구독하고 있는 수신자를 찾아서 받은 데이터를 전송해준다. SNS 알림설정처럼 일단 발신자가 퍼블리싱을 하고 나면, 브로커가 발신자(정확히는 발신자가 포스팅한 주제)에 관심을 보이는 구독자들에게 데이터를 보내주는 형식이다. 발신자도 구독한 쪽을 알 필요가 없고, 구독자도 발신자가 아니라 발신자의 데이터에 관심이 있기 때문에 서로 몰라도 된다. 중간 처리도 브로커가 다 해준다!

구성요소

MQTT 클라이언트
- 메세지를 보내고 받는 모든 디바이스. MQTT를 통해 통신하면 모두 클라이언트
MQTT 브로커
- 클라이언트 간 메세지 조정 백엔드 시스템
- 메세지 수신 및 필터링 / 클라이언트 식별 / 전송
- 클라이언트 권한 부여 및 인증
- 추가 분석을 위한 타 시스템으로 메세지 전달
- 누락 메세지 / 클라이언트 세션 처리
MQTT 연결
- 클라이언트 - 브로커의 통신 방식
- 클라이언트 = connect 를 보내 연결 시작, 브로커는 connack 으로 응답해 연결 확인
- TCP/IP 기반

Mosquitto Broker 설치

무료 MQTT broker인 Mosquitto를 사용했다.

https://mosquitto.org/download/

[Download

Source mosquitto-2.0.15.tar.gz (GPG signature) Git source code repository (github.com) Older downloads are available at https://mosquitto.org/files/ Binary Installation The binary packages listed be

mosquitto.org](https://mosquitto.org/download/)

위 링크에서 운영체제에 맞는 방법으로 설치해준다. 그리고 파이썬에서는

pip isntall paho-mqtt

로 패키지를 설치해준다. 위 패키지가 python에서 mqtt를 가능하도록 기능을 제공하고 있다. 외부접속을 하고 싶으면 따로 설정을 더해줘야하지만, 우선 지금은 로컬에서 테스트 할 것이라서 생략한다. 궁금하시다면 요기 https://iotmaker.kr/2021/08/23/mosquitto-remote-access-for-windows/

[윈도우즈에 설치한 Mosquitto에 원격 접속허용하기 | 산구루의 IoT 공방 윈도우즈에 설치한 Mosquitto

ESP32의 PWM출력을 위한 ledcSetup,ledcAttachPin,ledcWrite함수의 사용법을 설명합니다.

iotmaker.kr](https://iotmaker.kr/2021/08/23/mosquitto-remote-access-for-windows/)

그리고 mosquitto가 설치된 곳으로 이동해서 명령어를 입력한다.

C:\Program Files\mosquitto > mosquitto -v

하나 유의할 점은, 위 명령어를 실행했을 때 opening ipv6 listen socket on port 1883 이 안 나오고 뭔가 에러가 난다면 지금 1883 포트를 다른 프로그램에서 사용중인 것이다. 이때는 직접 1883 포트를 사용하고 있는 프로세스를 kill 해서 포트를 열어줘야한다. mosquitto는 기본적으로 1883번 포트를 사용한다.

netstat -ano

위 명령어를 사용해서 모든 포트와 pid를 확인하고, 0.0.0.0:1883을 쓰고 있는 녀석을 작업관리자-세부정보에서 중지해주고 다시 실행해보면 된다.

Python 예제

1. Publisher

import paho.mqtt.client as mqtt
import random
import time

mqttc = mqtt.Client("python_pub")
mqttc.connect("BROKER_IP_ADDRESS", 1883)

for i in range(5):
    mqttc.publish("Temperature", "0 " + str(random.uniform(10, 18)))
    print("published 0 ", i)
    time.sleep(1)
    mqttc.publish("Temperature2", "1 " + str(random.uniform(10, 18)))
    print("published 1", i)
    time.sleep(9)

위 예제는 간단하게 10에서 18 사이의 랜덤한 실수값을 Temperature과 Temeprature1이라는 주제에 10초에 한 번, 총 5번 퍼블리싱하는 내용이다.

mqttc = mqtt.Client("python_pub")
mqttc.connect("BROKER_IP_ADDRESS", 1883)

BROKER_IP_ADDRESS 에는 로컬 컴퓨터의 IP 주소를 넣어주면 된다. mqttc라는 이름으로 클라이언트를 선언했고, connect를 활용해 연결했다.

mqttc.publish("Temperature", "0 " + str(random.uniform(10, 18)))

publish 함수는 0번 인자로 Topic, 1번 인자로 메세지를 전송한다.

2. Subscriber

import paho.mqtt.client as mqtt
import time 

from db import models
from db.base import SessionLocal

def add_data(db: any, sensor_id: int, datetime: str, value: float):
    db_data = models.Degree(time=datetime, sensor_id=sensor_id, degree=value)
    # database session에 db_data 정보 추가
    db.add(db_data)
    # db에 반영
    db.commit()
    # 최신 db 정보 받아오기
    db.refresh(db_data)
    return db_data

# subscriber callback
def on_message(client, userdata, message):
        print(message)
        converted = str(message.payload.decode("utf-8")).split()
        sensor_id = int(converted[0])
        value = float(converted[1])
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        add_data(db=SessionLocal(), sensor_id=sensor_id, datetime=timestamp, value=value)

broker_address = "BROKER_IP_ADDRESS"
client1 = mqtt.Client("client1")
client1.connect(broker_address, 1883)
client1.subscribe("Temperature")
client1.on_message = on_message
client1.loop_forever()

publisher보다 긴 것은 데이터베이스 입력 때문이다. 데이터베이스 연관된 부분을 제외하고 들여다보자.

broker_address = "BROKER_IP_ADDRESS"
client1 = mqtt.Client("client1")
client1.connect(broker_address, 1883)
client1.subscribe("Temperature")
client1.on_message = on_message
client1.loop_forever()

똑같이 클라이언트를 선언하고, 브로커의 ip 주소와 포트로 연결한다. 그리고 subscribe 함수를 활용해서 관심있는 Topic을 구독했다. 메세지를 브로커에게서 전달받았을 때의 함수를 정의해주고, loop_forever()로 메세지를 기다린다. 이 부분은 기다리게 두지 않고 주기적으로 확인해서 받아와도 괜찮다.

주기적 확인으로 변경하면 클라이언트에서 다른 일을 동시에 처리하는 것도 가능하므로 경제적으로 프로세스를 사용할 수 있다. 지금은 그냥 쓰도록 하겠다.

def on_message(client, userdata, message):
        print(message)
        converted = str(message.payload.decode("utf-8")).split()
        sensor_id = int(converted[0])
        value = float(converted[1])
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        add_data(db=SessionLocal(), sensor_id=sensor_id, datetime=timestamp, value=value)

message를 브로커로부터 받은 뒤에 처리를 해서 db에 데이터를 넣어준다. 이 경우 온도센서라고 가정했기에 문자열을 가공해서 sensor id, degree로 나누어주었고 현재 시간을 포함해서 데이터를 db에 넣어주었다.

db 업데이트는 sqlalchemy 를 사용했으므로, 이번 포스트에서는 쓰지 않고! 다음 포스팅에서 api를 작성하면서 써보도록 하겠다.

실행은 일반 파이썬 파일 실행과 똑같이 터미널에 python publisher.py / python subscriber.py 해주면 된다.

이 예제에서 MQTT broker는 데이터 전달을 보장해주지 않는다. 그러므로 publisher가 먼저 동작해서 데이터를 보내고, subscriber가 나중에 동작하게 되면 publisher가 앞서 보낸 데이터는 날아가게 된다. 이것은 broker 세팅을 변경해서 해결할 수 있다. 이를 QoS라고 한다. 0은 보내고 잊어버리는 것, 1은 받았는지 모르겠으면 정해진만큼 추가로 전송, 2는 받았다는 것이 확실하도록 보장하는 방식이다.

다만 2로 갈수록 성능이 떨어지고 0으로 갈수록 데이터 손실이 발생하기 때문에 주로 0~1로 사용하고 상위 어플리케이션에서 데이터 손실을 커버하는 방법이 일반적이다.