1.绠€浠媘ica-mqtt鏄竴涓熀浜巘-io鐨勭畝鍗曘€佷綆寤惰繜銆侀珮鎬ц兘鐨刴qtt鐗╄仈缃戝紑婧愮粍浠躲€傝鎯呭弬瑙乵ica-mqttgitee婧愮爜mica-mqtt-example妯″潡銆?.鍔熻兘[x]鏀寔MQTTv3.1銆乿3.1.1鍜寁5.0鍗忚銆俒x]鏀寔MQTT瀹㈡埛绔€俒x]鏀寔MQTT鏈嶅姟鍣ㄦ湇鍔″櫒銆俒x]鏀寔MQTT閬楀槺娑堟伅銆俒x]鏀寔MQTT淇濈暀娑堟伅銆俒x]鏀寔鑷畾涔夋秷鎭紙mq锛夊鐞嗗拰杞彂瀹炵幇闆嗙兢銆俒x]MQTT瀹㈡埛绔樋閲屼簯mqtt杩炴帴demo銆俒x]鏀寔GraalVM缂栬瘧鏈満鍙墽琛屾枃浠躲€俒x]鏀寔蹇€熻闂甋pringboot椤圭洰锛坢ica-mqtt-spring-boot-starter锛夈€俒x]mica-mqtt-spring-boot-starter鏀寔瀵规帴Prometheus+Grafana銆?.鍋歔]娣诲姞websocket鏀寔锛堥鐮旀垚鍔燂級銆俒]浼樺寲mqttsession澶勭悊锛屾敮鎸乿5.04.鏇存柊璁板綍馃摑鏂囨。澧炲姞闆嗙兢澶勭悊姝ラ锛屽鍔爓ill娑堟伅锛屼繚鐣欐秷鎭娇鐢ㄥ満鏅€傗湪鍘婚櫎demo涓殑qos2鍙傛暟锛屾€ц兘鎹熷け杈冨ぇ锛岄伩鍏嶈鎿嶄綔銆傗湪閬楀槺锛屼繚鐣欐秷鎭唴閮ㄦ秷鎭浆鍙戞娊璞°€傗湪娣诲姞mica-mqtt-spring-boot-example銆傛劅璋sq锛園鍐锋湀瀹富锛塸r銆傗湪mica-mqtt-spring-boot-starter鏀寔瀹㈡埛绔闂拰鏈嶅姟绔紭鍖栥€傛劅璋sq锛園鍐锋湀瀹富锛塸r銆傗湪mica-mqtt-spring-boot-starter鏈嶅姟鍣ㄦ敮鎸佹寚鏍囨敹闆嗐€傚彲浠ヨ繛鎺rometheus+Grafana鐩戞帶銆傗湪褰搈qtt鏈嶅姟鍣ㄦ帴鍙椾竴涓繛鎺ユ椂锛岄鍏堝垽鏂clientId鏄惁杩樻湁鍏朵粬杩炴帴锛屽鏋滄湁鍒欒В缁戝苟鍏抽棴鍏朵粬杩炴帴銆傗瑔锔忓崌绾ica-auto鍒?.1.3淇ide澶氭ā鍧楀閲忕紪璇戦棶棰樸€?.Springboot蹇€熸帴鍏?.1娣诲姞渚濊禆net.dreamlumica-mqtt-spring-boot-starter1.0.2/dependency>5.2鏈嶅姟鍣▂ml閰嶇疆mqtt:server:enabled:true#鏄惁鍚敤锛岄粯璁わ細trueip:127.0.0.1#鏈嶅姟鍣╥pdefault:127.0.0.1port:5883#绔彛锛岄粯璁わ細1883name:Mica-Mqtt-Server#鍚嶇О锛岄粯璁わ細Mica-Mqtt-Serverbuffer-allocator:HEAP#鍫嗗唴瀛樺拰鍫嗗鍐呭瓨锛岄粯璁わ細鍫嗗唴瀛榟eartbeat-timeout:120000#蹇冭烦瓒呮椂锛屽崟浣嶆绉掞紝榛樿锛?000*120read-buffer-size:8092#鎺ユ敹鏁版嵁鐨勭紦鍐插尯澶у皬锛岄粯璁わ細8092max-bytes-in-message:8092#娑堟伅瑙ｆ瀽鐨勬渶澶у瓧鑺傞暱搴︼紝榛樿锛?092debug:true#濡傛灉鍚敤prometheus鎸囨爣鏀堕泦锛屽缓璁甤lose5.3鏈嶅姟鍣ㄧ鍙疄鐜扮殑鎺ュ彛锛堟敞鍐屼负SpringBean鍗冲彲锛夋帴鍙ｆ槸鍚﹀繀椤绘爣鏄嶪MqttServerAuthHandler鐢ㄤ簬瀹㈡埛绔璇併€侷MqttMessageListener鐢ㄤ簬娑堟伅鐩戝惉銆侷MqttConnectStatusListener鐢ㄤ簬杩炴帴鐘舵€佺洃鎺с€侷MqttSessionManager涓嶉€傜敤浜庝細璇濈鐞嗐€侷MqttMessageStorecluster鏄痽es锛宻tand-alone鏄疦o锛屼細骞朵繚鐣欐秷鎭瓨鍌ㄣ€侫bstractMqttMessageDispatcher闆嗙兢鏄紝鍗曟満涓嶈锛屾秷鎭浆鍙戯紝(浼氾紝淇濈暀娑堟伅杞彂锛塈pStatListenernot-ioip鐘舵€佺洃鍚?.4Server鑷畾涔夐厤缃紙鍙€夛級@Configuration(proxyBeanMethods=false)publicclassMqttServerCustomizerConfiguration{@BeanpublicMqttServerCustomizeractiveRecordPluginCustomizer(){returnnewMqttServer(Customizer){Overridepublicvoidcustomize(MqttServerCreatorcreator){//杩欓噷鍙互鑷畾涔塩reator锛屼細瑕嗙洊yml涓殑閰嶇疆System.out.println("----------------MqttServerCustomizer------------------");}};}}5.5MqttServerTemplate绀轰緥importnet.dreamlu.iot.mqtt.codec.MqttQoS;importnet.dreamlu.iot.mqtt.spring.server.MqttServerTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;importjava.nio.ByteBuffer;/***@authorwsq*/@ServicepublicclassServerService{@鑷姩瑁呴厤绉佹湁MqttServerTemplate鏈嶅姟鍣紱publicbooleanpublish(Stringbody){server.publishAll("/test/123",ByteBuffer.wrap(body.getBytes()));杩斿洖鐪燂紱}}5.6鍩轰簬mq娑堟伅骞挎挱闆嗙兢澶勭悊瀹炵幇IMqttConnectStatusListener澶勭悊璁惧鐘舵€佸瓨鍌ㄣ€侷MqttMessageListener灏嗘秷鎭浆鍙戠粰mq锛屼笟鍔℃寜闇€澶勭悊mq娑堟伅銆傚疄鐜癐MqttMessageStore浠ュ瓨鍌ㄩ仐鍢卞拰棰勮娑堟伅銆傚疄鐜癆bstractMqttMessageDispatcher鍚憁q鍙戦€佹秷鎭紝mq骞挎挱鍥瀖qtt闆嗙兢锛宮qtt鍚戣澶囧彂閫佹秷鎭€備笟鍔℃秷鎭彂閫佸埌mq锛宮q骞挎挱鍒癿qtt闆嗙兢锛宮qtt鍙戦€佹秷鎭埌璁惧銆?.7Prometheus+Grafana鐩戞帶杩炴帴寰楃泭浜巘-io鐨勮壇濂借璁°€傜洃鎺ф寚鏍囩洿鎺ヨ繛鎺ュ埌t-iostat銆傜洰鍓嶆敮鎸佷互涓嬫寚鏍囷紝鏈潵浼氫笉鏂畬鍠勩€傛敮鎸佸緱鎸囨爣璇存槑mqtt_connections_accepted鍏辨帴鍙楄繃杩炴帴鏁癿qtt_connections_closed鍏抽棴杩囩殑杩炴帴鏁癿qtt_connections_size褰撳墠杩炴帴鏁癿qtt_messages_handled_packets宸插鐞嗘秷鎭暟mqtt_messages_handled_bytes宸插鐞嗘秷鎭瓧鑺傛暟mqtt_messages_received_packets宸叉帴鏀舵秷鎭暟mqtt_messages_received_bytes宸插鐞嗘秷鎭瓧鑺傛暟mqtt_messages_send_packets宸插彂閫佹秷鎭暟mqtt_messages_send_bytesForthenumberofsentmessagebytes,pleaserefertothedocumentformoreusagemethodssuchasmica-mqtt-spring-boot-starterclient:https://gitee.com/596392912/mica-mqtt/tree/master/mica-mqtt-spring-boot-starter6.Ordinaryjavaprojectaccess6.1mavendependsonnet.dreamlumica-mqtt-core1.0.26.2mica-mqttclient//InitializemqttclientMqttClientclient=MqttClient.create().ip("127.0.0.1").port(1883)//Default:1883.username("admin").password("123456").version(MqttVersion.MQTT_5)//Default:3_1_1.clientId("xxxxxx")//Default:MICA-MQTT-prefixand36-nanoseconds.connect();//杩炴帴//娑堟伅璁㈤槄锛屾柟娉曠被浼約ubxxxclient.subQos0("/test/#",(topic,payload)->{logger.info(topic+'\t'+ByteBufferUtil.toString(payload));});//鍙栨秷璁㈤槄client.unSubscribe("/test/#");//鍙戦€佹秷鎭痗lient.publish("/test/client",ByteBuffer.wrap("浜戞瘝鏈€鐗涚毊".getBytes(StandardCharsets.UTF_8)));//鏂紑杩炴帴client.disconnect();//閲嶆柊杩炴帴client.reconnect();//鍋滄client.stop();6.3mica-mqttserver//娉ㄦ剰锛氫负浜嗘帴鍙楁洿澶氱殑杩炴帴锛堜綆鍐呭瓨锛夛紝璇锋坊鍔爅vm鍙傛暟-Xss129kMqttServermqttServer=MqttServer.create()//default:127.0.0.1.ip("127.0.0.1")//default:1883.port(1883)//default:8092(mqtt榛樿鏈€澶ф秷鎭ぇ灏?锛屼负浜嗗噺灏戝唴瀛橈紝鍙互鍑忓皬杩欎釜鍙傛暟锛屽鏋滄秷鎭お澶?t-io浼氬皾璇曞娆¤В鏋愶紙鏍规嵁瀹為檯涓氬姟鎯呭喌鎺ㄨ崘锛?readBufferSize(512)//鑷畾涔塧uthentication.authHandler((clientId,userName,password)->true)//娑堟伅鐩戝惉銆俶essageListener((clientId,topic,mqttQoS,payload)->{logger.info("clientId:{}topic:{}mqttQoS:{}message:{}",clientId,topic,mqttQoS,ByteBufferUtil.toString(payload));})//ssl閰嶇疆.useSsl("","","")//鑷畾涔夊鎴风娉ㄩ攢鍜岀绾跨洃鍚?connectStatusListener(newIMqttConnectStatusListener(){@Overridepublicvoidonline(StringclientId){}@Overridepublicvoidoffline(StringclientId){}})//鑷畾涔夋秷鎭浆鍙戯紝鍙互閫氳繃mq骞挎挱瀹炵幇闆嗙兢澶勭悊銆俶essageDispatcher(newIMqttMessageDispatcher(){@Overridepublicvoidconfig(MqttServermqttServer){}@Overridepublicbooleansend(Messagemessage){returnfalse;}@Overridepublicbooleansend(StringclientId,Messagemessage){杩斿洖false;}}).debug()//鍚敤t-io璋冭瘯淇℃伅log.start();//鍙戦€佺粰鏌愪釜瀹㈡埛绔痬qttServer.publish("clientId","/test/123",ByteBuffer.wrap("浜戞瘝鏈€鐗涚毊".getBytes()));//鍙戦€佺粰鏈瘽棰樼殑鎵€鏈夊湪绾垮惉浼桟lientmqttServer.publishAll("/test/123",ByteBuffer.wrap("浜戞瘝鏈€鐗涚毊".getBytes()));//鍋滄鏈嶅姟mqttServer.stop();7.鏁堟灉婕旂ず8.鍏虫敞鎴戜滑鎵弿涓婃柟浜岀淮鐮侊紝姣忓ぉ閮芥湁鏇村绮惧僵鍐呭鎺ㄨ崘锛?/p>
