由于最近公司在開(kāi)發(fā)一款后臺(tái)與安卓的更新系統(tǒng)，經(jīng)過(guò)再三研究之后，也是選擇Mqtt這個(gè)目前流行的框架。為了能夠讓項(xiàng)目運(yùn)營(yíng)起來(lái)，最終雖說(shuō)是選擇ActiveMQ。但在這個(gè)過(guò)程中，也是發(fā)現(xiàn)Apollo作為服務(wù)器也是相當(dāng)不錯(cuò)。當(dāng)然對(duì)于后者已經(jīng)被apace放棄，不過(guò)今天還是和大家整理一下SpringBoot+MQTT+apollo實(shí)現(xiàn)訂閱發(fā)布功能的全過(guò)程。

對(duì)于項(xiàng)目首先需要用到的前提東西，比如Apollo如何下載，以及MQTT測(cè)試工具在這里就不多說(shuō)。如果真的不懂私聊Damon吧，在這里就不浪費(fèi)時(shí)間。

對(duì)于項(xiàng)目，首先你所需要引入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>

其目標(biāo)就是將MQTT用在項(xiàng)目組中接著就是項(xiàng)目yml文件的配置，使用properties，以葫蘆畫(huà)瓢就行了：

applicaiton.yml

mqtt: username: admin password: password host-url: tcp://127.0.0.1:8161 # 你自己服務(wù)器的地址和端口，這個(gè)需要改 clientID: test1 # 這個(gè)改不改隨意，但不同的客戶(hù)端肯定不能一樣 default-topic: home/garden/fountain # 默認(rèn)主題 timeout: 100 keepalive: 100# Tomcatserver: tomcat: uri-encoding: UTF-8 max-threads: 1000 min-spare-threads: 30 port: 8088

注意host-url，這就是你apollo的地址

來(lái)到第三步，此時(shí)就是項(xiàng)目?jī)?nèi)的文件：

MqttConfig文件

@Component@ConfigurationProperties('mqtt')@Setter@Getterpublic class MqttConfig { @Autowired private MqttPushClient mqttPushClient; /** * 用戶(hù)名 */ // @Value('username') private String username; /** * 密碼 */ private String password; /** * 連接地址 */ private String hostUrl; /** * 客戶(hù)Id */ private String clientID; /** * 默認(rèn)連接話題 */ private String defaultTopic; /** * 超時(shí)時(shí)間 */ 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開(kāi)頭的主題 mqttPushClient.subscribe(defaultTopic, 0); return mqttPushClient; }}

目的就是配置所對(duì)應(yīng)的消息

第四步就是發(fā)布以及訂閱等功能：

MqttPushClient

@Componentpublic 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; } /** * 客戶(hù)端連接 * * @param host ip+端口 * @param clientID 客戶(hù)端Id * @param username 用戶(hù)名 * @param password 密碼 * @param timeout 超時(shí)時(shí)間 * @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(); } } /** * 訂閱某個(gè)主題 * * @param topic 主題 * @param qos 連接方式 */ public void subscribe(String topic, int qos) { logger.info('開(kāi)始訂閱主題' + topic); try { MqttPushClient.getClient().subscribe(topic, qos); } catch (MqttException e) { e.printStackTrace(); } }}

訂閱主題以及發(fā)布的方式等內(nèi)容更多編寫(xiě)最后在搞個(gè)測(cè)試看看我們的結(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, '測(cè)試一下發(fā)布消息'); return 'ok'; } // 發(fā)送自定義消息內(nèi)容（使用默認(rèn)主題） @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'; }}

如此一來(lái)就OK！

你可以使用MQTT.fx進(jìn)行測(cè)試。用Postman發(fā)出，就能夠查看最終的結(jié)果。在這里，因?yàn)闀r(shí)間的原因就不多說(shuō)，有啥有趣的問(wèn)題，咱們可以一同探討。 希望你希望，Damon將會(huì)不斷的分享各種有趣的開(kāi)發(fā)小故事給大家?jiàn)蕵?lè)。下一期，或在POI實(shí)現(xiàn)導(dǎo)出導(dǎo)入或者是ActiveMQ進(jìn)行選擇。

到此這篇關(guān)于SpringBoot+MQTT+apollo實(shí)現(xiàn)訂閱發(fā)布功能的示例的文章就介紹到這了,更多相關(guān)SpringBoot+MQTT+apollo訂閱發(fā)布內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！