开源物联网设备与平台对接的消息中间件，实测支持大规模并发，基于Python + TCP+ MQTT

#-*- coding:utf-8 -*- __author__ ='huangzhi'#line:2 try :#line:7 import
SocketServer #line:8 except ImportError :#line:9 import socketserver #line:10
try :#line:11 from SocketServer import ThreadingTCPServer ,StreamRequestHandler
,TCPServer #line:12 except ImportError :#line:13 from socketserver import
ThreadingTCPServer ,StreamRequestHandler ,TCPServer #line:14 import uuid
#line:15 import paho.mqtt.client as mqtt #line:16 from multiprocessing import
Process #line:17 from threading import Thread #line:18 import traceback
#line:19 import socket #line:20 import sys #line:21 import json #line:22 import
urllib #line:23 import urllib2 #line:24 try :#line:25 import httplib #line:26
except ImportError :#line:27 pass #line:28 import time #line:29 class myPost
:#line:31 sc_url =""#line:32 gws_url =""#line:33 shbd_url =""#line:34 def
__init__ (OO0OO0OO00OO0O0O0 ,OO0O00O0O00OOOOOO ,O0O00OO000OO00O0O
,OOO000OO0OO0OOO00 ):#line:36 OO0OO0OO00OO0O0O0 .sc_url =OO0O00O0O00OOOOOO
#line:37 OO0OO0OO00OO0O0O0 .gws_url =O0O00OO000OO00O0O #line:38
OO0OO0OO00OO0O0O0 .shbd_url =OOO000OO0OO0OOO00 #line:39 pass #line:40 def
SaveChannelData (O00OOOO000OOOOOO0 ,O0OOOOO00OO0000O0 ,OOO0O0OO0000O0000
,OO0000OO00O00OOOO ):#line:42 OO0O00O0OOO00000O =""#line:43 try :#line:44
OO0O0OOO0O0000000 =json .loads (OOO0O0OO0000O0000 )#line:45 except :#line:46
return #line:47 print ("提交网关设备数据到%s" % O00OOOO000OOOOOO0.sc_url ) #line:48
print (json .dumps (OO0O0OOO0O0000000 ))#line:49 for OO0OO0O0OO0OO0O00 in
OO0O0OOO0O0000000 ["data"]:#line:51 OO0O00O0OOO00000O =OO0OO0O0OO0OO0O00
["ndid"]#line:53 OOO00000O00O00000 =json .dumps (OO0OO0O0OO0OO0O00
["channel"])#line:54 print (json .dumps (OO0OO0O0OO0OO0O00
["channel"]))#line:55 OO0O00O0OO0O0O000 ={"f0003":O0OOOOO00OO0000O0
,"f0004":OO0O00O0OOO00000O ,"f0005":"","f0006":OOO00000O00O00000
,"f0009":OO0000OO00O00OOOO }#line:56 OO0O00O0OO0O0O000 =urllib .urlencode
(OO0O00O0OO0O0O000 )#line:57 print (OO0O00O0OO0O0O000 )#line:58
O0O0O0O0O0O0OOOO0 =urllib2 .Request (O00OOOO000OOOOOO0 .sc_url
,OO0O00O0OO0O0O000 )#line:59 try :#line:61 urllib2 .urlopen (O0O0O0O0O0O0OOOO0
,timeout =2 )#line:62 except :#line:63 traceback .print_exc ()#line:64 return
#line:65 def GetwayStatusData (OOOO000OOOO00O0O0 ,O00O00OO00O000O00
,O0OO0O00OOO0OOO00 ,O0O0OOO0OO00O00O0 ):#line:79 print
("提交网关状态到%s"%OOOO000OOOO00O0O0 .gws_url )#line:80 O000OO0OO0000O0O0
={"SerialNo":O00O00OO00O000O00 ,"State":O0OO0O00OOO0OOO00
,"iccid":O0O0OOO0OO00O00O0 }#line:81 O000OO0OO0000O0O0 =urllib .urlencode
(O000OO0OO0000O0O0 )#line:82 print (O000OO0OO0000O0O0 )#line:83
O0OO0O0OOOO00O0O0 =urllib2 .Request (OOOO000OOOO00O0O0 .gws_url
,O000OO0OO0000O0O0 )#line:84 try :#line:86 urllib2 .urlopen (O0OO0O0OOOO00O0O0
,timeout =2 )#line:87 except :#line:88 traceback .print_exc ()#line:89 return
#line:90 def SaveHeartBeatData (O0O00O000OO00O0O0 ,O0O00O00O00O00O00
,OO00O0OOOOO000OO0 ):#line:92 print ("提交心跳信号到%s"%O0O00O000OO00O0O0 .shbd_url
)#line:93 O0OOOOOOO0O0OO0OO ={"f0003":O0O00O00O00O00O00
,"f0009":OO00O0OOOOO000OO0 }#line:94 O0OOOOOOO0O0OO0OO =urllib .urlencode
(O0OOOOOOO0O0OO0OO )#line:95 print (O0OOOOOOO0O0OO0OO )#line:96
OOOO00OOO00000OOO =urllib2 .Request (O0O00O000OO00O0O0 .shbd_url
,O0OOOOOOO0O0OO0OO )#line:97 try :#line:99 urllib2 .urlopen (OOOO00OOO00000OOO
,timeout =2 )#line:100 except :#line:101 traceback .print_exc ()#line:102
return #line:103 class myG :#line:114 getway_list ={}#line:116 connCount =0
#line:118 def __init__ (O0O0OOO00OOO00000 ):#line:120 pass #line:121 def
add_keyvalue (OOO00OO000O000O00 ,O0OOO000O00000O0O ,OO0OO0OOO0OO0000O
):#line:123 OO0000OOO0OOO0O0O =dict (OOO00OO000O000O00 .getway_list )#line:124
OO0000OOO0OOO0O0O [O0OOO000O00000O0O ]=OO0OO0OOO0OO0000O #line:125 print
('添加连接对象成功')#line:126 OOO00OO000O000O00 .getway_list =OO0000OOO0OOO0O0O
#line:127 return OO0000OOO0OOO0O0O #line:128 def key_exists (O00O000O0OO0000OO
,OOOOO0OO00O00O0O0 ):#line:130 OOOOOOOO0OO0000O0 =dict (O00O000O0OO0000OO
.getway_list )#line:131 return OOOOOOOO0OO0000O0 .has_key (OOOOO0OO00O00O0O0
)#line:132 def remove_key (O00OO0O0O0O00OO0O ,OOOOO00O0O0OO0OO0 ):#line:134
O0OOO0OO0OOO0O0OO =dict (O00OO0O0O0O00OO0O .getway_list )#line:135 del
O0OOO0OO0OOO0O0OO [OOOOO00O0O0OO0OO0 ]#line:136 return O0OOO0OO0OOO0O0OO
#line:137 def get_value (OOOOO0OOOO000O0OO ,OOOO0O00O0OO0OOO0 ):#line:139
return OOOOO0OOOO000O0OO .getway_list .get (OOOO0O00O0OO0OOO0 )#line:140 class
mymqtt ():#line:142 client =None #line:143 def __init__ (O0000O0O00O0O0000
):#line:144 OO00O0O0OO00000OO =mqtt .Client ()#line:145 OO00O0O0OO00000OO
.username_pw_set ("admin","password")#line:146 OO00O0O0OO00000OO .on_connect
=O0000O0O00O0O0000 .on_connect #line:147 OO00O0O0OO00000OO .on_message
=O0000O0O00O0O0000 .on_message #line:148 O0O000O00000000OO
="134.98.1.210"#line:149 OO00O0O0OO00000OO .connect (O0O000O00000000OO ,1883
,160 )#line:151 OOOO0000O00OOOOOO =uuid .uuid1 ()#line:153 OO00O0O0OO00000OO
.user_data_set (OOOO0000O00OOOOOO )#line:154 OO00O0O0OO00000OO .loop_start
()#line:156 O0000O0O00O0O0000 .client =OO00O0O0OO00000OO #line:157 def send
(OO00OO00O0OOOOO0O ,O00O00OO00O000000 ,O0O00O0OO0OO000OO ):#line:159 try
:#line:160 OO00OO00O0OOOOO0O .client .publish (O00O00OO00O000000
,O0O00O0OO0OO000OO )#line:161 except :#line:162 traceback .print_exc
()#line:163 return #line:164 def on_connect (O0000O0OOOOOO00O0
,OOO00O0O0OOOO0OOO ,OO0OOO0O0OOOO000O ,OO00O000OOOOOOO00 ,OO0OOO00000OO0000
):#line:166 OOO00O0O0OOOO0OOO .subscribe ("juganwg")#line:169 def on_message
(O000OOO0000O00O0O ,O0OO0O0O00OO0OO00 ,O0O0OO0OO0OO00000 ,OO0OO0OOO00OOOOOO
):#line:173 O00O0000O000OO00O ="%s"%(str (OO0OO0OOO00OOOOOO .topic ))#line:174
O0OOO00O00OO00O00 ="%s"%(str (OO0OO0OOO00OOOOOO .payload ))#line:175 if
O00O0000O000OO00O !="juganwg":#line:176 return #line:177 if not
O0OOO00O00OO00O00 :#line:179 return #line:180 print
("从(%s)接受到指令:%s"%(O00O0000O000OO00O ,O0OOO00O00OO00O00 ))#line:181 if
O0OOO00O00OO00O00 .find ("}")==-1 :#line:183 return #line:184 O0OOO00O00OO00O00
=O0OOO00O00OO00O00 [O0OOO00O00OO00O00 .index ('{'):O0OOO00O00OO00O00 .rindex
('}')+1 ]#line:185 O0000OO0O0OOO000O =""#line:186 try :#line:187
O0000OO0O0OOO000O =json .loads (O0OOO00O00OO00O00 )#line:188 OOOOOO0OO0O000O0O
=O0000OO0O0OOO000O ['name']#line:189 OOO0000O0OOO0O000 =O0000OO0O0OOO000O
['type']#line:190 OOO0000OOO0OO0O00 =O0000OO0O0OOO000O ['seq']#line:191
O00O00O0OOO0OO000 =O0000OO0O0OOO000O ['gwid']#line:192 OOO00000O00O00O0O =myg
.get_value (O00O00O0OOO0OO000 )#line:195 if OOO00000O00O00O0O :#line:196 try
:#line:197 OOO00000O00O00O0O .send (O0OOO00O00OO00O00 )#line:198 except
:#line:199 return #line:200 except :#line:201 traceback .print_exc ()#line:202
return #line:203 class MyStreamRequestHandlerr (StreamRequestHandler
):#line:205 def SaveHeartBeatData (OO0O00O0OOO00O00O ,OOO00OOO00000OOOO
,O0OO0OOOOO000000O ):#line:206 OO00O00OO0OO00OO0 ="f0003="+OOO00OOO00000OOOO
+"&f0009="+O0OO0OOOOO000000O #line:207 OO00O0000000O00OO =urllib .urlencode
(OO00O00OO0OO00OO0 )#line:208 OO00O0O00O0OOO0O0
="http://134.98.1.37:8080/t0041/saveHeartbeatData"#line:209 O0O0O0O0OOOO0OOOO
={"Host":"134.98.1.37"}#line:210 O0000O0O00O0O0OO0 =httplib .HTTPConnection
("134.98.1.37:8080")#line:211 O0000O0O00O0O0OO0 .request (method ="POST",url
=OO00O0O00O0OOO0O0 ,body =OO00O0000000O00OO ,headers =O0O0O0O0OOOO0OOOO
)#line:212 def handle (OO0OO0000OOO0OO00 ):#line:214 myg .connCount =myg
.connCount +1 #line:215 O0O000OOO00OOOO0O =""#line:216 O00000O0OO00O00OO
=""#line:217 print ("有新的客户端连接,总连接数%d"%myg .connCount )#line:218 try :#line:219
while True :#line:220 try :#line:221 OO0O00OOO0O000O00 =OO0OO0000OOO0OO00
.request .recv (1024 )#line:223 OO0O00OOO0O000O00 ="%s"%(OO0O00OOO0O000O00
)#line:224 if (OO0O00OOO0O000O00 =='\xff'):#line:225 myg .connCount =myg
.connCount -1 #line:227 try :#line:228 mypost .GetwayStatusData
(O00000O0OO00O00OO ,"0",O0O000OOO00OOOO0O )#line:229 except :#line:230
traceback .print_exc ()#line:231 pass #line:232 print ("有客户端断开连接,总连接数%d"%myg
.connCount )#line:233 break #line:234 if not OO0O00OOO0O000O00 :#line:235 break
#line:236 print ("从(%r)接受到数据:%r"%(OO0OO0000OOO0OO00 .client_address
,OO0O00OOO0O000O00 ))#line:237 if OO0O00OOO0O000O00 .find ("}")==-1 :#line:239
break #line:240 OO0O00OOO0O000O00 =OO0O00OOO0O000O00 [OO0O00OOO0O000O00 .index
('{'):OO0O00OOO0O000O00 .rindex ('}')+1 ]#line:241 if not OO0O00OOO0O000O00
:#line:242 break #line:243 O00OOOOO0OO00O00O =""#line:244 try :#line:245
O00OOOOO0OO00O00O =json .loads (OO0O00OOO0O000O00 )#line:246 O0O0O00000O0OO0O0
=O00OOOOO0OO00O00O ['name']#line:247 O0OOOOOO00O0OO000 =O00OOOOO0OO00O00O
['type']#line:248 O0OO0O0OO00O0OO00 =O00OOOOO0OO00O00O ['seq']#line:249 except
:#line:250 break #line:251 try :#line:253 OOOO00O0OO0O00000 =O00OOOOO0OO00O00O
['iccid']#line:254 O0O000OOO00OOOO0O =OOOO00O0OO0O00000 #line:255 except
:#line:256 OOOO00O0OO0O00000 =O0O000OOO00OOOO0O #line:257 try :#line:259
OOOO00O000O0O0O0O =O00OOOOO0OO00O00O ['gwid']#line:260 O00000O0OO00O00OO
=OOOO00O000O0O0O0O #line:261 except :#line:262 OOOO00O000O0O0O0O
=O00000O0OO00O00OO #line:263 if (O0O0O00000O0OO0O0 =='login'and
O0OOOOOO00O0OO000 =='rqst'):#line:267 OO0000O00OOOO0O0O
='{"name":"login","type":"rsps","seq":%d,"result":"ok"}'%(O0OO0O0OO00O0OO00
)#line:268 print ("收到登陆请求，准备回应:%s"%OO0000O00OOOO0O0O )#line:270
OO0OO0000OOO0OO00 .request .send (OO0000O00OOOO0O0O .encode ())#line:271
OO0OO0000OOO0OO00 .wfile .flush ()#line:272 OO0OO0000OOO0OO00 .wfile .close
()#line:273 if myg .key_exists (OOOO00O000O0O0O0O ):#line:275 myg .remove_key
(OOOO00O000O0O0O0O )#line:276 myg .add_keyvalue (OOOO00O000O0O0O0O
,OO0OO0000OOO0OO00 .request )#line:277 if OOOO00O000O0O0O0O
=='undefine':#line:278 myg .remove_key ('undefine')#line:279 try :#line:280
mypost .GetwayStatusData (OOOO00O000O0O0O0O ,"1",OOOO00O0OO0O00000 )#line:281
except :#line:282 traceback .print_exc ()#line:283 pass #line:284 print
("登陆请求已经回应 %s"%(time .strftime ('%Y-%m-%d %H:%M:%S',time .localtime (time .time
()))))#line:285 if (O0O0O00000O0OO0O0 =='heartbeat'and O0OOOOOO00O0OO000
=='rqst'):#line:286 OO0000O00OOOO0O0O
='{"name":"heartbeat","type":"rsps","seq":%d,"time": "%s"}'%(O0OO0O0OO00O0OO00
,time .strftime ('%Y%m%d%H%M%S',time .localtime (time .time ())))#line:287
print ("收到心跳请求，准备回应:%s"%OO0000O00OOOO0O0O )#line:288 OO0OO0000OOO0OO00 .request
.send (OO0000O00OOOO0O0O .encode ())#line:289 OO0OO0000OOO0OO00 .wfile .flush
()#line:290 OO0OO0000OOO0OO00 .wfile .close ()#line:291 if myg .key_exists
(OOOO00O000O0O0O0O ):#line:293 myg .remove_key (OOOO00O000O0O0O0O )#line:294
myg .add_keyvalue (OOOO00O000O0O0O0O ,OO0OO0000OOO0OO00 .request )#line:295 if
OOOO00O000O0O0O0O =='undefine':#line:296 myg .remove_key ('undefine')#line:297
try :#line:298 mypost .SaveHeartBeatData (OOOO00O000O0O0O0O ,time .strftime
('%Y-%m-%d %H:%M:%S',time .localtime (time .time ())))#line:299 except
:#line:300 traceback .print_exc ()#line:301 pass #line:302 print ("心跳请求已经回应
%s"%(time .strftime ('%Y-%m-%d %H:%M:%S',time .localtime (time .time
()))))#line:303 if (O0O0O00000O0OO0O0 =='gw_status'and O0OOOOOO00O0OO000
=='ntf'):#line:305 OO00000O0OO000OO0 =O00OOOOO0OO00O00O ['alarm']#line:306 if
OO00000O0OO000OO0 =='on':#line:307 try :#line:308 mypost .GetwayStatusData
(OOOO00O000O0O0O0O ,'2',O0O000OOO00OOOO0O );#line:309 except :#line:310 pass
#line:311 if OO00000O0OO000OO0 =='off':#line:312 mypost .GetwayStatusData
(OOOO00O000O0O0O0O ,'1',O0O000OOO00OOOO0O );#line:313 try :#line:314 pass
#line:315 except :#line:316 pass #line:317 if (O0O0O00000O0OO0O0 =='data'and
O0OOOOOO00O0OO000 =='rqst'):#line:319 OO0000O00OOOO0O0O
='{"name":"data","type":"rsps","seq":%d,"result":"ok"}'%(O0OO0O0OO00O0OO00
)#line:320 print ("收到数据请求，准备回应:%s"%OO0000O00OOOO0O0O )#line:321
OO0OO0000OOO0OO00 .request .send (OO0000O00OOOO0O0O .encode ())#line:322
OO0OO0000OOO0OO00 .wfile .flush ()#line:323 OO0OO0000OOO0OO00 .wfile .close
()#line:324 try :#line:325 mypost .SaveChannelData (OOOO00O000O0O0O0O
,OO0O00OOO0O000O00 ,time .strftime ('%Y-%m-%d %H:%M:%S',time .localtime (time
.time ())))#line:326 except :#line:327 pass #line:328 print ("数据请求已经回应
%s"%(time .strftime ('%Y-%m-%d %H:%M:%S',time .localtime (time .time
()))))#line:330 if (O0OOOOOO00O0OO000 =='rsps'):#line:331 OO0000O00OOOO0O0O
='{"name":"data","type":"rsps","seq":%d,"result":"ok"}'%(O0OO0O0OO00O0OO00
)#line:332 print ("收到网关回应，准备推送:%s"%OO0O00OOO0O000O00 )#line:333 mymqtt1 .send
(OOOO00O000O0O0O0O ,OO0O00OOO0O000O00 )#line:334 print ("网关回应已经推送 %s"%(time
.strftime ('%Y-%m-%d %H:%M:%S',time .localtime (time .time ()))))#line:335
except :#line:336 traceback .print_exc ()#line:339 break #line:340 except
:#line:341 traceback .print_exc ()#line:342 myg .connCount =myg .connCount -1
#line:343 print ("有新的客户端连接,总连接数%d"%myg .connCount )#line:344 pass #line:345 if
__name__ =="__main__":#line:348 myg =myG ()#line:349 host ="0.0.0.0"#line:350
port =6503 #line:351 addr =(host ,port )#line:352 addr2 =("127.0.0.1",6503
)#line:353 server =ThreadingTCPServer (addr ,MyStreamRequestHandlerr )#line:357
mypost =myPost
("http://134.98.1.37:8080/t0040/saveChannelData","http://134.98.1.37:8080/t0037/updateState","http://134.98.1.37:8080/t0041/saveHeartbeatData")#line:363
mymqtt1 =mymqtt ()#line:364 print ('start')#line:365 t =Thread (target =server
.serve_forever ())#line:370 t .daemon =False #line:372 t .start ()#line:373
print ('start1')#line:376 print ('start2')#line:378
支持功能：设备登陆验证、设备数据接收、设备数据转发、平台指令下发到设备  等

本人原创未经许可，可随意转载！

• « 上一篇：kotlin data class 的那些坑解决
• » 下一篇：☆ C++中一个函数调用其他.cpp文件中的函数