MQTT - 1, 기본통신

🔷 1. MQTT 실험 환경 설정

📍 [1-1] Mosquitto 브로커 설정 (mosquitto.conf)

• Mosquitto는 오픈소스 MQTT 브로커
• 설정 파일: config/mosquitto.conf

listener 1883             # 포트 설정
allow_anonymous true      # 인증 없이 사용
persistence true          # 재시작 후에도 메시지 저장
persistence_location /mosquitto/data/
log_dest stdout           # 로그 출력

📍 [1-2] Docker로 브로커 실행 (옵션)

docker run -it -p 1883:1883 -v ./config:/mosquitto/config -v ./data:/mosquitto/data eclipse-mosquitto

---

🔷 1.  mqttPub_from_csv.py 코드 설명 (Publisher)

📌 전체 목적

CSV 파일에서 데이터를 읽어, 일정 간격으로 MQTT 브로커를 통해 메시지를 전송하고, 그 내용을 로컬 로그 파일에도 저장함.

📄 코드 단계별 설명

---

### 1. 라이브러리 및 경로 설정

import time, json, csv
import paho.mqtt.client as mqtt
import os

 

• paho.mqtt.client: MQTT 통신을 위한 파이썬 클라이언트
• json, csv: 메시지 포맷과 데이터 로딩
• os: 경로 확인 및 폴더 생성

---

### 2. 경로 및 설정 변수

CSV_PATH = "/app/industrial_robot_control_6G_network.csv"
LOG_PATH = "/app/pub_log.csv"
MQTT_BROKER = "mqtt-broker"
MQTT_PORT = 1883
MQTT_TOPIC = "test/latency"
PUBLISH_INTERVAL = 0.1​

• CSV_PATH: 발행할 데이터가 담긴 CSV 파일
• LOG_PATH: 발행 로그를 저장할 경로
• MQTT_BROKER: MQTT 브로커 주소 (mqtt-broker는 Docker 네트워크 alias로 사용될 수도 있음. 실제 환경에서는 192.168.x.x 등으로 변경 가능)
• PUBLISH_INTERVAL: 메시지 발행 주기 (초 단위, 예: 0.1초 → 10Hz)

---

### 3. MQTT 클라이언트 초기화 및 연결

client = mqtt.Client()
client.connect(MQTT_BROKER, MQTT_PORT)
client.loop_start()
msg_id = 0

 

• 클라이언트를 생성하고 브로커에 연결함.
• loop_start()는 비동기로 메시지 송수신 처리를 시작함.
• msg_id는 메시지 번호를 관리하는 카운터 (손실률 분석에 쓰임)

---

### 4. MQTT 클라이언트 초기화 및 연결

os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)

• 로그 디렉토리가 없으면 생성 (/app 내부는 Docker 컨테이너 기준)

---

### 5. 로그 파일 열기 & 헤더 쓰기

with open(LOG_PATH, "w", newline='') as logfile:
    log_writer = csv.writer(logfile)
    log_writer.writerow(["id", "timestamp", "payload"])

• pub_log.csv 생성 및 로그용 CSV writer 준비

---

### 6. CSV 파일을 한 줄씩 읽어서 MQTT로 전송

    with open(CSV_PATH, newline='') as csvfile:
        reader = csv.DictReader(csvfile)

        for row in reader:
            now = time.time()
            msg = {
                "id": msg_id,
                "timestamp": now,
                "data": row,
            }
            payload = json.dumps(msg)
            client.publish(MQTT_TOPIC, payload)
            ...
            log_writer.writerow([msg_id, now, payload]) 
            msg_id += 1
            time.sleep(PUBLISH_INTERVAL)

 

• CSV에서 한 줄씩 읽고
• 각 row를 포함한 JSON 메시지를 생성
• 메시지 ID와 timestamp를 포함하여 test/latency 토픽으로 MQTT 발행
• 발행 후 해당 메시지를 로그로 기록
• 지정된 주기(0.1초)로 전송 반복

---

### 7. 실행 종료 로그

print("[mqttPub_from_csv] Finished publishing.")

 

• 전체 발행이 끝났을 때 출력

---

🌐 네트워크 기반 실험에서의 실행 방법

퍼블리셔와 브로커는 같은 네트워크 내에서 통신 가능해야 함.

🖥️ 로컬 머신 1 (브로커 실행):

mosquitto -c config/mosquitto.conf

 

또는 Docker 사용 시:

docker run -it -p 1883:1883 -v $PWD/config:/mosquitto/config eclipse-mosquitto

---

🖥️ 머신 2 (퍼블리셔 실행):

브로커가 실행 중인 컴퓨터의 IP가 예를 들어 192.168.0.42라면:

MQTT_BROKER = "192.168.0.42"

으로 수정 후 실행:

python3 mqttPub_from_csv.py

📌 발표 시 강조할 키포인트

포인트	설명
네트워크 기반 통신	브로커와 퍼블리셔는 서로 다른 장비에서 동작 가능
메시지 구조	id, timestamp, data로 구성 → 정량 분석 가능
전송 간격	실험 목적에 따라 자유롭게 설정 가능
로그 기록	실시간 실험 후 latency, 손실률 분석에 활용
유연한 데이터 크기 조절	payload 크기 실험은 이전 버전 코드에 포함 가능 (generate_fixed_size_payload() 활용)

---

🔷2. mqttSub.py 코드 설명 (Subscriber)

📌 전체 목적

• MQTT 브로커에서 메시지를 수신
• 메시지마다 latency 측정
• 로그 기록 (sub_log.csv)
• 실험 종료 시 손실률 계산 (loss_result.csv)

---

📄 코드 단계별 설명

---

### 1. 라이브러리 및 설정 경로

import time, json, csv
import paho.mqtt.client as mqtt
import os

---

### 2. 파일 경로 및 변수 초기화

PUB_LOG_PATH = "/app/pub_log.csv"
SUB_LOG_PATH = "/app/sub_log.csv"
LOSS_LOG_PATH = "/app/loss_result.csv"

• 퍼블리셔가 기록한 메시지 로그를 읽어 손실률 분석에 활용

count = 0
total_latency = 0
received = 0
received_ids = set()

• 수신한 메시지 개수, ID 목록 등을 추적

---

### 3. 수신 로그 파일 생성 및 헤더 작성

os.makedirs("/app", exist_ok=True)
csv_file = open(SUB_LOG_PATH, "w", newline="")
writer = csv.writer(csv_file)
writer.writerow(["id", "timestamp", "latency_ms"])

• 수신 로그 파일을 /app/sub_log.csv로 생성하고 헤더를 기록

---

### 4. MQTT 수신 처리 함수 정의

def on_message(client, userdata, msg):

 

• 수신 시 호출됨
• payload 디코딩 → latency 계산

latency = (time.time() - raw["timestamp"]) * 1000

• 퍼블리셔가 기록한 timestamp와 현재 시간의 차이로 latency(ms) 계산

writer.writerow([msg_id, time.time(), round(latency, 3)])

 

• sub_log.csv에 저장

received_ids.add(int(msg_id))

• 메시지 ID 저장 → 손실률 계산에 활용됨

---

### 5. 손실률 계산 함수

def calculate_loss():

• 수신이 끝난 뒤 퍼블리셔 로그와 수신 로그를 비교하여
• 전송 개수, 수신 개수, 손실 개수, 손실률을 계산

 

loss_rate = (loss_count / total_sent) * 100

---

### 6. 구독 설정 및 실행

client = mqtt.Client()
client.connect("mqtt-broker", 1883)
client.subscribe("test/latency")
client.on_message = on_message

 

• 퍼블리셔와 동일한 토픽 test/latency에 구독

client.loop_forever()

• 무한 루프 형태로 실행 → 실험 종료 시 Ctrl+C

---

### 7. 종료 후 로그 및 손실률 출력

finally:
    csv_file.close()
    calculate_loss()

---

🌐 네트워크 기반 실행법 (Subscriber)

🖥️ 다른 장비에서 실행할 경우:

1. mqttSub.py에서 브로커 IP 수정:

client.connect("192.168.0.42", 1883)  # 브로커 실행 중인 장비의 IP

   2. 실행:

python3 mqttSub.py

📌 주의: 퍼블리셔가 보내기 시작한 후 실행해도 되지만, 동시 실행이 이상적
로그 저장은 /app/sub_log.csv, 손실률은 /app/loss_result.csv에 저장됨

---

✅ 실험 결과 분석 흐름 정리

파일	내용
pub_log.csv	발행된 메시지 로그 (id, timestamp, payload)
sub_log.csv	수신된 메시지 로그 (id, timestamp, latency_ms)
loss_result.csv	총 발송, 수신, 손실률 정보
pub_cpu.csv, sub_cpu.csv	각각의 CPU 사용률 (실험 부하 측정 목적)

---

🔬 분석 예시 (calculate_loss.py 결과 예시)

전송 수: 1000, 수신 수: 995, 손실 수: 5, 손실률: 0.50%

 

loss_result.csv 예시:

total_sent	total_received	loss_count	loss_rate(%)
1000	995	5	0.5

---

🎯 발표 시 강조 포인트 정리

구간	강조 내용
Publisher	payload 크기, 전송 주기, 로그 기반 실험 분석 가능
Subscriber	latency, 실시간 수신, 손실률 자동 계산
분석	CSV 기반으로 객관적인 지표 도출 가능
확장성	네트워크 기반 실험, 다양한 장비/환경에서 실험 가능

pub_cpu_logger.py / sub_cpu_logger.py 설명 (CPU 로깅용)

---

📌 공통 목적

• psutil 라이브러리를 사용해 CPU 사용률을 1초 단위로 측정
• CSV로 저장하여 시각화/비교 분석에 활용

1️⃣ pub_cpu_logger.py (퍼블리셔 측 CPU 로거)

import time
import csv
import psutil
import os

print("[pub-cpu] Starting CPU logging...")

log_path = "/app/pub_cpu.csv"
with open(log_path, "w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["timestamp", "cpu_percent"])
    try:
        while True:
            cpu = psutil.cpu_percent(interval=1)  # 1초마다 CPU 측정
            timestamp = time.time()
            writer.writerow([timestamp, cpu])
            print(f"[pub-cpu] {timestamp:.2f}, {cpu:.2f}%")
            f.flush()
    except KeyboardInterrupt:
        print("[pub-cpu] Logging stopped.")

---

2️⃣ sub_cpu_logger.py (서브스크라이버 측 CPU 로거)

import psutil
import time
import csv

log_path = "/app/sub_cpu.csv"
with open(log_path, mode="w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["timestamp", "cpu_percent"])

    while True:
        cpu = psutil.cpu_percent(interval=1)
        writer.writerow([time.time(), cpu])
        f.flush()

차이점: sub_cpu_logger.py는 간단히 구성되어 있지만 기능은 동일

---

📁 결과 파일 구조

파일명	내용
pub_cpu.csv	퍼블리셔 측 시간대별 CPU 사용률
sub_cpu.csv	서브스크라이버 측 시간대별 CPU 사용률

timestamp,cpu_percent
1712320764.89,4.3
1712320765.89,5.1
...

---

💡 실행 방법 (서로 다른 터미널에서 병렬 실행)

퍼블리셔 측:

python3 pub_cpu_logger.py

서브스크라이버 측:

python3 sub_cpu_logger.py

실험이 끝났으면 Ctrl+C로 종료 가능
(데이터는 실시간으로 저장되며, flush 사용됨)

---