[MQTT / Python / IoT] Python으로 구현하는 MQTT

MQTT란?

MQTT(Message Queueing Telemetry Transport)는 발행 - 구독 기반의 표준 메세지 송수신 프로토콜이다.

주로 IoT와 같이 제한된, 가벼운 장치간의 전송을 위한 프로토콜이다.

TCP/IP 기반 프로토콜 위에서 (인터넷 통신 상에서) 동작하기때문에, wifi모듈을 대체로 사용한다.

가볍고 빠르기 때문에 페메(facebook messenger)에서 채택한 프로토콜이라고 한다.

 

MQTT 프로토콜의 특징

MQTT 프로토콜 구조도

 

브로커(Broker)가 발행자 클라이언트(Publisher Client)와 구독자 클라이언트(Subscriber Client) 간의 송수신을 중개한다.

 

발행자 클라이언트는 특정 Topic을 가진 메시지를 브로커에게 발행(publish) 한다.

브로커는 그 메세지의 토픽을 확인하여 그 Topic을 구독(subscribe)중인 클라이언트에게 전송한다.

구독자 클라이언트는 특정 Topic을 구독할 수 있으며, 구독한 Topic의 메세지만을 수신한다.

 

MQTT의 QoS (Quality of Service)는 총 3단계로 이루어져있으며, 각 특징은 아래와 같다.

QoS 0	At Most Once (최대 한 번 수신되도록)	손실 발생 가능, 빠르고 가벼움 (중복 전송 없음)
QoS 1	At Least Once (적어도 한 번 수신되도록)	손실 없음, 대부분 기본 사용 (중복 전송 가능)
QoS 2	Exactly Once (정확히 한 번 수신되도록)	손실 없음, 오버헤드 큼, 중요한 데이터 처리에 사용 (중복 전송 없음)

 

---

MQTT 프로토콜 구현

 

파이썬으로 MQTT 클라이언트를 구현하기 위해, Eclipse paho-mqtt 라이브러리를 사용하였다.

**파이썬은 MQTT 브로커를 구현하는 라이브러리가 따로 없으므로 mosquitto를 이용하여 구현해야 함. 아래 블로그 참고.

https://dalkomit.tistory.com/119

MQTT 브로커 Mosquitto 설치

 

*본인은 기존 설치된 모스키토 서버(브로커)에 연결되는 클라이언트를 구현할 예정이다.

 

 

https://pypi.org/project/paho-mqtt/

paho-mqtt

$pip install paho-mqtt

 

※라즈베리파이 환경에서는 위 명령어로 설치가 안 될 가능성이 높다. 따라서 아래 명령어를 사용하여 설치한다.

 

$sudo apt install python3-paho-mqtt

 

온습도 센서 장치와의 연결을 가정하고, 라즈베리파이 환경에서 구동 될  Publisher Client 코드를 짜 보자.

---

mqtt_publisher.py

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

BROKER = "본인 브로커 서버 주소"
PORT = 1883
TOPIC = "test/topic/device1"

def main():
    # 1) 클라이언트 생성 & 브로커 연결
    mqttc = mqtt.Client()
    mqttc.connect(BROKER, PORT, 60)

    # 2) 네트워크 루프 별도 스레드로 시작
    #    (PING, 재연결 등 내부 처리)
    mqttc.loop_start()

    try:
        while True:
            # 3) 센서값 가정 (랜덤 생성)
            temperature = round(random.uniform(20.0, 30.0), 2)
            humidity    = round(random.uniform(40.0, 60.0), 2)
            pressure    = round(random.uniform(1010.0, 1015.0), 2)

            # 4) payload 생성
            payload = {
                "device_id": "device1",
                "timestamp": time.time(),  # 보낸 시각
                "data": {
                    "temperature": temperature,
                    "humidity": humidity,
                    "pressure": pressure,
                },
            }

            # 5) publish
            result = mqttc.publish(TOPIC, json.dumps(payload), qos=0)
            # 필요하면 전송 완료까지 기다리기
            # result.wait_for_publish()

            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Message published: {payload}")

            # 6) 2초 대기
            time.sleep(2)

    except KeyboardInterrupt:
        print("\n사용자 종료 (Ctrl+C)")

    finally:
        # 7) 정리
        mqttc.loop_stop()
        mqttc.disconnect()
        print("MQTT 연결 종료")

if __name__ == "__main__":
    main()

 

이 발행자 클라이언트는 온도, 습도, 기압 센서 정보를 2초마다 브로커에게 publish한다.

다양한 정보를 담아 publish하기 위해, json 형식의 데이터를 보낸다.

 

이제 해당 토픽(test/topic/device1)을 구독할 Subscriber Client 코드를 짜보자.

---

mqtt_subscriber.py

import paho.mqtt.client as mqtt
import json

def on_connect(client, userdata, flags, rc): #연결 시 동작을 위한 콜백 함수 정의
    if rc == 0: #연결 성공 시
        print("Connected successfully")
    else: #연결 실패 시
        print(f"Connection failed with code {rc}")
    client.subscribe("test/topic/device1") #토픽 구독
def on_message(client, userdata, msg): #메시지 수신 시 동작을 위한 콜백 함수 정의
    payload = json.loads(msg.payload.decode()) #메시지 페이로드 json형식으로 파싱
    print(f"Received message: {msg.topic} -> {payload['data']}") #메시지 출력
    print(f"Device ID: {payload['device_id']}") #디바이스 ID 출력
    print(f"timestamp: {payload['timestamp']}") #시간 출력

mqttc = mqtt.Client() #클라이언트 객체 생성
mqttc.on_connect = on_connect #연결 시 동작을 위한 콜백 함수 설정
mqttc.on_message = on_message #메시지 수신 시 동작을 위한 콜백 함수 설정

mqttc.connect("본인 브로커 서버 주소", 1883, 60) #서버에 연결

mqttc.loop_forever() #메시지 수신 대기

 

무척 간단하게 구현할 수 있었다. 이제 구독자 클라이언트와 발행자 클라이언트를 순차적으로 실행시켜보자.

 

---

실행 결과

Subscriber Client 실행

 

Subscriber Client는 보다 실제같은 환경에서의 구현을 위해 Rasberrypi 4 기기를 원격으로 연결하여 실행하였다.

 

Publisher Client 실행

 

2초마다 온습도 기압 정보를 주고 받는 모습을 확인 할 수 있다.

 

다음 포스팅에서는 이를 발전시켜, 여러 클라이언트에서 publish된 데이터를 큐 자료구조에 정렬하는 코드를 구현해보겠다.