본문 바로가기
Python

mqtt 참고사이트

by YJHTPII 2024. 3. 19.
반응형

https://jeong-f.tistory.com/172

 

[ 파이썬 ] HiveMQ MQTT 통신 테스트

MQTT 통신 M2M, IOT를 위한 프로토콜로서, 최소한의 전력과 패킷량으로 통신하므로, IOT와 모바일 애플리케이션 등의 통신에 매우 적합한 프로토콜 PUBLISH : 발행 SUBSCRIBE : 구독(모니터링) MQTT 브로커

jeong-f.tistory.com

 

 

 

MQTT 통신

M2M, IOT를 위한 프로토콜로서, 최소한의 전력과 패킷량으로 통신하므로,  IOT와 모바일 애플리케이션 등의 통신에 매우 적합한 프로토콜

 

 

  • PUBLISH  : 발행
  • SUBSCRIBE : 구독(모니터링)

 

MQTT 브로커 구동 하기

  • Mosquitto
  • HiveMQ
  • mosca
  • ActiveMQ
  • RabbitMQ

HiveMQ를 이용하여 테스트 브로커를 생성하고 파이썬과 통신

1. HiveMQ 사이트 회원 가입

아래 사이트에서 회원 가입을 먼저 합니다.

 

HiveMQ Cloud

 

console.hivemq.cloud

2.  HiveMQ Websocket Client 페이지 이동 및 연결

비밀 번호 입력 후 [Connect]를 누르면 연결된 경우 Connected 램프가 녹색으로 바뀝니다.

 

 

3. 모니터링을 위한 Subscription 추가

사용할 토픽 : paho/test1/multiple , paho/test2/multiple

 

 

MQTT는 QoS(Quality of Service)를 제공하는데, 총 3단계로 나뉘어 있습니다.

* 0 : 메시지는 한번만 전달되며, 전달이후의 수신과정을 체크하지 않는다.
* 1 : 메시지는 한번 이상 전달되고, 핸드셰이킹 과정을 추적하나, 엄격하게 추적하지 않기 때문에 중복수신의 가능성이 있다.
* 2 : 메세지는 한 번만 전달되고, 핸드 셰이킹의 모든 과정을 체크한다.

QoS의 단계가 높아질수록 통신의 품질은 향상되지만, 성능 저하의 가능성이 있습니다.

 

 

4. 파이썬 코드 작성

아래 부분을 입력하고  실행하면  10초 간격으로 paho/test1/multiple에  "test 1"을  paho/test2/multiple에 현재 시간을 찍어 줍니다.

  • <username> : 사용자 아이디
  • <password> : 비밀번호
  • <HostUrl> : Host 주소
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
''' pip install paho-mqtt
'''
 
 
#
# Copyright 2021 HiveMQ GmbH
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
 
import ssl
from time import sleep
from paho import mqtt
import paho.mqtt.client as paho
import paho.mqtt.publish as publish
from datetime import datetime
 
 
# use TLS for secure connection with HiveMQ Cloud
sslSettings = ssl.SSLContext(mqtt.client.ssl.PROTOCOL_TLS)
 
 
# put in your cluster credentials and hostname
auth = {'username'"<username>"'password'"<password>"}
 
while True:
    
    dt = datetime(20211231133542657813)
    result = dt.strftime("%Y년 %m월 %d일 %H시 %M분 %S.%f초")
    
    # create a set of 2 test messages that will be published at the same time
    msgs = [{'topic'"paho/test1/multiple"'payload'"test 1"}, ("paho/test2/multiple", result, 0, False)]
 
    publish.multiple(msgs, hostname="<host_url>", port=8883, auth=auth,
                 tls=sslSettings, protocol=paho.MQTTv31)
    sleep(10)
cs

 

5. 테스트 결과

 

 

 

참고용( HiveMQ 파이썬 샘플 public + subscribe)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
#
# Copyright 2021 HiveMQ GmbH
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import time
import paho.mqtt.client as paho
from paho import mqtt
 
# setting callbacks for different events to see if it works, print the message etc.
def on_connect(client, userdata, flags, rc, properties=None):
    print("CONNACK received with code %s." % rc)
 
# with this callback you can see if your publish was successful
def on_publish(client, userdata, mid, properties=None):
    print("mid: " + str(mid))
 
# print which topic was subscribed to
def on_subscribe(client, userdata, mid, granted_qos, properties=None):
    print("Subscribed: " + str(mid) + " " + str(granted_qos))
 
# print message, useful for checking if it was successful
def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.qos) + " " + str(msg.payload))
 
# using MQTT version 5 here, for 3.1.1: MQTTv311, 3.1: MQTTv31
# userdata is user defined data of any type, updated by user_data_set()
# client_id is the given name of the client
client = paho.Client(client_id="", userdata=None, protocol=paho.MQTTv5)
client.on_connect = on_connect
 
# enable TLS for secure connection
client.tls_set(tls_version=mqtt.client.ssl.PROTOCOL_TLS)
# set username and password
client.username_pw_set("<username>""<password>")
# connect to HiveMQ Cloud on port 8883 (default for MQTT)
client.connect("<host_url>"8883)
 
# setting callbacks, use separate functions like above for better visibility
client.on_subscribe = on_subscribe
client.on_message = on_message
client.on_publish = on_publish
 
# subscribe to all topics of encyclopedia by using the wildcard "#"
client.subscribe("encyclopedia/#", qos=1)
 
# a single publish, this can also be done in loops, etc.
client.publish("encyclopedia/temperature", payload="hot", qos=1)
 
# loop_forever for simplicity, here you need to stop the loop manually
# you can also use loop_start and loop_stop
client.loop_forever()
cs
 

 

반응형

댓글