SpringBoot+MQTT+apollo實現(xiàn)訂閱發(fā)布功能的示例
由于最近公司在開發(fā)一款后臺與安卓的更新系統(tǒng),經(jīng)過再三研究之后,也是選擇Mqtt這個目前流行的框架。為了能夠讓項目運營起來,最終雖說是選擇ActiveMQ。但在這個過程中,也是發(fā)現(xiàn)Apollo作為服務(wù)器也是相當不錯。當然對于后者已經(jīng)被apace放棄,不過今天還是和大家整理一下SpringBoot+MQTT+apollo實現(xiàn)訂閱發(fā)布功能的全過程。
對于項目首先需要用到的前提東西,比如Apollo如何下載,以及MQTT測試工具在這里就不多說。如果真的不懂私聊Damon吧,在這里就不浪費時間。
對于項目,首先你所需要引入maven包:
pom.xml
<!-- MQTT --> <dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-stream</artifactId> </dependency> <dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> </dependency>
其目標就是將MQTT用在項目組中
接著就是項目yml文件的配置,使用properties,以葫蘆畫瓢就行了:
applicaiton.yml
mqtt: username: admin password: password host-url: tcp://127.0.0.1:8161 # 你自己服務(wù)器的地址和端口,這個需要改 clientID: test1 # 這個改不改隨意,但不同的客戶端肯定不能一樣 default-topic: home/garden/fountain # 默認主題 timeout: 100 keepalive: 100 # Tomcat server: tomcat: uri-encoding: UTF-8 max-threads: 1000 min-spare-threads: 30 port: 8088
注意host-url,這就是你apollo的地址
來到第三步,此時就是項目內(nèi)的文件:
MqttConfig文件
@Component @ConfigurationProperties("mqtt") @Setter @Getter public class MqttConfig { @Autowired private MqttPushClient mqttPushClient; /** * 用戶名 */ // @Value("username") private String username; /** * 密碼 */ private String password; /** * 連接地址 */ private String hostUrl; /** * 客戶Id */ private String clientID; /** * 默認連接話題 */ private String defaultTopic; /** * 超時時間 */ private int timeout; /** * 保持連接數(shù) */ private int keepalive; @Bean public MqttPushClient getMqttPushClient() { System.out.println("hostUrl: "+ hostUrl); System.out.println("clientID: "+ clientID); System.out.println("username: "+ username); System.out.println("password: "+ password); System.out.println("timeout: "+timeout); System.out.println("keepalive: "+ keepalive); mqttPushClient.connect(hostUrl, clientID, username, password, timeout, keepalive); // 以/#結(jié)尾表示訂閱所有以test開頭的主題 mqttPushClient.subscribe(defaultTopic, 0); return mqttPushClient; } }
目的就是配置所對應(yīng)的消息
第四步就是發(fā)布以及訂閱等功能:
MqttPushClient
@Component public class MqttPushClient { private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class); @Autowired private PushCallback pushCallback; private static MqttClient client; private static MqttClient getClient() { return client; } private static void setClient(MqttClient client) { MqttPushClient.client = client; } /** * 客戶端連接 * * @param host ip+端口 * @param clientID 客戶端Id * @param username 用戶名 * @param password 密碼 * @param timeout 超時時間 * @param keepalive 保留數(shù) */ public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) { MqttClient client; try { client = new MqttClient(host, clientID, new MemoryPersistence()); MqttConnectOptions options = new MqttConnectOptions(); options.setCleanSession(true); options.setUserName(username); options.setPassword(password.toCharArray()); options.setConnectionTimeout(timeout); options.setKeepAliveInterval(keepalive); MqttPushClient.setClient(client); try { client.setCallback(pushCallback); client.connect(options); } catch (Exception e) { e.printStackTrace(); } } catch (Exception e) { e.printStackTrace(); } } /** * 發(fā)布 * * @param qos 連接方式 * @param retained 是否保留 * @param topic 主題 * @param pushMessage 消息體 */ public void publish(int qos, boolean retained, String topic, String pushMessage) { MqttMessage message = new MqttMessage(); message.setQos(qos); message.setRetained(retained); message.setPayload(pushMessage.getBytes()); MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic); if (null == mTopic) { logger.error("topic not exist"); } MqttDeliveryToken token; try { token = mTopic.publish(message); token.waitForCompletion(); } catch (MqttPersistenceException e) { e.printStackTrace(); } catch (MqttException e) { e.printStackTrace(); } } /** * 訂閱某個主題 * * @param topic 主題 * @param qos 連接方式 */ public void subscribe(String topic, int qos) { logger.info("開始訂閱主題" + topic); try { MqttPushClient.getClient().subscribe(topic, qos); } catch (MqttException e) { e.printStackTrace(); } } }
訂閱主題以及發(fā)布的方式等內(nèi)容更多編寫
最后在搞個測試看看我們的結(jié)果是否正確:
TestController
@RestController @RequestMapping("/") public class TestController { @Autowired private MqttPushClient mqttPushClient; @GetMapping(value = "/publishTopic") public String publishTopic() { String topicString = "home/garden/fountain"; mqttPushClient.publish(0, false, topicString, "測試一下發(fā)布消息"); return "ok"; } // 發(fā)送自定義消息內(nèi)容(使用默認主題) @RequestMapping("/publishTopic/{data}") public String test1(@PathVariable("data") String data) { String topicString = "home/garden/fountain"; mqttPushClient.publish(0,false,topicString, data); return "ok"; } // 發(fā)送自定義消息內(nèi)容,且指定主題 @RequestMapping("/publishTopic/{topic}/{data}") public String test2(@PathVariable("topic") String topic, @PathVariable("data") String data) { mqttPushClient.publish(0,false,topic, data); return "ok"; } }
如此一來就OK!
你可以使用MQTT.fx進行測試。用Postman發(fā)出,就能夠查看最終的結(jié)果。在這里,因為時間的原因就不多說,有啥有趣的問題,咱們可以一同探討。 希望你希望,Damon將會不斷的分享各種有趣的開發(fā)小故事給大家娛樂。下一期,或在POI實現(xiàn)導(dǎo)出導(dǎo)入或者是ActiveMQ進行選擇。
到此這篇關(guān)于SpringBoot+MQTT+apollo實現(xiàn)訂閱發(fā)布功能的示例的文章就介紹到這了,更多相關(guān)SpringBoot+MQTT+apollo訂閱發(fā)布內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
spring cloud consul注冊的服務(wù)報錯critical的解決
這篇文章主要介紹了spring cloud consul注冊的服務(wù)報錯critical的解決,小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2019-03-03Java使用ProcessBuilder?API優(yōu)化流程
Java?的?Process?API?為開發(fā)者提供了執(zhí)行操作系統(tǒng)命令的強大功能,這篇文章將詳細介紹如何使用?ProcessBuilder?API?來方便的操作系統(tǒng)命令,需要的可以收藏一下2023-06-06Java實現(xiàn)的生成二維碼和解析二維碼URL操作示例
這篇文章主要介紹了Java實現(xiàn)的生成二維碼和解析二維碼URL操作,結(jié)合實例形式分析了Java創(chuàng)建與解析二維碼,以及文件讀寫等相關(guān)操作技巧,需要的朋友可以參考下2018-07-07Spring Boot整合mybatis并自動生成mapper和實體實例解析
本文是小編給大家總結(jié)的關(guān)于Spring Boot整合mybatis并自動生成mapper和實體的內(nèi)容,非常不錯,具有一定的參考借鑒價值,需要的朋友參考下吧2018-05-05