之前写过EMQX在阿里云的云服务器上搭建流程，近期云服务器到期了，而且由于没有业务支撑短期也不打算再开云服务器了，但是物联网还是要用的，于是开了这个阿里云物联网的记录。
这个是比较详细的已有的博客记录
这个博客居然还有配套视频课程，如果有时间可以配套看看其实也能搞定
网上有不少已经成型的贴子了，其实没有必要自己再开贴记录，但是网上的记录不是自己的。
下面：
1、阿里云物联网平台是可以白嫖的。但是消息队列 RocketMQ 版是收费的，开始我不知道都开通了，结果消息队列一天扣我3块钱。。。
2、阿里云物联网这个玩意要是成功过会觉得非常简单，但是琢磨的时候一堆资料云里雾里的很是难搞。
3、开通物联网华东2的免费实例后首先是构建产品，说是产品其实就是给这些联网的设备啊整体起了个名字，比如水泵联动控制系统，或者LED-01s啥玩意的。
4、然后在这个产品下面添加俩个设备，一个是电机主体，一个是温度。
5、设备添加后查看进去主要是看它提供的MQTT连接要素
6、为了稳妥用EMQX提供的桌面客户端验证下是不是正常工作,吧内容都弄进去
7、到目前MQTT搭建的设备已经可以通过自带的发布订阅和阿里云服务器实现消息收发了，自带的这几个自定义直接就能用，这里多说下，订阅和发布限定的是非常严格的，订阅GET接口是可以接受到服务器消息的，发送给UPDATA接口是可以在服务器日志里查询发出内容的。
。8、消息流转，这个内容相对来说比较繁琐，很多搞设备的不适应这种文档，通俗讲就是把一个接口透传给另一个接口配置一步一步来。解析器。。。这其实就是给透传这个工作起个名，顺便注释下具体工作内容
9、下一步就是数据源-数据目的-数据规则三联工作了，首先是数据源就是问数据从哪来，记得要先创建数据源再选择，主要还是给数据源起个名。
起好名字点查看：
这里选提供数据的设备以及上报接口,updata
10、三联第二项数据目的。这里也是先创建数据目的，选择发布到另一个TOPIC这项。
11、解析器脚本，这里简单搞一搞就行
里边填上下面内容：
var data = payload("json");writeIotTopic(规格ID号,"/产品名字/设备名字/user/get",data)比如我这次配置，规格ID就是1002, 产品名字要替换填上、设备名字替换自己的设备名字，指定GET接口这样就会把消息送到目标设备的接受口，data表示数据标识。
这样一个基本的数据规则就算做完了。这要填的对，给哪个设备甚至跨产品以及自己给自己都能发信息。当然要做透传目前只完成了一个单向，还得反着做一次。
目前透传就做好了，我们看下效果
订阅都是
/产品名/设备名/user/get
发送都是
/产品名/设备名/user/update
下面首先是PC上线，电脑端我贴的程序非常简单,这第一个程序是订阅阻塞的，和我之前写的程序唯一区别是多了一个CID位置，看注释就可以。
import paho.mqtt.client as mqttdef on_connect(client, userdata, flags, rc): print("Connected with result code: " + str(rc))def on_message(client, userdata, msg): print(msg.topic + " " + str(msg.payload))ip = 'a1GKughxxxxxxxxxxxi.aliyuncs.com'# 连接IPname = 'numbKxxxxxxxxxxxxxxxPkbz'# 用户名password='f325c2dcc7xxxxxxxxxxxxxxx3eaa9a33b82a85c153' #密码cid='a1GKuD5xxxxxxxxxxxxxxxxxx9517126676|' #CIDtoc='a1GKxxxxxxxxxxxxxxxxxxget' #订阅的主题client = mqtt.Client(cid) #cid扔这里client.on_connect = on_connectclient.on_message = on_messageclient.username_pw_set(name, password=password) #这里也是：账号密码扔这里client.connect(ip, 1883, 60) # 60为keepalive的时间间隔 心跳包活client.subscribe(toc, qos=0)#前边是主题 # 订阅的主题client.loop_forever() # 阻塞并保持连接 然后发送的这样的
import paho.mqtt.client as mqttdef on_connect(client, userdata, flags, rc): print("Connected with result code: " + str(rc))def on_message(client, userdata, msg): print(msg.topic + " " + str(msg.payload))ip = 'a1qoB5i3jxb.iot-xxxshanghai.aliyuncs.com'name = 'machxxxxxi3jxb'password='9954f7bfbfxxxx90f19a14b7937abc3'cid='a1qoB5i3jxbxxxxxxx,timestamp=1669537156063|'toc='/a1qoB5ixxxxxupdate'client = mqtt.Client(cid)client.on_connect = on_connectclient.on_message = on_messageclient.username_pw_set(name, password=password) #需要验证账号密码就带上这句，准许匿名就不带这句client.connect(ip, 1883, 60) # 600为keepalive的时间间隔，也就是发布周期，一般来说一次连接保证在发布周期内再次发送即可确保不掉线。client.publish(toc, payload="{'data':'33sd333'}", qos=0)#前边是主题 后边是消息关于连接超时 loop()，这个有点像心跳包也有点像喂狗函数反正就是规定时间内要么发数据或者执行下loop()维持心跳，如果用阻塞可以自己维持心跳，如果不阻塞就得自己跳。。。
client.connect('127.0.0.1', 1883, 5) # keeplive仅为5秒for i in range(100): client.publish('fifa', payload=f'amazing{i}', qos=0) # client.loop() # 或者loop() time.sleep(4) # 不能超过5秒对于订阅客户端，一种方法是使用loop_start()保持连接，然后写个死循环阻塞程序，保持监听。
client.connect('127.0.0.1', 1883, 5)client.subscribe('fifa', qos=0)client.loop_start()while True: pass当然像之前的写法直接阻塞也是可以的.
tips闭坑指南：在云服务器上运行订阅端不能在死循环里写个PASS就完事，他会耗尽CPU的，要适当的睡眠让服务器调度资源降低CPU占用率。
这里有我之前写过的自建MQTT的连接例程，传送门在此
ESP32的直接粘这里备用吧，验证可用，懒得整了就放这得了
#-*-coding:utf-8-*-from umqtt.simple import MQTTClientfrom machine import Pinimport networkimport timeimport machineimport dhtfrom machine import Timerimport json#---以下的参数值都需要根据自己的环境修改-----------------------------------------------led=Pin(2,Pin.OUT) #ESP32的引脚2接了LED灯，可根据自己的ESP32板子的LED引脚来设置SSID = "Ggc0622" #填写自己的WIFI名称PASSWORD = "gaoke0622" #填写自己的WIFI密码SERVER = 'ia1xxxxxxxxxxxxxxyuncs.com' # mqttHostUrlCLIENT_ID = "a1qoB5ixxxxxxmestamp=1x52747772|" # clientIdusername = 'TMPSxxxx3jxb' #usernamepassword = '8962272c6e3xxxxxx92d1fa55' #密码publish_TOPIC = '/a1qoxxxxxpdate'subscribe_TOPIC = '/a1qxxxxxxr/get'#---以上的参数值都需要根据自己的环境修改-----------------------------------------------client = Nonemydht = Nonewlan = Nonedef ConnectWifi(ssid, passwd): global wlan wlan = network.WLAN(network.STA_IF) # create a wlan object wlan.active(True) # Activate the network interface wlan.disconnect() # Disconnect the last connected WiFi wlan.connect(ssid, passwd) # connect wifi while (wlan.ifconfig()[0] == '0.0.0.0'): time.sleep(1) print(wlan.ifconfig())def sub_cb(topic, msg): global led print((topic, msg)) #msg = str(msg) print(type(msg)) print(msg) msg = json.loads(msg) print(msg) if msg['lightStatus'] =='ON': print('receive ON') led.value(1) print('led ON') if msg['lightStatus'] =='OFF': print('receive OFF') led.value(0) print('led OFF')def heartbeatTimer(mytimer): global client global led led.value(1) try: mymessage = '{"heartbeat":"Device1"}' print('============================') print(mymessage) client.publish(topic=publish_TOPIC, msg=mymessage, retain=False, qos=0) except Exception as ex_results2: print('exception', ex_results2) print('this is error') mytimer.deinit()# finally:# machine.reset()def run(): global client global led global wlan print('start to connect mqtt ali') try: mydht = 15 #么啥用，人家的例程 懒得删 ConnectWifi(SSID, PASSWORD) client = MQTTClient(CLIENT_ID, SERVER, 0, username, password, 60) # create a mqtt client print('client:%s' % str(client)) led.value(1) client.set_callback(sub_cb) ########## 设置回调函数 client.connect() # 连接 client.subscribe(subscribe_TOPIC) # client subscribes to a topic mytimer = Timer(0) mytimer.init(mode=Timer.PERIODIC, period=50000, callback=heartbeatTimer) #心跳50秒维持发送'{"heartbeat":"Device1"}' while True: client.wait_msg() # wait message except Exception as ex_results: print('exception1', ex_results) finally: if (client is not None): led.value(0) client.disconnect() wlan.disconnect() wlan.active(False) return 'FAILED'#-*-coding:utf-8-*-import timewhile (True): if run() == 'FAILED': print('FAILED,retry to connect') time.sleep(5)下面这个之前我自己写的也能用，。。。都堆这吧，测试可用，思路不一样，不开线程也不阻塞系统
from umqtt.simple import MQTTClientimport timeimport timeimport networkwl=network.WLAN(network.STA_IF)wl.active(True)def lan_on_off(): if wl.isconnected()==0: while wl.isconnected()==0 : wl.connect('Ggc0622','gaoke0622') time.sleep(10)lan_on_off()print('联网成功')class Mqtt_run(): def __init__(self,CLIENT_ID,SERVER,username,password,list_sub):# 设备名 ， 服务器地址，端口 ， 账号， 密码,订阅列表 self.mqtt_mast = MQTTClient(CLIENT_ID, SERVER, 0, username, password, 60) # create a mqtt client while 1: try: self.mqtt_mast.connect() except: continue break self.mqtt_mast.set_callback(self.recdate1)# 绑定回调函数，名字别错 self.mqtt_mast.connect() # 连接 print(list_sub) self.mqtt_mast.subscribe(list_sub)#设置订阅的主体， def recdate1(self,t,m):#这是回调函数，有信息并触发后都在这里执行 ###############我就是填充业务逻辑的地方############### print("我在这里运行",t,m) ##################################################ip = 'ia1xxxxxaliyuncs.com'name = 'TMxxxxi3jxb'password='89622xxxx6cde0f33680fd9092d1fa55'cid='a1qoB5i3xxxxxx56,tixx669552747772|'toc='/a1qoB5xxxx/get'poc='/a1qoBxxxser/update'if __name__=='__main__': #from mqtt1 import Mqtt_run a=Mqtt_run(cid,ip,name,password,toc) #设备名 ， 服务器地址，端口 ， 账号， 密码,订阅列表 while 1: a.mqtt_mast.check_msg() #轮询消息，主函数中周期越快越好,没这个就听不叫了 a.mqtt_mast.publish(poc,'{"data":"uuuu"}') # *****前边是发往哪个主题，后面是内容 发送数据***************************** time.sleep(5) #延时，别刷屏 print('5')