산학 6단계 · 생성기 MQTT 송신

목표

생성기가 TCP 말고 MQTT로도 보낼 수 있게 한다. 같은 데이터를 두 방식으로 보내서 지연·유실·재연결을 비교하기 위해서다. (10.28)

A. TCP와 MQTT 차이

TCP  : 생성기 ───────────────▶ 수신부           (직접 연결)
MQTT : 생성기 ──▶ 브로커 ──▶ 수신부              (중간에 브로커)

• 토픽 : 메시지에 붙이는 주소. 환자마다 ecg/0, ecg/1 … 받는 쪽은 ecg/+ 하나로 전부 구독 (+ = 그 자리 아무거나)
• QoS 0 : 한 번 보내고 끝 (TCP 지금 방식과 비슷) / QoS 1 : 브로커가 받았다고 답할 때까지 다시 보냄

B. 코드 구조 (server/stream/stream_tcp.py)

class TcpSender:
    def send(self, msgs):
        lines = [json.dumps(m) + "\n" for m in msgs]
        self.sock.sendall("".join(lines).encode())

class MqttSender:
    def send(self, msgs):
        for m in msgs:
            info = self.client.publish(f"{self.topic}/{m['channel_id']}", json.dumps(m), qos=self.qos)
            if info.rc != 0:
                self.failed += 1

• 보내는 부분만 두 클래스로 나눔. 메시지를 만드는 코드는 그대로 → 두 방식이 같은 내용을 보내야 비교가 공정함
• TcpSender : 0.1초마다 환자 전원 메시지를 줄바꿈으로 이어서 한 번에 보냄 (기존 방식)
• MqttSender : 환자마다 자기 토픽으로 하나씩 publish. 줄바꿈은 안 붙임
• info.rc != 0 : 브로커 연결 끊김·큐 가득 참이면 실패로 셈
• 옵션 : --transport tcp|mqtt(기본 tcp, 기존 스크립트 그대로), --topic, --qos, 포트 기본값 TCP 8000 / MQTT 1883
• 확인용 구독기 mqtt_check.py : 받은 건수, seq 빈칸(유실), 지연 p50·p95 출력

C. 결과

• 로컬에서 지연이 높았던 건 테스트용 파이썬 브로커가 느려서. 실제 브로커(mosquitto)에서는 1ms대
• 다음 : 수신부(MQTT 받는 부분)까지 붙여서 20·50·100명으로 TCP와 비교

트러블슈팅

MQTT 테스트에서 1,000건 중 900건만 받음

• 증상 : 받음 900건, 그런데 seq 빈칸(유실)은 0
• 원인 : 구독기를 20초만 켜뒀는데 생성기가 처음 실행이라 PhysioNet 기록을 받느라 앞에서 약 11초를 씀 → 마지막 1초(100건)가 오기 전에 구독기가 먼저 꺼짐
• 확인 : 유실이면 중간에 seq 빈칸이 생겨야 하는데 0 → 끝부분이 잘린 것
• 해결 : 기록이 캐시된 상태에서 다시 실행 → 1,000건 전부

남은 작업

• 생성기 MQTT 송신 + 서버에서 송신 확인
• MQTT 수신부까지 붙여 TCP와 비교 (20·50·100명, 지연·유실·재연결)