Springboot集成mqtt【demo】
程序员文章站
2022-07-15 08:09:47
...
项目目录如下
启动类如下
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@EnableAutoConfiguration(exclude = {
org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration.class
})
@SpringBootApplication
public class MqttApplication {
public static void main(String[] args) throws MqttException {
SpringApplication.run(MqttApplication.class, args);
//订阅主题,之后控制台打印消息。证明整合成功
// MqttPushClient.getInstance().subscribe("topic1");
}
}
MqttController
package com.example.mqtt.mqtt;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
/**
* MQTT消息发送
*/
@RestController
public class MqttController {
/**
* 发送MQTT消息
*
* @param message 消息内容
* @return 返回
*/
@ResponseBody
@GetMapping(value = "/mqtt")
public ResponseEntity<String> sendMqtt(@RequestParam(value = "msg") String message) throws MqttException {
String kdTopic = "topic1";
MqttPushClient.getInstance().publish(kdTopic, "稍微来点鸡血");
return new ResponseEntity<>("OK", HttpStatus.OK);
}
}
MqttPushClient
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Slf4j
@Configuration
public class MqttPushClient {
private String url = "tcp://127.0.0.1:61613";
private String clientId = "mqttProducer";
@Autowired
MqttConfig config;
private static final byte[] WILL_DATA;
static {
WILL_DATA = "offline".getBytes();
}
private static volatile MqttPushClient mqttPushClient = null;
@Bean
public static MqttPushClient getInstance() throws MqttException {
if (null == mqttPushClient) {
synchronized (MqttPushClient.class) {
if (null == mqttPushClient) {
mqttPushClient = new MqttPushClient();
}
}
}
return mqttPushClient;
}
public MqttPushClient() throws MqttException {
connect();
}
private MqttClient client;
private void connect() throws MqttException {
try {
client = new MqttClient(url, clientId, new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录,
// 这里设置为true表示每次连接到服务器都以新的身份连接
options.setCleanSession(true);
// 设置连接的用户名
options.setUserName("admin");
// 设置连接的密码
options.setPassword("password".toCharArray());
options.setServerURIs(StringUtils.split("tcp://127.0.0.1:61613", ","));
// 设置超时时间 单位为秒
options.setConnectionTimeout(100);
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线,但这个方法并没有重连的机制
options.setKeepAliveInterval(20);
// 设置“遗嘱”消息的话题,若客户端与服务器之间的连接意外中断,服务器将发布客户端的“遗嘱”消息。
options.setWill("willTopic", WILL_DATA, 2, false);
client.setCallback(new PushCallback());
client.connect(options);
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 发布,默认qos为0,非持久化
*
* @param topic
* @param pushMessage
*/
public void publish(String topic, String pushMessage) {
publish(0, false, topic, pushMessage);
}
/**
* 发布
*
* @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 = client.getTopic(topic);
if (null == mTopic) {
log.error("topic not exist");
}
MqttDeliveryToken token;
try {
token = mTopic.publish(message);
token.waitForCompletion();
} catch (MqttPersistenceException e) {
e.printStackTrace();
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 订阅某个主题,qos默认为0
*
* @param topic
*/
public void subscribe(String topic) {
subscribe(topic, 0);
}
/**
* 订阅某个主题
*
* @param topic
* @param qos
*/
public void subscribe(String topic, int qos) {
try {
client.subscribe(topic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
PushCallback
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
public class PushCallback implements MqttCallback {
@Override
public void connectionLost(Throwable cause) {
System.out.println("连接断开,可以做重连");
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
System.out.println("接收消息主题 : " + topic);
System.out.println("接收消息Qos : " + message.getQos());
System.out.println("接收消息内容 : " + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println(token);
try {
System.out.println(token.getMessage());
} catch (MqttException e) {
e.printStackTrace();
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
pom
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.7.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>mqtt</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>mqtt</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--mqtt 相关依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
<!--<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.0</version>
</dependency>-->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.16.14</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.integration/spring-integration-jmx -->
<!-- https://mvnrepository.com/artifact/org.springframework/spring-jmx -->
<!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-security</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
package com.example.mqtt.mqtt;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
@Slf4j
@Component
public class ReportMqtt implements MqttCallback {
@Value("${mqtt.url}")
public String HOST;
@Value("${mqtt.consumer.defaultTopic}")
public String TOPIC;
@Value("${mqtt.username}")
private String name;
@Value("${mqtt.password}")
private String passWord;
private MqttClient client;
private MqttConnectOptions options;
//clientId不能重复所以这里我设置为系统时间
String clientid = String.valueOf(System.currentTimeMillis());
@PostConstruct
public void result() {
try {
// host为主机名,clientid即连接MQTT的客户端ID,一般以唯一标识符表示,MemoryPersistence设置clientid的保存形式,默认为以内存保存
client = new MqttClient(HOST, clientid, new MemoryPersistence());
// MQTT的连接设置
options = new MqttConnectOptions();
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录,这里设置为true表示每次连接到服务器都以新的身份连接
options.setCleanSession(true);
// 设置连接的用户名
options.setUserName(name);
// 设置连接的密码
options.setPassword(passWord.toCharArray());
// 设置超时时间 单位为秒
options.setConnectionTimeout(10);
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线,但这个方法并没有重连的机制
options.setKeepAliveInterval(3600);
// 设置回调
client.setCallback(this);
client.connect(options);
//订阅消息
int[] Qos = {1};
String[] topic1 = {TOPIC};
client.subscribe(topic1, Qos);
} catch (Exception e) {
log.info("ReportMqtt客户端连接异常,异常信息:" + e);
}
}
@Override
public void connectionLost(Throwable throwable) {
try {
log.info("程序出现异常,DReportMqtt断线!正在重新连接...:");
client.close();
this.result();
log.info("ReportMqtt重新连接成功");
} catch (MqttException e) {
log.info(e.getMessage());
}
}
@Override
public void messageArrived(String topic, MqttMessage message) {
log.info("接收消息主题:" + topic);
log.info("接收消息Qos:" + message.getQos());
log.info("接收消息内容 :" + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
log.info("消息发送成功");
}
}
配置文件
# src/main/resources/config/mqtt.properties
##################
# MQTT 配置
##################
# 用户名
mqtt.username=admin
# 密码
mqtt.password=password
# 推送信息的连接地址,如果有多个,用逗号隔开,如:tcp://127.0.0.1:61613,tcp://192.168.1.61:61613
mqtt.url=tcp://127.0.0.1:61613
##################
# MQTT 生产者
# 连接服务器默认客户端ID
mqtt.producer.clientId=mqttProducer
# 默认的推送主题,实际可在调用接口时指定
mqtt.producer.defaultTopic=topic1
##################
# MQTT 消费者
##################
# 连接服务器默认客户端ID
mqtt.consumer.clientId=mqttConsumer
# 默认的接收主题,可以订阅多个Topic,逗号分隔
mqtt.consumer.defaultTopic=topic1
运行截图
项目中遇到如下问题
加入spring权限jar包即可
mqtt后台地址
http://127.0.0.1:61680/console/index.html#
然后启动项目看下结果,是否有producer
访问项目url,地址如下 http://localhost:8080/mqtt?msg=“来点鸡血”
后续完善,许多漏洞。
上一篇: mqtt 使用
下一篇: Apache服务器安装
推荐阅读
-
SpringBoot集成JWT实现权限认证
-
五分钟后,你将学会在SpringBoot项目中如何集成CAT调用链
-
SpringBoot集成WebSocket长连接实际应用详解
-
Springboot 2.0.x 集成基于Centos7的Redis集群安装及配置
-
小白的springboot之路(十二)、集成log4j2日志
-
springcloud Springboot vue.js Activiti6 前后分离 跨域 工作流 集成代码生成器 shiro权限
-
Shiro Demo:SpringBoot+Shiro+Druid+MyBatis
-
Springboot Activiti6 工作流 集成代码生成器 shiro 权限 vue.js html 跨域 前后分离
-
springboot集成mybatisplus的方法
-
SpringBoot (1) idea下的环境搭建及demo