EMQX配置
登录地址
首先打开EMQX的管理界面，界面的地址如下，
http://192.168.1.110:18083/规则是IP就是MQTT的IP，端口是固定的18083，输入该地址后，展示界面如下：
然后输入用户名和密码，用户名和密码就是MQTT连接的账号和密码。
设置中文
登录系统后，界面是默认是英文的，我们需要设置为中文。点击右上角的【设置】图标，然后就可以选择中文了。
选择简体中文后，点击保存即可。
设置速率
依次按照下图的操作步骤点击，点击【管理】，点击【速率限制】，然后输入相应的参数，点击确定即可。
EMQX 提供对接入速度、消息速度的限制，从入口处避免了系统过载，保证了系统的稳定和可预测的吞吐。
限制器类型
EMQX 使用以下几种类型的限制器来限制速率:
| 类型 | 描述 | 过载后行为 |
|---|---|---|
| bytes_rate | 单个客户端每秒流入的消息的字节数大小 | 暂停接收客户端消息 |
| messages_rate | 单个客户端每秒流入的消息条数 | 暂停接收客户端消息 |
| max_conn_rate | 当前监听器每秒的连接数 | 暂停接收新的连接 |
限制器可以在监听器级别上工作。例如，要为默认的TCP监听器设置限制器，可以在 emqx.conf 中按以下进行配置：
listeners.tcp.default {bind = "0.0.0.0:1883"max_conn_rate = "1000/s"messages_rate = "1000/s"bytes_rate = "1000MB/s"
}最大待发PUBREL数量配置
最大待发PUBREL数量配置，这个配置非常的重要，如果不配置那么我们的客户端模拟发送数据，可以达到默认的100就无法发送其它的数据了。比如我用客户端写一个DEMO，每5秒钟发送一次，一次发送200条数据，那么你在你的监听的客户端去看，它是无法收到刚才发送的200秒数据的。
客户端代码
POM依赖如下
<!-- mqtt --><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId></dependency><!-- JSONObject对象依赖的jar包 开始 --><dependency><groupId>commons-beanutils</groupId><artifactId>commons-beanutils</artifactId><version>1.9.3</version></dependency><dependency><groupId>commons-collections</groupId><artifactId>commons-collections</artifactId><version>3.2.1</version></dependency><dependency><groupId>commons-lang</groupId><artifactId>commons-lang</artifactId><version>2.6</version></dependency><dependency><groupId>commons-logging</groupId><artifactId>commons-logging</artifactId><version>1.1.1</version></dependency><dependency><groupId>net.sf.ezmorph</groupId><artifactId>ezmorph</artifactId><version>1.0.6</version></dependency><dependency><groupId>net.sf.json-lib</groupId><artifactId>json-lib</artifactId><version>2.2.3</version><classifier>jdk15</classifier><!-- jdk版本 --></dependency><!-- Json依赖架包下载结束 -->application.properties配置文件
## MQTT##
mqtt.host=tcp://192.168.1.110:1883
mqtt.clientId=ClientId_test_0717
mqtt.userName=admin
mqtt.passWord=public
mqtt.timeout=10
mqtt.keepalive=20
mqtt.topic1=iot_demo_reportMqttConfiguration.java
package cn.renkai721.mqtt;import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/**1. @author renkai721@163.com2. @date 2023/07/17 16:40*/
@Configuration
public class MqttConfiguration {private static final Logger log = LoggerFactory.getLogger(MqttConfiguration.class);@Value("${mqtt.host}")String host;@Value("${mqtt.username}")String username;@Value("${mqtt.password}")String password;@Value("${mqtt.clientId}")String clientId;@Value("${mqtt.timeout}")int timeOut;@Value("${mqtt.keepalive}")int keepAlive;@Value("${mqtt.topic1}")String topic1;@Beanpublic MyMQTTClient myMQTTClient() {MyMQTTClient myMQTTClient = new MyMQTTClient(host, username, password, clientId, timeOut, keepAlive);for (int i = 0; i < 10; i++) {try {myMQTTClient.connect();return myMQTTClient;} catch (MqttException e) {log.error("MQTT connect exception,connect time = " + i);try {Thread.sleep(2000);} catch (InterruptedException e1) {e1.printStackTrace();}}}return myMQTTClient;}public String getTopic1() {return topic1;}public void setTopic1(String topic1) {this.topic1 = topic1;}}
MqttController.java
package cn.renkai721.mqtt;import net.sf.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedList;
import java.util.Queue;/*** @author renkai721@163.com* @date 2023/07/17 16:40*/
@RestController
@EnableAsync
@RequestMapping("/sun/mqtt")
public class MqttController {@Autowiredprivate MyMQTTClient myMQTTClient;@Autowiredprivate MyMqttService myMqttService;@Value("${mqtt.topic1}")String topic1;@PostMapping("/sendMsg")@ResponseBodypublic String sendMsg() {myMqttService.sendMsg();return "success";}
}
MqttMsg.java
package cn.renkai721.mqtt;/*** @author renkai721@163.com* @date 2023/07/17 16:40*/
public class MqttMsg {private String name = "";private String content = "";private String time = "";public String getName() {return name;}public void setName(String name) {this.name = name;}public String getContent() {return content;}public void setContent(String content) {this.content = content;}public String getTime() {return time;}public void setTime(String time) {this.time = time;}@Overridepublic String toString() {return "MqttMsg{" +"name='" + name + '\'' +", content='" + content + '\'' +", time='" + time + '\'' +'}';}
}
MyMQTTCallback.java
package cn.renkai721.mqtt;import com.alibaba.fastjson.JSON;
import io.netty.util.CharsetUtil;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.util.Map;/*** @author renkai721@163.com* @date 2023/07/17 16:40*/
public class MyMQTTCallback implements MqttCallbackExtended {//手动注入private MqttConfiguration mqttConfiguration = SpringUtils.getBean(MqttConfiguration.class);private static final Logger log = LoggerFactory.getLogger(MyMQTTCallback.class);private MyMQTTClient myMQTTClient;public MyMQTTCallback(MyMQTTClient myMQTTClient) {this.myMQTTClient = myMQTTClient;}/*** 丢失连接，可在这里做重连* 只会调用一次** @param throwable*/@Overridepublic void connectionLost(Throwable throwable) {log.error("mqtt connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage());long reconnectTimes = 1;while (true) {try {if (MyMQTTClient.getClient().isConnected()) {log.warn("mqtt reconnect success end 重新连接 重新订阅成功");return;}reconnectTimes+=1;log.warn("mqtt reconnect times = {} try again... mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);MyMQTTClient.getClient().reconnect();} catch (MqttException e) {log.error("mqtt断连异常", e);}try {Thread.sleep(5000);} catch (InterruptedException e1) {}}}/*** @param topic* @param mqttMessage* @throws Exception* subscribe后得到的消息会执行到这里面*/@Overridepublic void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {log.info("接收消息主题 : {}，接收消息内容 : {}", topic, new String(mqttMessage.getPayload()));//你自己的业务接口}/***连接成功后的回调 可以在这个方法执行 订阅主题 生成Bean的 MqttConfiguration方法中订阅主题 出现bug*重新连接后 主题也需要再次订阅 将重新订阅主题放在连接成功后的回调 比较合理* @param reconnect* @param serverURI*/@Overridepublic void connectComplete(boolean reconnect,String serverURI){log.info("MQTT 连接成功，连接方式：{}",reconnect?"重连":"直连");//订阅主题myMQTTClient.subscribe(mqttConfiguration.topic1, 1);}/*** 消息到达后* subscribe后，执行的回调函数* publish后，配送完成后回调的方法** */@Overridepublic void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());}
}MyMQTTClient.java
package cn.renkai721.mqtt;import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/**1. @author renkai721@163.com2. @date 2023/07/17 16:40*/
public class MyMQTTClient {private static final Logger LOGGER = LoggerFactory.getLogger(MyMQTTClient.class);private static MqttClient client;private String host;private String username;private String password;private String clientId;private int timeout;private int keepalive;public MyMQTTClient(String host, String username, String password, String clientId, int timeOut, int keepAlive) {this.host = host;this.username = username;this.password = password;this.clientId = clientId;this.timeout = timeOut;this.keepalive = keepAlive;}public static MqttClient getClient() {return client;}public static void setClient(MqttClient client) {MyMQTTClient.client = client;}/*** 设置mqtt连接参数*/public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive) {MqttConnectOptions options = new MqttConnectOptions();options.setUserName(username);options.setPassword(password.toCharArray());options.setConnectionTimeout(timeout);options.setKeepAliveInterval(keepalive);options.setCleanSession(true);options.setAutomaticReconnect(true);return options;}/*** 连接mqtt服务端，得到MqttClient连接对象*/public void connect() throws MqttException {if (client == null) {client = new MqttClient(host, clientId, new MemoryPersistence());client.setCallback(new MyMQTTCallback(MyMQTTClient.this));}MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepalive);if (!client.isConnected()) {client.connect(mqttConnectOptions);} else {client.disconnect();client.connect(mqttConnectOptions);}LOGGER.info("MQTT connect success");//未发生异常，则连接成功}/*** 发布，默认qos为0，非持久化*/public void publish(String pushMessage, String topic) {publish(pushMessage, topic, 2, false);}/*** 发布消息**/public void publish(String pushMessage, String topic, int qos, boolean retained) {MqttMessage message = new MqttMessage();message.setPayload(pushMessage.getBytes());message.setQos(qos);message.setRetained(retained);MqttTopic mqttTopic = MyMQTTClient.getClient().getTopic(topic);if (null == mqttTopic) {LOGGER.error("topic is not exist");}MqttDeliveryToken token;synchronized (this) {try {token = mqttTopic.publish(message);token.waitForCompletion(1000L);} catch (MqttPersistenceException e) {e.printStackTrace();} catch (MqttException e) {e.printStackTrace();}}}/*** 订阅某个主题** @param topic* @param qos*/public void subscribe(String topic, int qos) {try {MyMQTTClient.getClient().subscribe(topic, qos);} catch (MqttException e) {e.printStackTrace();}}/*** 取消订阅主题** @param topic 主题名称*/public void cleanTopic(String topic) {if (client != null && client.isConnected()) {try {client.unsubscribe(topic);} catch (MqttException e) {e.printStackTrace();}} else {System.out.println("取消订阅失败！");}}
}
MyMqttService.java
package cn.renkai721.mqtt;import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/**1. @author renkai721@163.com2. @date 2023/07/17 16:40*/
@Service
@EnableAsync
@Slf4j
public class MyMqttService {@Autowiredprivate MyMQTTClient myMQTTClient;@Value("${mqtt.topic1}")String topic1;private static Map<String,String> longitudeAndLatitudeMap = new ConcurrentHashMap<>();static {longitudeAndLatitudeMap.put("F23000195","101.80964,3.552734");longitudeAndLatitudeMap.put("F23000194","2.364087,48.663712");longitudeAndLatitudeMap.put("F23000193","23.171214,-30.008971");longitudeAndLatitudeMap.put("F23000192","-59.777335,-22.381016");longitudeAndLatitudeMap.put("F23000001","121.49702,31.368703");longitudeAndLatitudeMap.put("F23000003","77.311839,13.87632");longitudeAndLatitudeMap.put("F23000004","104.022841,30.579637");longitudeAndLatitudeMap.put("F23000005","-81.963907,28.519213");longitudeAndLatitudeMap.put("F23000008","47.935073,29.28128");longitudeAndLatitudeMap.put("F23000010","139.212965,36.169801");longitudeAndLatitudeMap.put("F23000308","120.23624,22.92936");longitudeAndLatitudeMap.put("F23000307","120.32395,22.61752");longitudeAndLatitudeMap.put("F23000306","120.66108,24.17684");longitudeAndLatitudeMap.put("F23000305","114.21892,22.28145");longitudeAndLatitudeMap.put("F23000304","109.59403,19.50146");longitudeAndLatitudeMap.put("F23000303","121.06061,13.20827");longitudeAndLatitudeMap.put("F23000302","120.53530,-1.87089");longitudeAndLatitudeMap.put("F23000301","110.32614,-7.80236");longitudeAndLatitudeMap.put("F23000300","145.24253,-27.52135");longitudeAndLatitudeMap.put("F23000299","171.44028,-42.56513");longitudeAndLatitudeMap.put("F23000298","-67.43192,-39.55812");longitudeAndLatitudeMap.put("F23000297","-69.34524,6.59255");longitudeAndLatitudeMap.put("F23000296","24.99610,-13.70725");longitudeAndLatitudeMap.put("F23000295","35.00423,40.48366");longitudeAndLatitudeMap.put("F23000294","-4.58675,39.57853");longitudeAndLatitudeMap.put("F23000293","26.61506,54.09425");longitudeAndLatitudeMap.put("F23000292","14.84079,63.88942");longitudeAndLatitudeMap.put("F23000291","25.79382,68.38100");longitudeAndLatitudeMap.put("F23000290","101.88504,48.52852");longitudeAndLatitudeMap.put("F23000289","140.88731,39.30062");longitudeAndLatitudeMap.put("F23000288","104.09272,63.46892");longitudeAndLatitudeMap.put("F23000287","-154.23031,64.15251");longitudeAndLatitudeMap.put("F23000286","-135.03542,64.83030");longitudeAndLatitudeMap.put("F23000285","-107.31865,56.20248");longitudeAndLatitudeMap.put("F23000284","-105.49845,36.93292");longitudeAndLatitudeMap.put("F23000283","-101.11341,25.21778");longitudeAndLatitudeMap.put("F23000282","-76.95431,20.93201");longitudeAndLatitudeMap.put("F23000281","-52.92645,-9.07194");longitudeAndLatitudeMap.put("F23000280","-66.51627,-39.98271");longitudeAndLatitudeMap.put("F23000279","24.59100,-30.56490");longitudeAndLatitudeMap.put("F23000278","46.50112,-20.91024");longitudeAndLatitudeMap.put("F23000277","45.70816,22.30069");longitudeAndLatitudeMap.put("F23000276","2.16526,48.04232");longitudeAndLatitudeMap.put("F23000275","-18.35814,65.29671");longitudeAndLatitudeMap.put("F23000274","-111.55389,67.25068");longitudeAndLatitudeMap.put("F23000273","-145.80579,67.71462");longitudeAndLatitudeMap.put("F23000272","-75.49705,-10.04124");longitudeAndLatitudeMap.put("F23000271","-46.14453,-8.69899");longitudeAndLatitudeMap.put("F23000270","-64.17184,8.13274");longitudeAndLatitudeMap.put("F23000269","-71.33817,19.00195");longitudeAndLatitudeMap.put("F23000268","-70.92216,50.46753");longitudeAndLatitudeMap.put("F23000267","-112.52365,64.33525");longitudeAndLatitudeMap.put("F23000266","-131.10565,64.87117");longitudeAndLatitudeMap.put("F23000265","-156.62124,67.81950");longitudeAndLatitudeMap.put("F23000264","-77.36511,41.03387");longitudeAndLatitudeMap.put("F23000263","-63.33919,51.62100");longitudeAndLatitudeMap.put("F23000262","-74.58598,58.31921");longitudeAndLatitudeMap.put("F23000261","170.80659,-44.61839");longitudeAndLatitudeMap.put("F23000260","172.42555,-41.82046");longitudeAndLatitudeMap.put("F23000259","162.41742,-9.23632");longitudeAndLatitudeMap.put("F23000258","132.39303,-27.54709");longitudeAndLatitudeMap.put("F23000257","119.73569,-1.86773");longitudeAndLatitudeMap.put("F23000256","162.85895,-9.38253");longitudeAndLatitudeMap.put("F23000255","46.73522,-20.52389");longitudeAndLatitudeMap.put("F23000254","23.48103,-30.90318");longitudeAndLatitudeMap.put("F23000253","38.64040,1.39179");longitudeAndLatitudeMap.put("F23000252","7.74615,61.69806");longitudeAndLatitudeMap.put("F23000251","17.01839,63.78956");longitudeAndLatitudeMap.put("F23000250","6.71590,61.06262");longitudeAndLatitudeMap.put("F23000249","91.16632,29.61968");longitudeAndLatitudeMap.put("F23000248","77.14594,28.75387");longitudeAndLatitudeMap.put("F23000247","102.79804,24.84042");longitudeAndLatitudeMap.put("F23000246","106.66662,26.62768");longitudeAndLatitudeMap.put("F23000245","108.38023,22.83429");longitudeAndLatitudeMap.put("F23000244","102.51244,18.98029");longitudeAndLatitudeMap.put("F23000243","110.32750,19.79340");longitudeAndLatitudeMap.put("F23000242","113.49507,22.18291");longitudeAndLatitudeMap.put("F23000241","113.36525,23.21881");longitudeAndLatitudeMap.put("F23000240","106.66662,26.62768");longitudeAndLatitudeMap.put("F23000239","106.61470,29.57430");longitudeAndLatitudeMap.put("F23000238","104.12218,30.61289");longitudeAndLatitudeMap.put("F23000237","108.95143,34.39607");longitudeAndLatitudeMap.put("F23000236","112.94983,28.20326");longitudeAndLatitudeMap.put("F23000235","114.25002,30.54549");longitudeAndLatitudeMap.put("F23000234","115.85977,28.63940");longitudeAndLatitudeMap.put("F23000233","119.23505,26.06623");longitudeAndLatitudeMap.put("F23000232","120.16974,30.18524");longitudeAndLatitudeMap.put("F23000231","121.20828,31.08334");longitudeAndLatitudeMap.put("F23000230","118.81963,32.12774");longitudeAndLatitudeMap.put("F23000229","113.62690,34.78259");longitudeAndLatitudeMap.put("F23000228","117.15795,36.62495");longitudeAndLatitudeMap.put("F23000227","114.50966,38.03479");longitudeAndLatitudeMap.put("F23000226","112.48449,37.84973");longitudeAndLatitudeMap.put("F23000225","106.35707,38.46477");longitudeAndLatitudeMap.put("F23000224","103.89052,35.97340");longitudeAndLatitudeMap.put("F23000223","101.78747,36.62495");longitudeAndLatitudeMap.put("F23000222","111.80944,40.85259");longitudeAndLatitudeMap.put("F23000221","116.45694,39.91944");longitudeAndLatitudeMap.put("F23000220","117.26181,38.91248");longitudeAndLatitudeMap.put("F23000219","123.46712,41.67541");longitudeAndLatitudeMap.put("F23000218","125.36247,43.83718");longitudeAndLatitudeMap.put("F23000217","126.55680,45.81742");longitudeAndLatitudeMap.put("F23000216","128.89353,40.46079");longitudeAndLatitudeMap.put("F23000215","125.67403,39.66278");longitudeAndLatitudeMap.put("F23000214","126.99818,37.54410");longitudeAndLatitudeMap.put("F23000213","127.80305,35.76601");longitudeAndLatitudeMap.put("F23000212","130.84080,32.04344");longitudeAndLatitudeMap.put("F23000211","132.58037,34.59356");longitudeAndLatitudeMap.put("F23000210","138.21448,36.67069");longitudeAndLatitudeMap.put("F23000209","142.86401,43.20051");longitudeAndLatitudeMap.put("F23000208","19.86489,48.70907");longitudeAndLatitudeMap.put("F23000207","24.66590,55.06168");longitudeAndLatitudeMap.put("F23000206","9.16551,56.52972");longitudeAndLatitudeMap.put("F23000205","8.06814,46.95432");longitudeAndLatitudeMap.put("F23000204","43.86993,58.44976");longitudeAndLatitudeMap.put("F23000203","64.44567,55.60901");longitudeAndLatitudeMap.put("F23000202","3.39487,9.48019");longitudeAndLatitudeMap.put("F23000201","-69.16891,-24.86286");longitudeAndLatitudeMap.put("F23000200","-76.85052,-1.52400");longitudeAndLatitudeMap.put("F23000199","-71.22648,-49.64933");longitudeAndLatitudeMap.put("F23000198","-111.26497,29.81043");longitudeAndLatitudeMap.put("F23000197","-51.43862,64.14108");longitudeAndLatitudeMap.put("F23000196","-132.21363,63.47784");longitudeAndLatitudeMap.put("F23000195","-96.15024,56.17903");}@Asyncpublic void sendMsg() {while (true){try {
// long id = 1675673147956924417l;long dev_id = 23000009l;long time = System.currentTimeMillis()/1000;for(Map.Entry<String,String> obj: longitudeAndLatitudeMap.entrySet()){
// id = id-1;dev_id = dev_id+1;
// String devId = "F"+dev_id;String devId = obj.getKey();int curOpening = (int)(Math.random()*1000+1);Map map = new HashMap<>();map.put("devId",devId);map.put("type","devdata");map.put("time",time);map.put("msgid",time);Map dataMap = new HashMap<>();dataMap.put("workStat",1);dataMap.put("workDelta",1);dataMap.put("curState",4);dataMap.put("enableStat",1);dataMap.put("overTemperature",0);dataMap.put("generalFault",0);dataMap.put("ovTorFault",0);dataMap.put("stValFault",0);dataMap.put("opnValOvTor",0);dataMap.put("clsValOvTor",0);dataMap.put("phOutFault",0);dataMap.put("curMoForce",0);dataMap.put("generalFault",0);dataMap.put("curOpening",curOpening);dataMap.put("anlgFeedback",0);dataMap.put("ctlPara",0);dataMap.put("prodId","F101MO230616000103030003");dataMap.put("prodModel","EMD20-72-1-eS1-T3WCG4-X");dataMap.put("proDate","20230616");dataMap.put("ratedMoment",170);dataMap.put("netsignal",5);dataMap.put("wifiSignal",0);dataMap.put("btStatus",0);dataMap.put("firmwareVer","1.0.6");dataMap.put("devId",devId);map.put("data",dataMap);JSONObject json = JSONObject.fromObject(map);String sendMsg = json.toString();myMQTTClient.publish(sendMsg, topic1);}
// 5秒钟发送一次Thread.sleep(1000*5);} catch (InterruptedException e) {e.printStackTrace();}}}
}
SpringUtils.java
package cn.renkai721.mqtt;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;/*** spring工具类 方便在非spring管理环境中获取bean** @author renkai721@163.com*/
@Component
public final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware
{/** Spring应用上下文环境 */private static ConfigurableListableBeanFactory beanFactory;private static ApplicationContext applicationContext;@Overridepublic void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException{SpringUtils.beanFactory = beanFactory;}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException{SpringUtils.applicationContext = applicationContext;}/*** 获取对象**/@SuppressWarnings("unchecked")public static <T> T getBean(String name) throws BeansException{return (T) beanFactory.getBean(name);}/*** 获取类型为requiredType的对象***/public static <T> T getBean(Class<T> clz) throws BeansException{T result = (T) beanFactory.getBean(clz);return result;}/*** 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true*/public static boolean containsBean(String name){return beanFactory.containsBean(name);}/*** 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）***/public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException{return beanFactory.isSingleton(name);}/*** @param name**/public static Class<?> getType(String name) throws NoSuchBeanDefinitionException{return beanFactory.getType(name);}/*** 如果给定的bean名字在bean定义中有别名，则返回这些别名**/public static String[] getAliases(String name) throws NoSuchBeanDefinitionException{return beanFactory.getAliases(name);}/*** 获取aop代理对象* @return*/@SuppressWarnings("unchecked")public static <T> T getAopProxy(T invoker){return (T) AopContext.currentProxy();}/*** 获取当前的环境配置，无配置返回null*/public static String[] getActiveProfiles(){return applicationContext.getEnvironment().getActiveProfiles();}/*** 获取当前的环境配置，当有多个环境配置时，只获取第一个*/public static String getActiveProfile(){final String[] activeProfiles = getActiveProfiles();return activeProfiles.length > 0 ? activeProfiles[0] : null;}
}
发送消息
程序启动后，默认只能连接MQTT，然后接收消息。发送消息是通过接口来触发的，所以我们需要手动调用一下接口。接口触发后在控制台就可以看见发送的日志和接收到的日志了。当然直接通过MQTTX客户端订阅消息，也可以看到发送的消息。
控制台日志
