SpringBoot+MQTT+apollo實現(xiàn)訂閱發(fā)布功能
但在這個過程中，也是發(fā)現(xiàn)Apollo作為服務(wù)器也是相當(dāng)不錯。當(dāng)然對于后者已經(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>
其目標(biāo)就是將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 # 默認(rèn)主題
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;
/**
* 默認(rèn)連接話題
*/
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)容（使用默認(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";
}
}
如此一來就OK！
你可以使用MQTT.fx進行測試。用Postman發(fā)出，就能夠查看最終的結(jié)果。在這里，因為時間的原因就不多說，有啥有趣的問題，咱們可以一同探討。
希望你希望，Damon將會不斷的分享各種有趣的開發(fā)小故事給大家娛樂。下一期，或在POI實現(xiàn)導(dǎo)出導(dǎo)入或者是ActiveMQ進行選擇。
關(guān)注公眾號：Java后端編程，回復(fù)下面關(guān)鍵字
要Java學(xué)習(xí)完整路線，回復(fù) 路線
缺Java入門視頻，回復(fù)： 視頻
要Java面試經(jīng)驗，回復(fù) 面試
缺Java項目，回復(fù)： 項目
進Java粉絲群： 加群
PS：如果覺得我的分享不錯，歡迎大家隨手點贊、在看。
（完） 加我"微信" 獲取一份 最新Java面試題資料 請備注：666，不然不通過～
最近好文
1、Spring Boot 實現(xiàn)掃碼登錄，這種方式太香了！！
2、SpringSecurity + JWT 實現(xiàn)單點登錄
3、基于 Vue+Spring 前后端分離管理系統(tǒng)ELAdmin
最近面試BAT，整理一份面試資料《Java面試BAT通關(guān)手冊》，覆蓋了Java核心技術(shù)、JVM、Java并發(fā)、SSM、微服務(wù)、數(shù)據(jù)庫、數(shù)據(jù)結(jié)構(gòu)等等。 獲取方式：關(guān)注公眾號并回復(fù) java 領(lǐng)取，更多內(nèi)容陸續(xù)奉上。 明天見(??ω??)??
評論
圖片
表情