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.com
2. @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;
@Bean
public 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 {
@Autowired
private MyMQTTClient myMQTTClient;
@Autowired
private MyMqttService myMqttService;
@Value("${mqtt.topic1}")
String topic1;
@PostMapping("/sendMsg")
@ResponseBody
public 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;
}
@Override
public 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
*/
@Override
public 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后得到的消息会执行到这里面
*/
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
log.info("接收消息主题 : {}，接收消息内容 : {}", topic, new String(mqttMessage.getPayload()));
//你自己的业务接口
}
/**
*连接成功后的回调 可以在这个方法执行 订阅主题 生成Bean的 MqttConfiguration方法中订阅主题 出现bug
*重新连接后 主题也需要再次订阅 将重新订阅主题放在连接成功后的回调 比较合理
* @param reconnect
* @param serverURI
*/
@Override
public void connectComplete(boolean reconnect,String serverURI){
log.info("MQTT 连接成功，连接方式：{}",reconnect?"重连":"直连");
//订阅主题
myMQTTClient.subscribe(mqttConfiguration.topic1, 1);
}
/**
* 消息到达后
* subscribe后，执行的回调函数
* publish后，配送完成后回调的方法
*
*
*/
@Override
public 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.com
2. @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.com
2. @date 2023/07/17 16:40
*/
@Service
@EnableAsync
@Slf4j
public class MyMqttService {
@Autowired
private 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");
}
@Async
public 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;
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException
{
SpringUtils.beanFactory = beanFactory;
}
@Override
public 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客户端订阅消息，也可以看到发送的消息。
控制台日志
![[MySql]JDBC编程](https://img-blog.csdnimg.cn/3761fcb922ee49ecaa14063c05dbb149.png)
