2025-10-20(파이썬 (스레드, pub,sub통합), 자바 (pub,sub 통합))

목적토픽 예시설명

차량 감지	building1/parking/slot1/status	상태 메시지: occupied, empty
거리 측정	building1/parking/slot1/distance	값 메시지: cm 단위 거리
LED 제어	building1/parking/slot1/led	메시지: green, red, off
LCD 표시	building1/parking/display	메시지: "남은 자리: 2"

 

파이썬에서 쓰레드를 작성하는 방법 

•  싱글 쓰레드

→ 모든 코드의 실행이 끝나야 주 쓰래드가 종료된다. 주 쓰래드가 종료되어야 프로세스가 종료된다.

import time

import string

#싱글 쓰래드 - 메인쓰레드가 종료되면 프로세스가 종료된다.

class MyThread:

    def __init__(self,name):

        self.name = name

       

    def thread_run(self):

        for i in range(10):

            print(str(i)+":" + self.name)

            time.sleep(0.5)

           

           

if __name__ == '__main__':

    print("주 쓰레드(main) 시작")

    t1 = MyThread("t1")

    t2 = MyThread("t2")

   

   

    t1.thread_run()

    t2.thread.run()

   

    print(list(string.ascii_uppercase))

   

•   멀티쓰레드

→  여러 개의 쓰레드가 동시에 실행 되며
주 쓰레드가 먼저 종료될 수 있다. 실행 중인 쓰레드가 하나라도 있으면 프로세스는 종료되지 않는다.

 

---

파이썬에서 쓰레드를 사용하는 방법

파이썬은 threading  모듈을 통해 멀티쓰레드를 지원

쓰레드를 작성하는 방법 두 가지

방법 1. Thread 클래스를 직접 생성해서 사용

import threading

def print_numbers():
    for i in range(1, 6):
        print(i)

t1 = threading.Thread(target=print_numbers)
t1.start()

단순히 함수를 독립 실행시키고 싶을 때 간단히 사용.

방법 2. Thread 클래스를 상속해서 사용 (클래스 기반)

가장 많이 쓰이는 구조
직접 쓰레드 클래스를 만들어서 실행 코드를 run() 안에 넣음

작성순서

1. threading.Thread 클래스를 상속해서 새로운 클래스 정의
2. 생성자에서 Thread의 생성자를 직접 호출 (Thread.__init__(self))
   → 파이썬은 부모 생성자가 자동 호출되지 않기 때문
3. run() 메서드 재정의
   → 쓰레드가 독립적으로 수행할 코드 작성
4. start() 호출
   → 내부적으로 새로운 쓰레드를 만들고 run() 실행

AlphaThread (대문자 출력)

import threading
import time

class AlphaThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)   # 부모 생성자 호출

    def run(self):
        for ch in range(65, 91):          # A(65) ~ Z(90)
            print(chr(ch), end=' ')
            time.sleep(0.1)

DigitThread (숫자 출력)

class DigitThread(threadin.Thread):
	def __init__(self):
    	threading.Thread.__init__(self)
        
    def run(self):
    	for i in range(1, 31):
        	print(i, end='   ')
            time.sleep(0.1)

실행부 (main)

if __name__ == "__main__":
    alpha = AlphaThread()
    digit = DigitThread()

    alpha.start()  # run()이 아닌 start()로 실행
    digit.start()

    alpha.join()   # 쓰레드 종료까지 대기 (선택)
    digit.join()

    print("\n모든 쓰레드 종료")

구분설명

모듈	threading
필수 메서드	__init__(), run(), start(), join()
핵심 포인트	start()가 run()을 자동 실행시킴
활용 예시	센서 제어, 네트워크 수신 대기, MQTT 메시지 처리 등

 

package basic;

import java.util.Scanner;
import java.util.UUID;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;

//자바쪽의 pub와 sub를 처리하는 클래스
public class MqttClientTest {
   private MqttClient mqttClient;
   private MqttConnectOptions mqttOptions;
   private final String broker = "tcp://192.168.14.19:1883";
   private final String pubTopic = "home/livingroom/light";
   private final String subTopic = "home/#";
   public MqttClientTest() {
      mqttOptions = new MqttConnectOptions();
      mqttOptions.setCleanSession(true);// 세션클리어 - 연결하면 구독정보 초기화
      mqttOptions.setKeepAliveInterval(60);// 브로커와 주기적 ping간격(초단위)

      // broker와 연결
      try {
         mqttClient = new MqttClient(broker, "client_"+UUID.randomUUID().toString());
         mqttClient.setCallback(new MqttCallback() {
            
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
               System.out.println("===============메시지도착================");
               System.out.println(message);//toString메소드 내부에서 페이로드를 출력하도록 재정
               System.out.println("topic:"+topic);
               System.out.println("id:"+message.getId());
               System.out.println("payload:"+new String(message.getPayload()));
               System.out.println("=========================================");
               System.out.print("원하는 작업 선택>>>>>>");
            }
            
            @Override
            public void deliveryComplete(IMqttDeliveryToken arg0) {
               System.out.println("deliveryComplete");
               
            }
            
            @Override
            public void connectionLost(Throwable arg0) {
               System.out.println("connectionLost");
               
            }
         });
         // MqttClient가 브로커에 접속하기 전에 브로커에 연결하기 위한 정보가 셋팅된 MqttConnectOptions객체를 셋팅
         mqttClient.connect(mqttOptions);
         System.out.println("연결");
         //객체생성하며 서버에 접속하고 구독신청까지 처리
         subscribe();
      } catch (MqttException e) {
         e.printStackTrace();
      }
   }
   //구독신청
   public void subscribe() {
      try {
         if (subTopic != null) {
            mqttClient.subscribe(subTopic, 0);
         }
      } catch (MqttException e) {
         // TODO Auto-generated catch block
         e.printStackTrace();
      }
   }
   //메시지보내기
   public void send(String topic, String msg) {
      MqttMessage message = new MqttMessage();
      message.setPayload(msg.getBytes());//바이트단위로 데이터를 전송할 수 있도록 변환해서 
      try {
         mqttClient.publish(topic, message);
      } catch (MqttPersistenceException e) {
         // TODO Auto-generated catch block
         e.printStackTrace();
      } catch (MqttException e) {
         // TODO Auto-generated catch block
         e.printStackTrace();
      }
         
   }
   //연결종료
   public void close() {
      if(mqttClient!=null) {
         try {
            mqttClient.disconnect();
            mqttClient.close();//리소스반납
         } catch (MqttException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
         }//연결종료
         
      }
   }
   public static void main(String[] args) {
      MqttClientTest mqttclient = new MqttClientTest();
      Scanner key = new Scanner(System.in);
      
      while(true) {
         System.out.println("1. LED켜기");
         System.out.println("2. LED켜기");
         System.out.println("3. 1번 LED끄기");
         System.out.println("4. 2번 LED끄기");
         System.out.println("5. 서보모터 각도 조절하기(+10)");
         System.out.println("6. 서보모터 각도 조절하기(-10)");
         System.out.print("원하는 작업 선택>>>>>>");
         int val = key.nextInt();
         String msg = "";
         if(val==1) {
            msg = "led:1:led_on";
         }else if(val==2) {
            msg = "led:2:led_on";
         }else if(val==3){
            msg = "led:1:led_off";
         }else if(val==4){
            msg = "led:2:led_off";
         }else if(val==5){
            msg = "plus"; //or 토픽으로 구분 - 편하게 작업
         }else if(val==6){
            msg = "minus";
         }else {
            break;
         }
         mqttclient.send(mqttclient.pubTopic, msg);   
         try {
            Thread.sleep(3000);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         
      }
      mqttclient.close();
   }

}

이 코드는 자바에서 MQTT(Paho 라이브러리) 를 이용해서
Publish(발행) 과 Subscribe(구독) 를 한 번에 처리하는 통합 클라이언트.
하나의 프로그램이 동시에 “명령 보내기(Pub)”와 “센서값 받기(Sub)” 역할을 하는 구조.

[Java 프로그램]
 ├── MQTT Broker (예: tcp://192.168.14.19:1883)
 │
 ├── Publish : home/livingroom/light   ← 제어 명령 송신
 │      예) led:1:led_on, plus, minus
 │
 └── Subscribe : home/#   ← 센서나 장치 상태 수신
        예) home/temperature, home/alert 등

MqttClient 객체

MQTT 브로커와 실제 네트워크 통신을 담당하는 핵심 클래스

MqttConnectOptions

연결 유지(KeepAlive)와 세션 초기화를 설정해서 안정적인 연결 유지

UUID.randomUUID()

각 실행마다 고유한 clientID를 주기 위해 (중복 방지)

 

mqttClient.setCallback()

비동기적으로 메시지 수신을 처리하기 위한 이벤트 핸들러 등록

messageArrived()

라즈베리파이 → Java 서버 방향 메시지 수신 시 자동 실행되는 콜백

send()

Java → 라즈베리파이 방향으로 명령 전달 (publish 기능)

subscribe()

home/# 토픽 전체 구독으로 여러 센서데이터 한 번에 수신 가능

close()

종료 시 MQTT 리소스를 안전하게 정리

정리 요약

LED 제어	home/livingroom/light	Java → Raspberry Pi	led:1:led_on
센서 데이터	home/temperature 등	Raspberry Pi → Java	temp:26.5
경보 알림	home/alert	Raspberry Pi → Java	alert:water_detected
서보모터 제어	home/livingroom/light	Java → Raspberry Pi	plus / minus