MQTT 재연결 후 구독 복구

문제 정의

연결이 단절되었다가 재연결된 뒤 메시지가 더 이상 수신되지 않는 경우가 있다. 구독 목록은 브로커가 세션에 저장하는 상태이므로, 세션이 새로 시작되면 구독도 함께 삭제된다. paho-mqtt는 기본 설정에서 연결할 때마다 세션을 새로 시작하며, 단절된 연결만 다시 수립한다.

연결 콜백 내 구독

on_connect는 브로커의 CONNACK을 수신할 때마다 호출된다. 재연결 때도 호출되므로, 구독을 이 콜백 내부에 두면 연결될 때마다 구독이 복구된다.

import paho.mqtt.client as mqtt

TOPICS = [("plant/line1/+/status", 1), ("plant/line1/+/alarm", 1)]


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code == 0:
        # 재연결할 때도 여기로 들어온다
        client.subscribe(TOPICS)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = lambda c, u, m: print(m.topic, m.payload)
client.connect("broker.example", 1883, keepalive=30)
# 연결이 끊기면 알아서 다시 연결한다
client.loop_forever()

connect() 직후에 한 번만 구독하는 코드는 첫 연결에서만 동작한다.

CallbackAPIVersion

paho-mqtt 2.x는 Client의 첫 인자로 콜백 버전을 받는다. VERSION2의 on_connect는 위 코드와 같이 인자 다섯 개를 받는다. VERSION1은 MQTT 3.1.1에서 client, userdata, flags, rc 네 개를 받으며 폐기 예정(deprecated)이다.

2.1에서는 첫 인자를 생략하면 VERSION1로 동작하면서 DeprecationWarning이 발생한다. 1.x처럼 첫 인자 위치에 client id 문자열을 지정하면 ValueError가 발생한다.

세션 유지 시의 처리

세션을 유지하도록 연결한 경우 flags.session_present가 True로 전달될 수 있다. 이는 브로커에 구독이 남아 있음을 의미한다. 이때 다시 구독해도 메시지가 손실되지는 않지만 retained 메시지(브로커가 보관하는 마지막 값)가 한 번 더 전달된다. 이를 방지하려면 False일 때만 구독한다.