海为物联云HMI如何设置MQTT?

发布网友发布时间：2022-04-24 14:51

共2个回答

懂视网时间：2022-05-03 14:49

步骤

本文使用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数据库

标签：default 在线 tca junit equals exist parameter repo logging

热心网友时间：2022-05-03 11:57

首先需要搭建MQTT服务器，然后搭建MySQL数据库，然后使用海为组态写段程序，最后配置客户端验证即可。具体可以参考内容 Haiwell（海为）HMI/CBOX/IPC MQTT 配置应用教程网页链接