SpringBoot+MQTT+apollo实现订阅发布功能的示例
由于最近公司在开发一款后台与安卓的更新系统，经过再三研究之后，也是选择Mqtt这个目前流行的框架。为了能够让项目运营起来，最终虽说是选择ActiveMQ。但在这个过程中，也是发现Apollo作为服务器也是相当不错。当然对于后者已经被apace放弃，不过今天还是和大家整理一下SpringBoot+MQTT+apollo实现订阅发布功能的全过程。
对于项目首先需要用到的前提东西，比如Apollo如何下载，以及MQTT测试工具在这里就不多说。如果真的不懂私聊Damon吧，在这里就不浪费时间。
对于项目，首先你所需要引入maven包：
pom.xml
org.springframework.integration spring-integration-stream org.springframework.integration spring-integration-mqtt
其目标就是将MQTT用在项目组中
接着就是项目yml文件的配置，使用properties，以葫芦画瓢就行了：
applicaiton.yml
mqtt: username:admin password:password host-url:tcp://127.0.0.1:8161#你自己服务器的地址和端口，这个需要改 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的地址
来到第三步，此时就是项目内的文件：
MqttConfig文件
@Component
@ConfigurationProperties("mqtt")
@Setter
@Getter
publicclassMqttConfig{
@Autowired
privateMqttPushClientmqttPushClient;
/**
*用户名
*/
//@Value("username")
privateStringusername;
/**
*密码
*/
privateStringpassword;
/**
*连接地址
*/
privateStringhostUrl;
/**
*客户Id
*/
privateStringclientID;
/**
*默认连接话题
*/
privateStringdefaultTopic;
/**
*超时时间
*/
privateinttimeout;
/**
*保持连接数
*/
privateintkeepalive;
@Bean
publicMqttPushClientgetMqttPushClient(){
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);
//以/#结尾表示订阅所有以test开头的主题
mqttPushClient.subscribe(defaultTopic,0);
returnmqttPushClient;
}
}
目的就是配置所对应的消息
第四步就是发布以及订阅等功能：
MqttPushClient
@Component
publicclassMqttPushClient{
privatestaticfinalLoggerlogger=LoggerFactory.getLogger(MqttPushClient.class);
@Autowired
privatePushCallbackpushCallback;
privatestaticMqttClientclient;
privatestaticMqttClientgetClient(){
returnclient;
}
privatestaticvoidsetClient(MqttClientclient){
MqttPushClient.client=client;
}
/**
*客户端连接
*
*@paramhostip+端口
*@paramclientID客户端Id
*@paramusername用户名
*@parampassword密码
*@paramtimeout超时时间
*@paramkeepalive保留数
*/
publicvoidconnect(Stringhost,StringclientID,Stringusername,Stringpassword,inttimeout,intkeepalive){
MqttClientclient;
try{
client=newMqttClient(host,clientID,newMemoryPersistence());
MqttConnectOptionsoptions=newMqttConnectOptions();
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(Exceptione){
e.printStackTrace();
}
}catch(Exceptione){
e.printStackTrace();
}
}
/**
*发布
*
*@paramqos连接方式
*@paramretained是否保留
*@paramtopic主题
*@parampushMessage消息体
*/
publicvoidpublish(intqos,booleanretained,Stringtopic,StringpushMessage){
MqttMessagemessage=newMqttMessage();
message.setQos(qos);
message.setRetained(retained);
message.setPayload(pushMessage.getBytes());
MqttTopicmTopic=MqttPushClient.getClient().getTopic(topic);
if(null==mTopic){
logger.error("topicnotexist");
}
MqttDeliveryTokentoken;
try{
token=mTopic.publish(message);
token.waitForCompletion();
}catch(MqttPersistenceExceptione){
e.printStackTrace();
}catch(MqttExceptione){
e.printStackTrace();
}
}
/**
*订阅某个主题
*
*@paramtopic主题
*@paramqos连接方式
*/
publicvoidsubscribe(Stringtopic,intqos){
logger.info("开始订阅主题"+topic);
try{
MqttPushClient.getClient().subscribe(topic,qos);
}catch(MqttExceptione){
e.printStackTrace();
}
}
}
订阅主题以及发布的方式等内容更多编写
最后在搞个测试看看我们的结果是否正确：
TestController
@RestController
@RequestMapping("/")
publicclassTestController{
@Autowired
privateMqttPushClientmqttPushClient;
@GetMapping(value="/publishTopic")
publicStringpublishTopic(){
StringtopicString="home/garden/fountain";
mqttPushClient.publish(0,false,topicString,"测试一下发布消息");
return"ok";
}
//发送自定义消息内容（使用默认主题）
@RequestMapping("/publishTopic/{data}")
publicStringtest1(@PathVariable("data")Stringdata){
StringtopicString="home/garden/fountain";
mqttPushClient.publish(0,false,topicString,data);
return"ok";
}
//发送自定义消息内容，且指定主题
@RequestMapping("/publishTopic/{topic}/{data}")
publicStringtest2(@PathVariable("topic")Stringtopic,@PathVariable("data")Stringdata){
mqttPushClient.publish(0,false,topic,data);
return"ok";
}
}
如此一来就OK！
你可以使用MQTT.fx进行测试。用Postman发出，就能够查看最终的结果。在这里，因为时间的原因就不多说，有啥有趣的问题，咱们可以一同探讨。希望你希望，Damon将会不断的分享各种有趣的开发小故事给大家娱乐。下一期，或在POI实现导出导入或者是ActiveMQ进行选择。
到此这篇关于SpringBoot+MQTT+apollo实现订阅发布功能的示例的文章就介绍到这了,更多相关SpringBoot+MQTT+apollo订阅发布内容请搜索毛票票以前的文章或继续浏览下面的相关文章希望大家以后多多支持毛票票！