SpringBoot整合MQTT最详细版（亲测有效）

一、导入pom.xml依赖!--mqtt依赖-- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-integration/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-stream/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency二、配置MQTT相关信息到application.ymlspring: mqtt: username: 你的账号 # 账号 password: 你的密码 # 密码 hostUrl: tcp://127.0.0.1:1883 # mqtt连接tcp地址 clientid: ${random.value} # 客户端Id不能相同采用随机数 ${random.value} default-topic: /testtopic/# # 默认主题 timeout: 3000 # 超时时间 keepalive: 600 # 保持连接 subscribeFlag: true #是否进行订阅true或者false enabled: true # 是否使用mqtt功能三、在项目中创建mqtt文件夹后添加AjaxResult.java、MqttConfig.java、MqttInit.java、MqttPushClient.java、PushCallback.java如图AjaxResult.java代码public class AjaxResult extends HashMapString, Object { private static final long serialVersionUID 1L; /** 状态码 */ public static final String CODE_TAG code; /** 返回内容 */ public static final String MSG_TAG msg; /** 数据对象 */ public static final String DATA_TAG data; /** * 初始化一个新创建的 AjaxResult 对象使其表示一个空消息。 */ public AjaxResult() { } /** * 初始化一个新创建的 AjaxResult 对象 * * param code 状态码 * param msg 返回内容 */ public AjaxResult(int code, String msg) { super.put(CODE_TAG, code); super.put(MSG_TAG, msg); } /** * 初始化一个新创建的 AjaxResult 对象 * * param code 状态码 * param msg 返回内容 * param data 数据对象 */ public AjaxResult(int code, String msg, Object data) { super.put(CODE_TAG, code); super.put(MSG_TAG, msg); if (StringUtils.isNotNull(data)) { super.put(DATA_TAG, data); } } /** * 返回成功消息 * * return 成功消息 */ public static AjaxResult success() { return AjaxResult.success(操作成功); } /** * 返回成功数据 * * return 成功消息 */ public static AjaxResult success(Object data) { return AjaxResult.success(操作成功, data); } /** * 返回成功消息 * * param msg 返回内容 * return 成功消息 */ public static AjaxResult success(String msg) { return AjaxResult.success(msg, null); } /** * 返回成功消息 * * param msg 返回内容 * param data 数据对象 * return 成功消息 */ public static AjaxResult success(String msg, Object data) { return new AjaxResult(HttpStatus.SUCCESS, msg, data); } /** * 返回警告消息 * * param msg 返回内容 * return 警告消息 */ public static AjaxResult warn(String msg) { return AjaxResult.warn(msg, null); } /** * 返回警告消息 * * param msg 返回内容 * param data 数据对象 * return 警告消息 */ public static AjaxResult warn(String msg, Object data) { return new AjaxResult(HttpStatus.WARN, msg, data); } /** * 返回错误消息 * * return 错误消息 */ public static AjaxResult error() { return AjaxResult.error(操作失败); } /** * 返回错误消息 * * param msg 返回内容 * return 错误消息 */ public static AjaxResult error(String msg) { return AjaxResult.error(msg, null); } /** * 返回错误消息 * * param msg 返回内容 * param data 数据对象 * return 错误消息 */ public static AjaxResult error(String msg, Object data) { return new AjaxResult(HttpStatus.ERROR, msg, data); } /** * 返回错误消息 * * param code 状态码 * param msg 返回内容 * return 错误消息 */ public static AjaxResult error(int code, String msg) { return new AjaxResult(code, msg, null); } /** * 是否为成功消息 * * return 结果 */ public boolean isSuccess() { return Objects.equals(HttpStatus.SUCCESS, this.get(CODE_TAG)); } /** * 是否为警告消息 * * return 结果 */ public boolean isWarn() { return Objects.equals(HttpStatus.WARN, this.get(CODE_TAG)); } /** * 是否为错误消息 * * return 结果 */ public boolean isError() { return Objects.equals(HttpStatus.ERROR, this.get(CODE_TAG)); } /** * 方便链式调用 * * param key 键 * param value 值 * return 数据对象 */ Override public AjaxResult put(String key, Object value) { super.put(key, value); return this; } }MqttConfig.java代码Component ConfigurationProperties(spring.mqtt) public class MqttConfig { Autowired private MqttPushClient mqttPushClient; /** * 用户名 */ private String username; /** * 密码 */ private String password; /** * 连接地址 */ private String hostUrl; /** * 客户Id */ private String clientId; /** * 默认连接话题 */ private String defaultTopic; /** * 超时时间 */ private int timeout; /** * 保持连接数 */ private int keepalive; /** * mqtt功能使能 */ private boolean enabled; public String getUsername() { return username; } public void setUsername(String username) { this.username username; } public String getPassword() { return password; } public void setPassword(String password) { this.password password; } public String getHostUrl() { return hostUrl; } public void setHostUrl(String hostUrl) { this.hostUrl hostUrl; } public String getClientId() { return clientId; } public void setClientId(String clientId) { this.clientId clientId; } public String getDefaultTopic() { return defaultTopic; } public void setDefaultTopic(String defaultTopic) { this.defaultTopic defaultTopic; } public int getTimeout() { return timeout; } public void setTimeout(int timeout) { this.timeout timeout; } public int getKeepalive() { return keepalive; } public void setKeepalive(int keepalive) { this.keepalive keepalive; } public boolean isEnabled() { return enabled; } public void setEnabled(boolean enabled) { this.enabled enabled; } //创建MQTT客户端 public MqttPushClient getMqttPushClient() { if(enabled true){ String mqtt_topic[] StringUtils.split(defaultTopic, ,); System.out.println(开始连接clientId); //连接 mqttPushClient.connect(hostUrl, clientId, username, password, timeout, keepalive); System.out.println(开始订阅); for(int i0; imqtt_topic.length; i){ //订阅主题 mqttPushClient.subscribe(mqtt_topic[i], 1); } } return mqttPushClient; } }MqttInit.java代码Component public class MqttInit implements ApplicationRunner { Autowired private MqttConfig mqttConfig; /** * 初始化客户端用于接收生产者发过来的消息项目运行就会创建好 * param args * throws Exception */ Override public void run(ApplicationArguments args) throws Exception{ mqttConfig.getMqttPushClient(); } }MqttPushClient.java代码Component Order(value 2) 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 保留数 * param callback 是否回调 */ public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) { System.out.println(进入连接----------------clientID); 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); System.out.println(连接成功clientID); } catch (Exception e) { e.printStackTrace(); } } catch (Exception e) { e.printStackTrace(); } } /** * 发布 * * param qos 连接方式 * QoS 0最多一次消息发送后不进行确认也不重试是最低的服务质量等级。这种方式可能会导致消息丢失但传输效率最高。 * QoS 1至少一次确保消息至少被送达一次。如果发送方没有收到确认它可能会重试发送消息这可能导致消息重复。 * QoS 2恰好一次保证消息准确无误地送达一次不丢失也不重复是最高的服务质量等级但相应地增加了通信的开销。 * param retained 是否保留 * param topic 主题 * param pushMessage 消息体 */ public AjaxResult 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(); return AjaxResult.success(); } catch (MqttPersistenceException e) { e.printStackTrace(); return AjaxResult.error(); } catch (MqttException e) { e.printStackTrace(); return AjaxResult.error(); } } /** * 订阅某个主题 * * 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(); } } }PushCallback.java代码Component Order(value 1) public class PushCallback implements MqttCallback { private static final Logger logger LoggerFactory.getLogger(MqttPushClient.class); Autowired private MqttConfig mqttConfig; private static MqttClient client; private static String _topic; private static String _qos; private static String _msg; Override public void connectionLost(Throwable throwable) { System.out.println(重连); // 连接丢失后一般在这里面进行重连 logger.info(连接断开可以做重连); boolean flagtrue; while(flag) { try { Thread.sleep(5000); if (client null || !client.isConnected()) { mqttConfig.getMqttPushClient(); } flagfalse; } catch (Exception e) { e.printStackTrace(); } } } Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { // subscribe后得到的消息会执行到这里面 logger.info(接收消息主题 : topic); logger.info(接收消息Qos : mqttMessage.getQos()); logger.info(接收消息内容 : new String(mqttMessage.getPayload())); System.out.println(接收消息主题 : topic); System.out.println(接收消息Qos : mqttMessage.getQos()); System.out.println(接收消息内容 : new String(mqttMessage.getPayload())); _topic topic; _qos mqttMessage.getQos(); _msg new String(mqttMessage.getPayload()); } /** * 参数回调的方法 * param iMqttDeliveryToken */ Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { logger.info(deliveryComplete--------- iMqttDeliveryToken.isComplete()); } //别的Controller层会调用这个方法来 获取 接收到的硬件数据 public String receive() { JSONObject jsonObject new JSONObject(); jsonObject.put(topic, _topic); jsonObject.put(qos, _qos); jsonObject.put(msg, _msg); return jsonObject.toString(); } }最后整合结束四、测试创建mttqController控制器发送消息如图mttqController代码RestController RequestMapping(/mttpTest) public class mttqController extends BaseController { Autowired private MqttConfig mqttConfig; Autowired private MqttPushClient mqttPushClient; GetMapping(value /a) public AjaxResult a() { JSONObject jsonObject new JSONObject(); jsonObject.put(hello, 你好); jsonObject.put(msg, 成功); AjaxResult publish mqttPushClient.publish(2,false,/testtopic/5,jsonObject.toString()); return success(null); } }浏览器输入接口地址调试发送成功使用MQTTX客户端工具查看消息成功被发送springboot控制台也就是PushCallback.java里的回调方法messageArrived也成功打印出了客户端接收到的消息如图