MQTT服务器搭建通信及保存文件到MySql数据库
Posted p935074243
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了MQTT服务器搭建通信及保存文件到MySql数据库相关的知识，希望对你有一定的参考价值。
步骤
本文使用Mybatis，如果你不会，那就照猫画虎吧
1、搭建数据库
CREATE DATABASE `iot`;
USE `iot`;
DROP TABLE IF EXISTS `messages`;
CREATE TABLE `messages` (
`id` INT(10) NOT NULL AUTO_INCREMENT COMMENT ‘信息id‘,
`topic` VARCHAR(100) NOT NULL COMMENT ‘topic‘,
`temperature` int(3) NOT NULL COMMENT ‘温度‘,
`himidity` int(3) NOT NULL COMMENT ‘湿度‘,
`PM2_5` int(3) NOT NULL COMMENT ‘PM2_5‘,
`fan` int(1) NOT NULL COMMENT ‘风扇状态‘,
`light` int(1) NOT NULL COMMENT ‘灯光状态‘,
`power` int(1) NOT NULL COMMENT ‘机器状态‘,
`msg` VARCHAR(1000) NOT NULL COMMENT ‘消息‘,
primary key(`id`)
) ENGINE=INNODB DEFAULT CHARSET=utf8;
2、连接数据库
然后在IDEA边框上会有一个Database，点开它
3、IDEA新建一个普通的maven项目，在pom.xml中导入maven依赖
如果你的maven不能自动导入，建议去看看maven配置
<?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>
<groupId>org.example</groupId>
<artifactId>MyBatis-study</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<!--导入依赖-->
<dependencies>
<!--mysql驱动-->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.47</version>
</dependency>
<!--mybatis-->
<dependency>
<groupId>org.mybatis</groupId>
<artifactId>mybatis</artifactId>
<version>3.5.2</version>
</dependency>
<!--junit-->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.eclipse.paho/org.eclipse.paho.client.mqttv3 -->
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.4</version>
</dependency>
</dependencies>
<!--build中配置resources，来放置我们资源导出失败的问题-->
<build>
<resources>
<resource>
<directory>src/main/resource</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>true</filtering>
</resource>
<resource>
<directory>src/main/java</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>true</filtering>
</resource>
</resources>
</build>
</project>
4、创建mybatis的核心配置文件mybatis-config.xml
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration
PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
<settings>
<setting name="logImpl" value="STDOUT_LOGGING"/>
</settings>
<!--配置数据源，交给Spring去做-->
<typeAliases>
<package name="com.iot.pojo"/>
</typeAliases>
<environments default="development">
<environment id="development">
<transactionManager type="JDBC"/><!--事务管理-->
<dataSource type="POOLED">
<property name="driver" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://localhost:3306/iot?useSSL=true&useUnicode=true&characterEncoding=UTF-8"/>
<property name="username" value="root"/>
<property name="password" value="root"/>
</dataSource>
</environment>
</environments>
<mappers>
<mapper class="com.iot.dao.MessagesMapper"/>
</mappers>
</configuration>
5、编写mybatis工具类MyBatisUtils.java
package com.iot.utils;
import org.apache.ibatis.io.Resources;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.ibatis.session.SqlSessionFactoryBuilder;
import java.io.IOException;
import java.io.InputStream;
//SqlSessionFactory --> SqlSession
public class MyBatisUtils {
private static SqlSessionFactory sqlSessionFactory;
static {
try {
//使用mybatis第一步，获取SqlSessionFactory对象
String resource = "mybatis-config.xml";
InputStream inputStream = Resources.getResourceAsStream(resource);
sqlSessionFactory = new SqlSessionFactoryBuilder().build(inputStream);
} catch (IOException e) {
e.printStackTrace();
}
}
//既然有了 SqlSessionFactory，顾名思义，我们可以从中获得 SqlSession 的实例。
// SqlSession 提供了在数据库执行 SQL 命令所需的所有方法。
// 你可以通过 SqlSession 实例来直接执行已映射的 SQL 语句。
public static SqlSession getSqlSession(){
return sqlSessionFactory.openSession();
}
}
6、创建实体类Messages.java
package com.iot.pojo;
public class Messages {
private int id;
private String topic;
private int temperature;
private int himidity;
private int PM2_5;
private int fan;
private int light;
private int power;
private String msg;
public Messages(int id, String topic, int temperature, int himidity, int PM2_5, int fan, int light, int power, String msg) {
this.id = id;
this.topic = topic;
this.temperature = temperature;
this.himidity = himidity;
this.PM2_5 = PM2_5;
this.fan = fan;
this.light = light;
this.power = power;
this.msg = msg;
}
}
7、创建Dao接口MessageMapper.java
package com.iot.dao;
import com.iot.pojo.Messages;
public interface MessagesMapper {
//添加订阅信息到库
public int addMes(Messages messages);
//获取数据库中数据的数量
public int getCount();
}
8、创建实现接口的配置文件MessageMapper.xml
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.iot.dao.MessagesMapper">
<insert id="addMes" parameterType="Messages">
insert into iot.messages(id, topic, temperature, himidity, PM2_5, fan, light, power, msg)
VALUES(#{id},#{topic},#{temperature},#{himidity},#{PM2_5},#{fan},#{light},#{power},#{msg});
</insert>
<select id="getCount" resultType="int">
select count(*) as result from iot.messages;
</select>
</mapper>
9、创建Client.java、Server.java、PushCallback.java
Client.java
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class Client {
public static final String HOST = "tcp://127.0.0.1:61613";
public static final String TOPIC = "IOT";
private static final String clientid = "client";
private MqttClient client;
private MqttConnectOptions options;
private String userName = "admin";
private String passWord = "password";
private void start() {
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(userName);
// 设置连接的密码
options.setPassword(passWord.toCharArray());
// 设置超时时间 单位为秒
options.setConnectionTimeout(10);
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
options.setKeepAliveInterval(20);
// 设置回调
client.setCallback(new PushCallback());
MqttTopic topic = client.getTopic(TOPIC);
//setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
options.setWill(topic, "close".getBytes(), 2, true);
client.connect(options);
//订阅消息
int[] Qos = {1};
String[] topic1 = {TOPIC};
client.subscribe(topic1, Qos);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws MqttException {
Client client = new Client();
client.start();
}
}
Server.java
import com.iot.dao.MessagesMapper;
import com.iot.pojo.Messages;
import com.iot.utils.MyBatisUtils;
import org.apache.ibatis.session.SqlSession;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import java.awt.*;
import java.util.Scanner;
public class Server {
public static final String HOST = "tcp://127.0.0.1:61613";
public static final String TOPIC = "IOT";
private static final String clientid = "server";
//各参数阈值
private static final int temperatureMax = 50;
private static final int temperatureMin = 0;
private static final int temperatureStaVal = 28;
private static final int temperatureDifVal = 12;
private static final int himidityMax = 100;
private static final int himidityMin = 0;
private static final int himidityStaVal = 40;
private static final int himidityDifVal = 20;
private static final int PM2_5Max = 700;
private static final int PM2_5Min = 0;
private static final int PM2_5StaVal = 300;
private static final int PM2_5DifVal = 200;
private static int fan;
private static int power;
private static int light;
private static int temperatureCur;
private static int himidityCur;
private static int PM2_5Cur;
private MqttClient client;
private MqttTopic iot;
private String userName = "admin";
private String passWord = "password";
private MqttMessage message;
public Server() throws MqttException {
// MemoryPersistence设置clientid的保存形式，默认为以内存保存
client = new MqttClient(HOST, clientid, new MemoryPersistence());
connect();
}
private void connect() {
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
options.setUserName(userName);
options.setPassword(passWord.toCharArray());
// 设置超时时间
options.setConnectionTimeout(10);
// 设置会话心跳时间
options.setKeepAliveInterval(20);
try {
client.setCallback(new PushCallback());
client.connect(options);
iot = client.getTopic(TOPIC);
} catch (Exception e) {
e.printStackTrace();
}
}
public void publish(MqttTopic topic, MqttMessage message) throws MqttPersistenceException,
MqttException {
MqttDeliveryToken token = topic.publish(message);
token.waitForCompletion();
System.out.println("message is published completely! " + token.isComplete());
}
public static int getRandom(int max, int min) {
return (int) (Math.random() * (max - min) + min);
}
public static String[] getState(String project, int curVal, int staVal, int difVal) {
String[] msg = new String[2];
if (curVal <= staVal) {
msg[0] = "当前" + project + "：" + curVal + "
";
msg[1] = "0";
} else if (curVal <= staVal + difVal) {
msg[0] = "当前" + project + "：" + curVal + "
";
msg[1] = "10";
} else {
msg[0] = "当前" + project + "：" + curVal + "
";
msg[1] = "100";
}
return msg;
}
public static String getVal() {
temperatureCur = getRandom(temperatureMax, temperatureMin);
himidityCur = getRandom(himidityMax, himidityMin);
PM2_5Cur = getRandom(PM2_5Max, PM2_5Min);
String[] msg1 = getState("temperature", temperatureCur, temperatureStaVal, temperatureDifVal);
String[] msg2 = getState("himidity", himidityCur, himidityStaVal, himidityDifVal);
String[] msg3 = getState("PM2_5", PM2_5Cur, PM2_5StaVal, PM2_5DifVal);
int flag = Integer.parseInt(msg1[1]) + Integer.parseInt(msg2[1]) + Integer.parseInt(msg3[1]);
String ele;
if (flag < 10) {
fan = 0;
light = 0;
power = 1;
ele = "适宜工作！
";
} else if (flag < 100) {
fan = 1;
light = 1;
power = 1;
ele = "已经高于标准值，请注意！已开启风扇" + fan + "级，灯光" + light + "级
";
} else {
fan = 2;
if (msg3[1].equals("100")) {
light = 2;
} else
light = 1;
power = 0;
ele = "不适工作，请速离开！已关闭机器，开启风扇" + fan + "级，灯光" + light + "级
";
}
String sep = "------------消息体---------------
";
String sep2 = "-------------------------------";
return sep + msg1[0] + msg2[0] + msg3[0] + ele + sep2;
}
public static void main(String[] args) throws MqttException, AWTException {
Server server = new Server();
Scanner scanner = new Scanner(System.in);
SqlSession sqlSession= MyBatisUtils.getSqlSession();
MessagesMapper messagesMapper=sqlSession.getMapper(MessagesMapper.class);
int k=messagesMapper.getCount()+1;
do {
String msg = getVal();
server.message = new MqttMessage();
server.message.setQos(2);
server.message.setRetained(true);
server.message.setPayload(("推送" + TOPIC + "的信息
" + msg).getBytes());
server.publish(server.iot, server.message);
Messages messages=new Messages(k,TOPIC,temperatureCur,himidityCur,PM2_5Cur,fan,light,power,msg);
System.out.println(messages);
//保存数据到数据库
int i = messagesMapper.addMes(messages);
if(i>0){
System.out.println("保存成功！");
}else{
System.out.println("保存失败！");
}
sqlSession.commit();
System.out.println(server.message.isRetained() + "------ratained状态");
k++;
} while (scanner.nextInt() == 0);
sqlSession.close();
System.out.println("已停止发送！");
}
}
PushCallback.java
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
/**
* 发布消息的回调类
*
* 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
* 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
* 在回调中，将它用来标识已经启动了该回调的哪个实例。
* 必须在回调类中实现三个方法：
*
* public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
*
* public void connectionLost(Throwable cause)在断开连接时调用。
*
* public void deliveryComplete(MqttDeliveryToken token))
* 接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。
* 由 MqttClient.connect 激活此回调。
*
*/
public class PushCallback implements MqttCallback {
public void connectionLost(Throwable cause) {
// 连接丢失后，一般在这里面进行重连
System.out.println("连接断开，可以做重连");
}
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
public void messageArrived(String topic, MqttMessage message) throws Exception {
// subscribe后得到的消息会执行到这里面
System.out.println("===================================================");
System.out.println("接收消息主题 : " + topic);
System.out.println("接收消息Qos : " + message.getQos());
System.out.println("接收消息内容 : " + new String(message.getPayload(), "UTF-8"));
System.out.println("===================================================");
}
}
10、点击启动测试
注意：当Server启动的时候，你可以输入0实现再次发布随机消息
以上是关于MQTT服务器搭建通信及保存文件到MySql数据库的主要内容，如果未能解决你的问题，请参考以下文章