MQTT로 센서 데이터를 보낼 때 QoS를 1로 설정하면 통신 품질 문제를 모두 해결한 것처럼 느끼기 쉽습니다. 하지만 연결이 끊겼다가 복구되는 환경에서는 같은 측정값이 여러 번 도착할 수 있습니다. 서버가 수신할 때마다 누적 카운터를 올리면 실제 측정 횟수와 저장 결과가 달라집니다.
이 글은 MQTT 메시지의 전달과 애플리케이션의 저장 완료를 구분하고, 중복 수신에도 같은 결과를 만드는 설계를 설명합니다. 코드는 실제 브로커에 접속하지 않는 Python 3·SQLite 학습 예제입니다.
QoS가 책임지는 구간을 구분하기
QoS 0은 최대 한 번, QoS 1은 최소 한 번, QoS 2는 프로토콜 수준에서 정확히 한 번 전달하는 방식입니다. QoS 1에서는 중복이 허용됩니다. 장치에서 브로커로 보내는 구간과 브로커에서 구독자에게 보내는 구간의 QoS도 각각 살펴봐야 합니다. OASIS MQTT 5.0 표준, 4.3절
장치가 브로커의 PUBACK을 받았다는 사실만으로 구독 서버의 DB 트랜잭션까지 완료되었다고 볼 수는 없습니다. 구독 프로그램이 메시지를 받은 직후 종료되거나, 저장은 성공했지만 애플리케이션 응답을 보내기 전에 끊길 수 있기 때문입니다.
따라서 요구사항을 먼저 문장으로 적습니다. “최신 온도가 보이면 된다”와 “모든 생산 이벤트가 중복 없이 기록되어야 한다”는 서로 다른 저장·재시도 정책을 필요로 합니다.
재전송에도 유지되는 Event ID 설계
아래는 설명용 측정 메시지입니다. 실제 장치의 표준 형식은 아닙니다.
{
"device_id": "sensor-01",
"boot_id": "session-a",
"seq": 17,
"temperature_c": 25.1
}
이 예제에서는 (device_id, boot_id, seq)를 한 측정의 식별자로 사용합니다. 같은 측정값을 재전송할 때 세 항목을 유지하고, 새로운 측정이 생겼을 때만 번호를 증가시킵니다.
boot_id는 재부팅 후 seq가 다시 0부터 시작하는 경우를 구분합니다. 실제 제품에서는 충돌 가능성이 충분히 낮은 부팅 식별자나 영속 카운터 등으로 생성해야 합니다. 예제의 session-a를 모든 장치와 부팅에 그대로 사용하면 안 됩니다. MQTT Packet Identifier는 재사용될 수 있으므로 장기 보관용 이벤트 ID로 대신 쓰지 않습니다.
같은 ID인데 내용이 달라지면 단순 중복으로 지우지 말고 생산자 오류로 기록해야 합니다. ID를 만든 시점의 측정 내용도 재전송 동안 유지해야 하는 이유입니다.
SQLite로 중복 저장과 충돌 확인하기
다음 코드를 mqtt_event_demo.py로 저장하고 python mqtt_event_demo.py로 실행합니다. 메모리 DB만 사용하며 디스크·브로커·실제 장치를 변경하지 않습니다.
import sqlite3
db = sqlite3.connect(":memory:")
db.execute("""
CREATE TABLE readings (
device_id TEXT NOT NULL,
boot_id TEXT NOT NULL,
seq INTEGER NOT NULL,
temperature_c REAL NOT NULL,
PRIMARY KEY (device_id, boot_id, seq)
)
""")
def store_reading(device_id, boot_id, seq, temperature_c):
with db:
cursor = db.execute("""
INSERT INTO readings VALUES (?, ?, ?, ?)
ON CONFLICT(device_id, boot_id, seq) DO NOTHING
""", (device_id, boot_id, seq, temperature_c))
if cursor.rowcount == 1:
return "stored"
saved = db.execute("""
SELECT temperature_c FROM readings
WHERE device_id=? AND boot_id=? AND seq=?
""", (device_id, boot_id, seq)).fetchone()[0]
if saved != temperature_c:
raise ValueError("same event ID with different data")
return "duplicate"
print(store_reading("sensor-01", "session-a", 17, 25.1))
print(store_reading("sensor-01", "session-a", 17, 25.1))
print(store_reading("sensor-01", "session-b", 17, 25.2))
print(db.execute("SELECT COUNT(*) FROM readings").fetchone()[0])
예상 결과는 stored, duplicate, stored, 2입니다. 장치가 재부팅해도 새 부팅 식별자를 사용하면 이전 번호와 충돌하지 않습니다. SQLite의 충돌 처리 동작은 공식 UPSERT 문서에서 확인할 수 있습니다.
이 코드는 하나의 연결로 순차 실행하는 작은 예제입니다. 운영 서버에는 스키마·범위 검사, 실제 영속 저장소, 동시 접속 처리와 보관 정책이 추가로 필요합니다. 예제의 실수 비교 역시 단순 비교이며, 실제로는 정규화한 전체 메시지나 내용 해시로 충돌을 확인할 수 있습니다.
상태 토픽과 명령 토픽의 수명 정하기
Retained 메시지는 토픽의 최근 상태를 새 구독자에게 전달할 때 유용합니다. 반면 “모터를 한 번 움직여라” 같은 일회성 명령을 retained로 남기면 재접속한 장치가 오래된 명령을 받을 수 있습니다. 상태 조회와 일회성 동작을 별도 토픽으로 나누고, 명령에는 ID와 유효 시간을 함께 설계하는 편이 검토하기 쉽습니다.
MQTT 5의 Message Expiry Interval은 메시지 수명을 지정합니다. Session Expiry Interval은 연결 종료 후 세션 상태를 유지하는 기간이므로 서로 다른 설정입니다. MQTT 3.1.1 장치가 섞여 있다면 5.0 속성에만 의존하지 말고 수신 애플리케이션의 시각·만료 검사도 설계해야 합니다. OASIS MQTT 5.0 표준, 3.1.2.11.2·3.3.1.3·3.3.2.3.3절
장치 시계가 동기화되지 않았을 때 만료 시각을 어떻게 판단할지도 정합니다. 측정 시각과 서버 수신 시각을 나누어 저장하면 지연된 데이터와 현재 상태를 구분하는 데 도움이 됩니다.
연결을 일부러 끊으며 확인할 항목
| 시험 상황 | 확인할 결과 |
|---|---|
| 같은 ID·같은 내용을 세 번 전송 | 저장 결과는 한 개 |
| 같은 ID·다른 내용을 전송 | 조용히 덮어쓰지 않고 충돌 기록 |
| 새 boot_id로 seq를 다시 사용 | 새 측정으로 저장 |
| 저장 직후 수신 프로그램 재시작 | 영속 저장된 ID로 재처리 방지 |
| 오래된 명령이 늦게 도착 | 정의한 유효 시간 정책 적용 |
DB 저장과 물리 장치 동작은 하나의 DB 트랜잭션으로 묶이지 않습니다. 밸브를 움직인 직후 전원이 꺼진 상황까지 다루려면 장치 측 명령 ID 저장, 실행 상태 조회, 재시도 시 응답 규칙이 별도로 필요합니다. 위 저장 예제만으로 물리 동작의 정확히 한 번 실행을 보장하지 않습니다.
브로커 연결과 구독·발행은 자료실의 MQTTX 사용법으로 먼저 확인할 수 있습니다. 실제 프로젝트에서는 중복 수신 횟수, ID 충돌, 저장 실패와 마지막 정상 수신 시각을 함께 관찰하세요.
작성·문서 확인: 2026년 9월 13일. Python 예제의 저장·중복·충돌 동작을 확인했으며 실제 장치의 통신 시험 결과를 의미하지 않습니다.