[MQTT] presence detection 에서 qos 와 retain 설정

 

최근 진행한 것 중에 유용하게 잘 사용한 paho-mqtt option이 있어서 기록한다.

 

아래와 같은 mqtt 를 개발하고 있었다

mqtt 통신의 신뢰성을 높이기 위해서 main client 에서 sub client 의 연결 상태(online/offline)를 실시간으로 확인하고 싶었다.

 

그래서 qos, retain을 설정하고, will_set도 등록했다

아래 예시를 보며 어떻게 사용했는지 확인해보자!

 

✨ 여기서 새로 알게된 용어 : presence detection

→ IoT, MQTT, 채팅 시스템에 많이 쓰이는 개념이다.

→ 현재 누가 연결되었는지(online), 누가 연결이 끊어졌는지(offline)에 대한 상태를 시스템이 실시간으로 파악하는 기능이다.

→ "클라이언트의 접속 상태 감지" 라고 생각하면 될 듯 !

 

 

 

우선, sub client에서는 {"client1": "online"} 이런 형식으로 publish 할 예정 !!

 

Main Client 측 스크립트에는 이렇게 추가했다.

self.status_map = {}

def on_connect(self, client, userdata, flags, rc):
    ...
    self.subscribe("status/#")
    logger.info("[MQTT] Subscribe 'agents/status'!")
    ...

def on_message(self, client, userdata, msg):
    ...
    if msg.topic.startswith('status'):
        self._handle_status_update(message)
    ...
  
def _handle_status_update(self, message):
    agent = message.get("agent")
    status = message.get("status")
    if agent and status:
        self.status_map[agent] = status
        logger.info(f"[STATUS] {agent} → {status}")
    logger.info(f'[STATUS_MAP] status_map: {self.status_map}')

def get_agent_status(self, agent):
    return self.status_map.get(agent, "unknown")

def is_agent_online(self, agent):
    return self.get_agent_status(agent) == "online"

 

(1) status_map : 연결된 agents 상태를 담을 dictionary

 

(2) on_connect()

- mqtt 연결 완료 후 "status/#" 구독 시작

- "status/#" : "status/" 으로 시작하는 모든 topic을 구독

 

(3) on_message()

- message를 받으면 _handle_status_update 실행

 

(4) _handle_status_update()

- status_map이라는 dict에 {“agent1”: “online”, “agent2”: “offline”} 형식으로 상태를 저장

 

(5) is_agent_online()

- 나중에 agent 연결 확인할 때 사용할 함수

 

sub client 측에는 online/offline 을 publish 하도록 추가했다.

1️⃣ connect 되면 online 전송

def on_connect(self, client, userdata, flags, rc):
    ...
    online_message = {self.client_id: "online"}
    self.client.publish("agents/status", json.dumps(online_message), qos=1, retain=True)

 

2️⃣ 정상적인 disconnect 에서 offline 전송

def on_disconnect(self, client, userdata, flags, rc):
    offline_message = {self.client_id: "offline"}
    self.client.publish("agents/status", json.dumps(offline_message), qos=1, retain=True)

 

3️⃣ 비정상적인 disconnect 에서 offline 전송

client.connect(broker_host, broker_port)
will_message = {client_id: "offline"}
client.will_set("agents/status", payload=json.dumps(will_message), qos=1, retain=True)

* will_set : client가 비정상적으로 disconnect 되었을 때 broker가 대신 publish하는 message

 

⭐️ qos=1, retain=True

• 설정을 추가한 이유
  • sub client가 먼저 mqtt 연결되고 나서 Main Client가 "status/#" 구독을 하게되면, sub client들의 상태를 받을 수 없다.
• retain=True
  • broker에 마지막 메시지를 저장(retain) 하라고 알려주는 flag
  • 새로운 subscriber가 구독하면 broker는 저장된 메시지를 즉시 전달한다.
• qos=1
  • 메시지가 최소 1회 이상 전달됨 (at least once)
  • publish → broker → subscriber 간 ACK(확인 응답) 을 주고받음
• 메시지를 저장하는 과정 (publish → broker)에 실패하지 않게 하고 싶다면 qos=1 이상 쓰는게 유리.
• 네트워크가 안정적이면 qos=0 으로 해도 괜찮으나, presence detection(클라이언트의 접속 상태 감지, online/offline) 에서는 실시간성이 중요하고 최신 정보가 broker에 잘 저장되어야 하니까 qos=1 으로 설정
• qos=2는 너무 무겁다