sunpeng
/
dji_demo
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

大疆demo提交

This commit is contained in:
陈璐 2025-04-24 09:29:10 +08:00
commit cd8c43d016
41 changed files with 5280 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

137
pom.xml Normal file
View File

@ -0,0 +1,137 @@
<?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 https://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>3.0.4</version>--> <!-- jdk17使用 -->
<version>2.7.12</version> <!-- jdk11使用 -->
<relativePath/> <!-- lookup parent from repository -->
</parent>
<artifactId>demo-airport</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>demo-airport</name>
<description>demo-airport</description>
<properties>
<java.version>11</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<!-- Redis 起始依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
<version>2.0.4.RELEASE</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>2.5.2</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
</dependency>
<dependency>
<groupId>com.thoughtworks.xstream</groupId>
<artifactId>xstream</artifactId>
<version>1.4.20</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-compress</artifactId>
<version>1.26.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<!--mqtt相关依赖-->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>sts20150401</artifactId>
<version>1.1.4</version>
</dependency>
<!-- https://mvnrepository.com/artifact/log4j/log4j -->
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
</dependency>
<dependency>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
<version>2.0.9</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

17
src/main/java/com/example/math/DemoAirportApplication.java Normal file
View File

@ -0,0 +1,17 @@
package com.example.math;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication(scanBasePackages = {"com.example.*"})
public class DemoAirportApplication {
public static void main(String[] args) {
SpringApplication.run(DemoAirportApplication.class, args);
System.out.println("demo 启动成功~");
}
}

186
src/main/java/com/example/math/config/RedisConfig.java Normal file
View File

@ -0,0 +1,186 @@
package com.example.math.config;
import com.example.math.utils.RedisUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
@Configuration
//@PropertySource("classpath:redis.properties")
public class RedisConfig {
@Value("${spring.redis.host}")
private String hostName;
@Value("${spring.redis.port}")
private Integer port;
@Value("${spring.redis.password}")
private String password;
@Value("${spring.redis.timeout}")
private Integer timeout;
/*
@Value("${redis.maxIdle}")
private Integer maxIdle;
@Value("${redis.maxTotal}")
private Integer maxTotal;
@Value("${redis.maxWaitMillis}")
private Integer maxWaitMillis;
@Value("${redis.minEvictableIdleTimeMillis}")
private Integer minEvictableIdleTimeMillis;
@Value("${redis.numTestsPerEvictionRun}")
private Integer numTestsPerEvictionRun;
@Value("${redis.timeBetweenEvictionRunsMillis}")
private long timeBetweenEvictionRunsMillis;
@Value("${redis.testOnBorrow}")
private boolean testOnBorrow;
@Value("${redis.testWhileIdle}")
private boolean testWhileIdle;
*/
/**
* JedisPoolConfig 连接池
* @return
@Bean public JedisPoolConfig jedisPoolConfig() {
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
// 最大空闲数
jedisPoolConfig.setMaxIdle(maxIdle);
// 连接池的最大数据库连接数
jedisPoolConfig.setMaxTotal(maxTotal);
// 最大建立连接等待时间
jedisPoolConfig.setMaxWaitMillis(maxWaitMillis);
// 逐出连接的最小空闲时间 默认1800000毫秒(30分钟)
jedisPoolConfig.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
// 每次逐出检查时 逐出的最大数目 如果为负数就是 : 1/abs(n), 默认3
jedisPoolConfig.setNumTestsPerEvictionRun(numTestsPerEvictionRun);
// 逐出扫描的时间间隔(毫秒) 如果为负数,则不运行逐出线程, 默认-1
jedisPoolConfig.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);
// 是否在从池中取出连接前进行检验,如果检验失败,则从池中去除连接并尝试取出另一个
jedisPoolConfig.setTestOnBorrow(testOnBorrow);
// 在空闲时检查有效性, 默认false
jedisPoolConfig.setTestWhileIdle(testWhileIdle);
return jedisPoolConfig;
}
/**
* 单机版配置
* @Title: JedisConnectionFactory
* @param @param jedisPoolConfig
* @param @return
* @return JedisConnectionFactory
* @autor lpl
* @date 2018年2月24日
* @throws
@Bean public JedisConnectionFactory jedisConnectionFactory(){
RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
redisStandaloneConfiguration.setHostName(hostName);
redisStandaloneConfiguration.setPort(port);
redisStandaloneConfiguration.setPassword(password);
JedisClientConfiguration.JedisClientConfigurationBuilder jedisClientConfigurationBuilder = JedisClientConfiguration.builder();
jedisClientConfigurationBuilder.connectTimeout(Duration.ofMillis(timeout));
JedisConnectionFactory factory = new JedisConnectionFactory(redisStandaloneConfiguration,jedisClientConfigurationBuilder.build());
return factory;
}
*/
/**
* 实例化 RedisTemplate 对象 jredis实现方式，springboot2.x以后使用下面的方法
*
* @return
@Bean public RedisTemplate<String, Object> functionDomainRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
initDomainRedisTemplate(redisTemplate, redisConnectionFactory);
return redisTemplate;
}
*/
/**
* lettuce实现redis方式
*
* @param redisConnectionFactory
* @return
*/
@Bean
public RedisTemplate<String, Object> redisCacheTemplate(LettuceConnectionFactory redisConnectionFactory) {
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
initDomainRedisTemplate(redisTemplate, redisConnectionFactory);
return redisTemplate;
}
/**
* 设置数据存入 redis 的序列化方式,并开启事务
*
* @param redisTemplate
* @param factory
*/
private void initDomainRedisTemplate(RedisTemplate<String, Object> redisTemplate, RedisConnectionFactory factory) {
//如果不配置Serializer，那么存储的时候缺省使用String，如果用User类型存储，那么会提示错误User can't cast to String！
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setHashKeySerializer(new StringRedisSerializer());
redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
// 开启事务
redisTemplate.setEnableTransactionSupport(true);
redisTemplate.setConnectionFactory(factory);
}
/**
* 注入封装RedisTemplate
*
* @return RedisUtil
* @throws
* @Title: redisUtil
* @autor lpl
* @date 2017年12月21日
*/
@Bean(name = "redisUtils")
public RedisUtils redisUtils(RedisTemplate<String, Object> redisTemplate) {
RedisUtils redisUtil = new RedisUtils();
redisUtil.setRedisTemplate(redisTemplate);
return redisUtil;
}
/**
* 使CacheComponent的redisTemplate组件的key使用StringRedisSerializer而非默认的JdkSerializationRedisSerializer
* 避免key出现字节码的情况
* @param factory redis链接
* @return RedisTemplate
*/
/*@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(factory);
RedisSerializer redisSerializer = new StringRedisSerializer();
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
//key使用StringRedisSerializer序列化
redisTemplate.setKeySerializer(redisSerializer);
//value使用jackson2JsonRedisSerializer序列化
redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
return redisTemplate;
}*/
/**
* 创建Redis消息监听者容器
* @param factory
* @return
*/
@Bean("redisMessageListenerContainer")
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory factory){
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
return container;
}
}

14
src/main/java/com/example/math/constants/CommonConstants.java Normal file
View File

@ -0,0 +1,14 @@
package com.example.math.constants;
/**
* @Author: 吴彬
* @CreateTime: 2023-08-07 16:11
* @Description: 公共常量
* @Version: 1.0
*/
public class CommonConstants {
/**
* 无人机异步飞行
*/
public static final String ASYNFLY = "asynFly";
}

162
src/main/java/com/example/math/constants/DJIMethonConstants.java Normal file
View File

@ -0,0 +1,162 @@
package com.example.math.constants;
/**
* @Author: 吴彬
* @CreateTime: 2023-08-07 16:11
* @Description: 公共常量
* @Version: 1.0
*/
public class DJIMethonConstants {
/**
* 直播方法
*/
public static final String LIVE_START_PUSH = "live_start_push";
/**
* events 飞行任务时间
*/
public static final String FLIGHTTASK_PROGRESS = "flighttask_progress";
/**
* 大疆机场关于 组织 设备绑定码校验回复
*/
public static final String AIRPORT_BIND_STATUS = "airport_bind_status";
/**
* 大疆机场关于 组织 设备绑定码校验回复 二次握手？？
*/
public static final String AIRPORT_ORGANIZATION_GET = "airport_organization_get";
/**
* 机场建立连接？？？
*/
public static final String AIRPORT_ORGANIZATION_BIND = "airport_organization_bind";
/**
* 与阿里云建立连接
*/
public static final String STORAGE_CONFIG_GET = "storage_config_get";
/**
* 下发任务后的二次航线获取
*/
public static final String FLIGHTTASK_RESOURCE_GET = "flighttask_resource_get";
/**
* 一键起飞
*/
public static final String TAKEOFF_TO_POINT = "takeoff_to_point";
/**
* 飞行控制权抢夺
*/
public static final String FLIGHT_AUTHORITY_GRAB = "flight_authority_grab";
/**
* 负载控制权抢夺
*/
public static final String PAYLOAD_AUTHORITY_GRAB = "payload_authority_grab";
/**
* 进入指令飞行控制模式
*/
public static final String DRC_MODE_ENTER = "drc_mode_enter";
/**
* 退出指令飞行控制模式
*/
public static final String DRC_MODE_EXIT = "drc_mode_exit";
/**
* 负载控制—切换相机模式
*/
public static final String CAMERA_MODE_SWITCH = "camera_mode_switch";
/**
* 负载控制—开始拍照
*/
public static final String CAMERA_PHOTO_TAKE = "camera_photo_take";
/**
* 媒体文件上传结果上报
*/
public static final String FILE_UPLOAD_CALLBACK = "file_upload_callback";
/**
* 负载控制—停止拍照
*/
public static final String CAMERA_PHOTO_STOP = "camera_photo_stop";
/**
* 转头云台角度
*/
public static final String CAMERA_LOOK_AT = "camera_look_at";
/**
* 负载控制—画面拖动控制
*/
public static final String CAMERA_SCREEN_DRAG = "camera_screen_drag";
/**
* 负载控制—重置云台 云台 上下转动
*/
public static final String GIMBAL_RESET = "gimbal_reset";
/**
* DRC-飞行器急停
*/
public static final String DRONE_EMERGENCY_STOP = "drone_emergency_stop";
/**
* DRC-飞行控制
*/
public static final String DRONE_CONTROL = "drone_control";
/**
* 航线暂停
*/
public static final String FLIGHTTASK_PAUSE = "flighttask_pause";
/**
* 航线恢复
*/
public static final String FLIGHTTASK_RECOVERY = "flighttask_recovery";
/**
* 负载控制—变焦
*/
public static final String CAMERA_FOCAL_LENGTH_SET = "camera_focal_length_set";
/**
* 切换无人机直播镜头
*/
public static final String LIVE_LENS_CHANGE = "live_lens_change";
/**
* 一键返航
*/
public static final String RETURN_HOME = "return_home";
/**
* 取消任务
*/
public static final String FLIGHTTASK_UNDO = "flighttask_undo";
/**
* 指点飞行
*/
public static final String FLY_TO_POINT = "fly_to_point";
/**
* 结束 flyto 飞向目标点任务
*/
public static final String FLY_TO_POINT_STOP = "fly_to_point_stop";
/**
* 更新 flyto 目标点
*/
public static final String FLY_TO_POINT_UPDATE = "fly_to_point_update";
}

51
src/main/java/com/example/math/constants/DJITopicConstants.java Normal file
View File

@ -0,0 +1,51 @@
package com.example.math.constants;
/**
* @Author: 陈璐
* @CreateTime: 2023-08-07 16:11
* @Description: 大疆公用topic名称
* @Version: 1.0
*/
public class DJITopicConstants {
//机场sn号
private static String airPortSn = "7CTDM3D00BVY4C";
//机场镜头参数
//无人机sn号
private static String droneSn = "1581F6Q8D243100C605L";
/**
* 机场OSD数据
*/
public static final String airportOSD = String.format("thing/product/%s/osd", droneSn);
/**
* 无人机OSD数据
*/
public static final String droneOSD = String.format("thing/product/%s/osd", airPortSn);
/**
* 机场events事件
*/
public static final String EVENTS = String.format("thing/product/%s/events", airPortSn);
/**
* 机场请求无人机上线事件 需要回复 sys/product/7CTDM3D00BVY4C/status
*/
public static final String STATUS = String.format("sys/product/%s/status", airPortSn);
/**
* 机场 无人机推流失败的回复 thing/product/7CTDM3D00BVY4C/services_reply
*/
public static final String SERVICES_REPLY = String.format("thing/product/%s/services_reply", airPortSn);
/**
* 大疆机场关于 开发者平台账号校验的回复 thing/product/7CTDM3D00BVY4C/requests
*/
public static final String REQUESTS = String.format("thing/product/%s/requests", airPortSn);
}

198
src/main/java/com/example/math/controller/AudioTransformProcess.java Normal file
View File

@ -0,0 +1,198 @@
package com.example.math.controller;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.concurrent.TimeUnit;
@Slf4j
public class AudioTransformProcess {
private final Process ffmpegProcess;
private final String threadName;
private static int processCount;
public AudioTransformProcess(String[] command) throws IOException {
log.debug("initializing ffmpegProcess...");
ProcessBuilder pb = new ProcessBuilder(command);
this.ffmpegProcess = pb.start();
this.threadName = Thread.currentThread().getName();
synchronized (AudioTransformProcess.class){
processCount++;
}
log.debug("ffmpegProcess initialized.Alive process count:{}", processCount);
}
/**
* 获取流式地从wav或者mp3转为opus格式的指令示例,实时管道方式
* -i: 输入文件
* -vn: 输出文件禁用视频流
* -ac: 设置输出文件音频通道的数量
* -ar: 设置输出文件音频采样率
* -acodec: 设置输出文件音频编解码器
* -y: 无需询问即可覆盖输出文件
* @return
*/
public static String[] getTransformToOpusPipeCommandExample(String originFormat){
return new String[]{"ffmpeg", "-y", "-f", originFormat, "-i", "pipe:", "-f", "opus", "pipe:"};
}
/**
* 获取流式地从MP3或者wav转为pcm格式的指令示例,实时管道方式
* -i: 输入文件
* -vn: 输出文件禁用视频流
* -ac: 设置输出文件音频通道的数量
* -ar: 设置输出文件音频采样率
* -acodec: 设置输出文件音频编解码器
* -y: 无需询问即可覆盖输出文件
* @return
*/
public static String[] getAudioToPcmPipeCommandExample(){
return new String[]{"ffmpeg","-i", "pipe:0", "-f", "s16le","-ac","1","-ar","16000", "pipe:1","-y"};
}
/**
* 获取通过文件方式从wav或者mp3转为pcm格式的流指令示例
* -i: 输入文件
* -vn: 输出文件禁用视频流
* -ac: 设置输出文件音频通道的数量
* -ar: 设置输出文件音频采样率
* -acodec: 设置输出文件音频编解码器
* -y: 无需询问即可覆盖输出文件
* -b:a 指定比特率
* @param sourceFile 源格式文件
* @return
*/
public static String[] getAudioToPcmFileCommandExample(String sourceFile,String targetFile){//注意，ffmpeg不会创建文件夹，当不存在文件夹时不会输出到文件，也不会报错
if((sourceFile.endsWith(".wav") || sourceFile.endsWith(".mp3"))&& targetFile.endsWith(".pcm"))
return new String[]{"ffmpeg", "-i", sourceFile,"-acodec","pcm_s16le","-ac","1","-ar","16000","-f","s16le", targetFile,"-y"};
else throw new RuntimeException("param invalid");
}
/**
* 获取通过文件方式从opus转为wav格式的流指令示例
* -i: 输入文件
* -vn: 输出文件禁用视频流
* -ac: 设置输出文件音频通道的数量
* -ar: 设置输出文件音频采样率
* -acodec: 设置输出文件音频编解码器
* -y: 无需询问即可覆盖输出文件
* @param sourceFile 源格式文件
* @return
*/
public static String[] getOpusToWavFileCommandExample(String sourceFile,String targetFile){//注意，ffmpeg不会创建文件夹，当不存在文件夹时不会输出到文件，也不会报错
if(sourceFile.endsWith(".opus")&& targetFile.endsWith(".wav"))
return new String[]{"ffmpeg", "-i", sourceFile,"-vn","-acodec","pcm_s16le","-ac","1","-ar","48000","-b:a","16k",targetFile,"-y"};
else throw new RuntimeException("param invalid");
}
/**
* 获取通过文件方式从wav或者mp3转为opus格式的流指令示例
* -i: 输入文件
* -vn: 输出文件禁用视频流
* -ac: 设置输出文件音频通道的数量
* -ar: 设置输出文件音频采样率
* -acodec: 设置输出文件音频编解码器
* -y: 无需询问即可覆盖输出文件
* -b:a 指定比特率
* @param sourceFile 源格式文件
* @return
*/
public static String[] getTransformToOpusFileCommandExample(String sourceFile, String targetFile){//注意，ffmpeg不会创建文件夹，当不存在文件夹时不会输出到文件，也不会报错
if((sourceFile.endsWith(".wav")|| sourceFile.endsWith(".mp3")) && targetFile.endsWith(".opus"))
return new String[]{"ffmpeg", "-i", sourceFile,"-vn","-acodec","libopus","-ac","1","-ar","16000","-b:a","16k",targetFile,"-y"};
else throw new RuntimeException("param invalid");
}
/**
* 抽取音频文件前n秒指令示例
* @param sourceFile
* @param targetFile
* @param seconds
* @return
*/
public static String[] getExtractFirstSecondsCommandExample(String sourceFile,String targetFile,int seconds){//注意，ffmpeg不会创建文件夹，当不存在文件夹时不会输出到文件，也不会报错
return new String[]{"ffmpeg", "-i", sourceFile,"-vn","-c:a", "copy", "-t", String.valueOf(seconds),targetFile,"-y"};
}
/**
* 异步写入管道，由用户决定执行的线程
* @param inputStream
*/
public void asyncWriteToPipe(InputStream inputStream){
String currentThreadName = Thread.currentThread().getName();
if(currentThreadName.equals(threadName)){
throw new RuntimeException("not async.");
}
try (OutputStream stdin = ffmpegProcess.getOutputStream()) {
byte[] buffer = new byte[1024];
int b = 0;
while((b = inputStream.read(buffer))!=-1) {
stdin.write(buffer, 0, b);
log.debug("write to pip {} B.",b);
}
} catch (IOException e) {
log.error(e.getMessage());
e.printStackTrace();
} finally {
try {
log.debug("closing stdin pipeline...");
ffmpegProcess.getOutputStream().close();
log.debug("stdin closed.");
} catch (IOException e) {
log.error(e.getMessage());
e.printStackTrace();
}
}
}
/**
* 从ffmpeg管道中获取输入流，用于异步流式读取转换结果
* @return
*/
public InputStream getReaderFromPipe(){
return ffmpegProcess.getInputStream();
}
/**
* 等待ffmpeg命令执行结束
* @return
* @throws InterruptedException
*/
public int waitFor() throws InterruptedException {
return ffmpegProcess.waitFor();
}
/**
* 等待ffmpeg命令执行结束
* @param timeout
* @param unit
* @return
* @throws InterruptedException
*/
public boolean waitFor(long timeout, TimeUnit unit) throws InterruptedException {
return ffmpegProcess.waitFor(timeout,unit);
}
/**
* 获取存在的AudioTransformProcess对象的个数
* @return
*/
public int getProcessCount(){
return processCount;
}
@Override
protected void finalize() throws Throwable {
synchronized (AudioTransformProcess.class){
processCount--;
}
}
}

69
src/main/java/com/example/math/controller/ExampleController.java Normal file
View File

@ -0,0 +1,69 @@
package com.example.math.controller;
import com.alibaba.fastjson2.JSONObject;
import com.example.math.mqtt.SendMsg;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;
/**
* demo mqtt send receive
*/
@RestController
@RequestMapping(path="/example")
public class ExampleController {
@GetMapping("/v1/access")
public String access() {
System.out.println("enter access123 timstamp is [" + System.currentTimeMillis() + "].");
return "finish...";
}
// public static void main(String[] args) {
// Map<String,Object> credentials = new HashMap<>();
// credentials.put("access_key_id" ,"STS.NSpXAi8To2Yb6EoH2XEFgxZ37");
// credentials.put("access_key_secret" ,"BRG6pmgRr5rQ2je164881vsZLVfLp1kr7vtES5KWSNj5");
// credentials.put("expire" ,3600);
// credentials.put("security_token" ,"CAIS8AJ1q6Ft5B2yfSjIr5DFE/vd1YtOhZuJNGPerDINSclLl5+Y1Tz2IHBPenVqA+Efsfoym2BU7PYalqJIRoReREvCUcZr8sz6ZLsAh9CT1fau5Jko1be0ewHKeQKZsebWZ+LmNpy/Ht6md1HDkAJq3LL+bk/Mdle5MJqP+/kFC9MMRVuAcCZhDtVbLRcYgq18D3bKMuu3ORPHm3fZCFES2jBxkmRi86+ysIP+phPVlw/90fRH5dazcIStKtVhN5c6UtK80+Exd6/I3yNc6gINtoUO1fcVqGic5YDEXwgMv0XbbdC5qIM/cFVLAYEhALNBofTGkvl1h/fejYyfyWwWbboLD36FFdn/npabSL/zZo4jF6zyPnPWycyCLYXlM7PTn9XS4mmgEvhWyR8YYXREIlmyof91cNNTFEzqzSTRZSvSfU7pK/nMvLvsaRHGCP7imk1TMeRnBGgjbCUN1leKH8FLDUoiOj0ICd6hgSZoMng2wRqAASqOBZnJmcA4+fb4OuIztj0Ne4IQTB+Id8Vo9jMoiw1gD7ZZmW45q2riU54LRZ8oS5jA92DXCVy4C+iGYpPxHLtaBmxiWlMd+8hwvD3brxAcRr5nBVbINra/SsQvpFNbyd3F1BVLtjoR6dMsq2xIvL28/VOKSA4raa7Kf3PgAApUIAA=");
//
//
// Map<String,Object> outputItem = new HashMap<>();
// outputItem.put("bucket" ,"hyslj-dev");
// outputItem.put("endpoint" ,"https://oss-cn-shanghai.aliyuncs.com");
// outputItem.put("object_key_prefix" ,"b4cfaae6-bd9d-4cd0-8472-63b608c3c581");
// outputItem.put("provider" ,"ali");
// outputItem.put("region" ,"sh");
// outputItem.put("credentials" ,credentials);
//
// Map<String,Object> data = new HashMap<>();
// data.put("output",outputItem);
// data.put("result",0);
//
// Map<String,Object> content = new HashMap<>();
// content.put("tid","123");
// content.put("bid","456");
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","storage_config_get");
// content.put("data",data);
//
// String msg = JSONObject.toJSONString(content);
//// log.info("will send msg is {}" ,msg);
//
// System.out.println(msg);
// }
}

65
src/main/java/com/example/math/controller/FfmpegPipeExample.java Normal file
View File

@ -0,0 +1,65 @@
package com.example.math.controller;
import com.example.math.controller.AudioTransformProcess;
import lombok.extern.slf4j.Slf4j;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Slf4j
public class FfmpegPipeExample {
static ExecutorService executor = Executors.newFixedThreadPool(5);
public static void main(String[] args) throws IOException {
long startTimestamp = System.currentTimeMillis();
String[] command = AudioTransformProcess.getAudioToPcmPipeCommandExample();
AudioTransformProcess process = new AudioTransformProcess(command);
String file ="00.wav";
String outputFile = "00_t.pcm";
FfmpegPipeExample example = new FfmpegPipeExample();
executor.execute(()-> example.write(file,process));
try(OutputStream outputStream = new BufferedOutputStream(Files.newOutputStream(Paths.get(outputFile)));
InputStream stdout = process.getReaderFromPipe();){
log.info("开始读取输出管道");
byte[] buffer = new byte[8192];
int bytesRead;
while ((bytesRead = stdout.read(buffer)) != -1) {
outputStream.write(buffer, 0, bytesRead);
outputStream.flush();
log.info("从输出管道读取{}字节", bytesRead);
}
process.waitFor(100, TimeUnit.SECONDS);
log.info("cost time:{} ms.",System.currentTimeMillis() - startTimestamp);
} catch ( InterruptedException e) {
log.error(e.getMessage());
}
}
// @Async
public void write(String file, AudioTransformProcess process){
try(InputStream inputStream = new BufferedInputStream(Files.newInputStream(Paths.get(file)))){
process.asyncWriteToPipe(inputStream);//异步写入管道
}catch (Exception e){
log.error(e.getMessage());
}
}
}

1395
src/main/java/com/example/math/controller/OperateController.java Normal file
View File

File diff suppressed because it is too large Load Diff

22
src/main/java/com/example/math/entity/dto/BreakPoint.java Normal file
View File

@ -0,0 +1,22 @@
package com.example.math.entity.dto;
import lombok.Data;
/**
* 航线断点信息
*/
@Data
public class BreakPoint {
/** 断点序号 **/
private int index;
/** 当前航段进度 **/
private double progress;
/** 断点状态 **/
private int state;
/** 航线 ID **/
private int wayline_id;
}

14
src/main/java/com/example/math/entity/dto/ExecutableConditions.java Normal file
View File

@ -0,0 +1,14 @@
package com.example.math.entity.dto;
import lombok.Data;
/**
* 任务执行条件
*/
@Data
public class ExecutableConditions {
/** 存储容量 **/
private int storage_capacity;
}

16
src/main/java/com/example/math/entity/dto/File.java Normal file
View File

@ -0,0 +1,16 @@
package com.example.math.entity.dto;
import lombok.Data;
/**
* 航线文件对象
*/
@Data
public class File {
/** 文件 MD5 签名 **/
private String fingerprint;
/** 文件 URL **/
private String url;
}

16
src/main/java/com/example/math/entity/dto/FileDto.java Normal file
View File

@ -0,0 +1,16 @@
package com.example.math.entity.dto;
import lombok.Data;
/**
* 航线文件对象
*/
@Data
public class FileDto {
/** 文件 MD5 签名 **/
private String fingerprint;
/** 文件 URL **/
private String url;
}

77
src/main/java/com/example/math/entity/dto/FlightTask.java Normal file
View File

@ -0,0 +1,77 @@
package com.example.math.entity.dto;
import com.example.math.entity.dto.BreakPoint;
import com.example.math.entity.dto.ExecutableConditions;
import com.example.math.entity.dto.FileDto;
import com.example.math.entity.dto.ReadyConditions;
import com.example.math.entity.dto.SimulateMission;
import lombok.Data;
/**
* 下发任务对象
*/
@Data
public class FlightTask {
/**
* 航线文件对象
*/
private FileDto file;
/**
* 任务就绪条件
*/
private ReadyConditions ready_conditions;
/**
* 任务执行条件
*/
private ExecutableConditions executable_conditions;
/**
* 航线断点信息
*/
private BreakPoint break_point;
/**
* 是否在模拟器中执行任务
*/
private SimulateMission simulate_mission;
/**
* 计划 ID
*/
private String flight_id;
/**
* 开始执行时间
* 任务开始执行时间毫秒时间戳。可选字段。当 task_type 为 0 或 1 时必填，为 2 时非必填。
*/
private long execute_time;
/**
* 任务类型
* {"0":"立即任务","1":"定时任务","2":"条件任务"}
* 立即任务和定时任务均由 execute_time 指定执行时间。条件任务支持 ready_conditions 字段指定任务就绪条件。任务可在指定时间段内满足一定条件后即可执行。立即任务媒体上传优先级最高，定时任务和条件任务媒体上传优先级相同。
*/
private int task_type;
/** 返航高度 **/
private int rth_altitude;
/** 返航高度模式
* {"0":"智能高度","1":"设定高度"}
* **/
private int rth_mode;
/** 遥控器失控动作 {"0":"返航","1":"悬停","2":"降落"} **/
private int out_of_control_action;
/** 航线失控动作 {"0":"继续执行航线任务","1":"退出航线任务，执行遥控器失控动作"} **/
private int exit_wayline_when_rc_lost;
/** 航线精度类型 {"0":"GPS 任务","1":"高精度 RTK 任务"} **/
private int wayline_precision_type;
}

31
src/main/java/com/example/math/entity/dto/ReadyConditions.java Normal file
View File

@ -0,0 +1,31 @@
package com.example.math.entity.dto;
import lombok.Data;
import java.util.Date;
/**
* 任务就绪条件
*/
@Data
public class ReadyConditions {
/** 电池容量 **/
private int battery_capacity = 30;
/** 任务可执行时段的开始时间 - 300*1000 **/
private long begin_time;
/** 任务可执行时段的结束时间 + 300*1000 **/
private long end_time;
public long initializeBeginTime(Date date) {
return date.getTime() - 300 * 1000;
}
public long initializeEndTime(Date date) {
return date.getTime() + 300 * 1000;
}
}

19
src/main/java/com/example/math/entity/dto/SimulateMission.java Normal file
View File

@ -0,0 +1,19 @@
package com.example.math.entity.dto;
import lombok.Data;
/**
* 是否在模拟器中执行任务
* 可选字段，用于在室内进行模拟任务调试。
*/
@Data
public class SimulateMission {
/** 是否开启模拟器任务 {"0":"不开启","1":"开启"} 当次任务打开或关闭模拟器 **/
private int is_enable;
/** 纬度 **/
private double latitude;
/** 经度 **/
private double longitude;
}

0
src/main/java/com/example/math/enums/CameraTypeEnum.java Normal file
View File

0
src/main/java/com/example/math/enums/LiveUrlTypeEnum.java Normal file
View File

0
src/main/java/com/example/math/enums/OutOfControlActionEnum.java Normal file
View File

0
src/main/java/com/example/math/enums/TaskTypeEnum.java Normal file
View File

0
src/main/java/com/example/math/enums/VideoQualityEnum.java Normal file
View File

0
src/main/java/com/example/math/enums/WindDirectionEnum.java Normal file
View File

0
src/main/java/com/example/math/mqtt/IpUtils.java Normal file
View File

127
src/main/java/com/example/math/mqtt/MqttConfig.java Normal file
View File

@ -0,0 +1,127 @@
package com.example.math.mqtt;
import com.example.math.mqtt.consumer.MqttConsumerCallBack;
import com.example.math.utils.IpUtils;
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.context.annotation.Configuration;
import javax.annotation.PostConstruct;
@Configuration
@Slf4j
public class MqttConfig {
@Value("${spring.mqtt.username}")
private String username;
@Value("${spring.mqtt.password}")
private String password;
@Value("${spring.mqtt.url}")
private String hostUrl;
@Value("${spring.mqtt.client.id}")
private String clientId;
@Value("${spring.mqtt.default.topic}")
private String defaultTopic;
/**
* 客户端对象
*/
private MqttClient client;
/**
* 在bean初始化后连接到服务器
*/
@PostConstruct
public void init(){
connect();
}
/**
* 客户端连接服务端
*/
public void connect(){
try{
clientId += "dji_mqtt_zscl";
//创建MQTT客户端对象
String ip = IpUtils.getHostIp();
//创建MQTT客户端对象
client = new MqttClient(hostUrl, ip + "_" + clientId,new MemoryPersistence());
//连接设置
MqttConnectOptions options = new MqttConnectOptions();
//是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
//设置为true表示每次连接服务器都是以新的身份
options.setCleanSession(true);
//设置连接用户名
options.setUserName(username);
//设置连接密码
options.setPassword(password.toCharArray());
//设置超时时间，单位为秒
options.setConnectionTimeout(0);
//设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
options.setKeepAliveInterval(20);
//设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
options.setWill("willTopic",(clientId + "与服务器断开连接").getBytes(),0,false);
//设置回调
client.setCallback(new MqttConsumerCallBack());
client.connect(options);
//订阅主题
//消息等级，和主题数组一一对应，服务端将按照指定等级给订阅了主题的客户端推送消息
//订阅主题
// client.subscribe("thing/product/+/requests",1);
// client.subscribe("thing/product/+/requests_reply",1);
// client.subscribe("thing/product/+/services",1);
// client.subscribe("thing/product/+/services_reply",1);
// client.subscribe("sys/product/+/status",1);
// client.subscribe("thing/product/+/osd",1);
// client.subscribe("thing/product/8UUXN2T00A01RU/events",1);
} catch(MqttException e){
e.printStackTrace();
}
}
public MqttDeliveryToken publish(int qos,boolean retained,String topic,String message){
MqttMessage mqttMessage = new MqttMessage();
mqttMessage.setQos(qos);
mqttMessage.setRetained(retained);
mqttMessage.setPayload(message.getBytes());
//主题的目的地，用于发布/订阅信息
MqttTopic mqttTopic = client.getTopic(topic);
//提供一种机制来跟踪消息的传递进度
//用于在以非阻塞方式（在后台运行）执行发布是跟踪消息的传递进度
MqttDeliveryToken token = null;
try {
//将指定消息发布到主题，但不等待消息传递完成，返回的token可用于跟踪消息的传递状态
//一旦此方法干净地返回，消息就已被客户端接受发布，当连接可用，将在后台完成消息传递。
token = mqttTopic.publish(mqttMessage);
token.waitForCompletion();
} catch (MqttException e) {
e.printStackTrace();
}
return token;
}
public boolean isconnect() {
try {
String ip = IpUtils.getHostIp();
//创建MQTT客户端对象
client = new MqttClient(hostUrl,ip,new MemoryPersistence());
}catch (Exception e){
e.printStackTrace();
}
return client.isConnected();
}
}

17
src/main/java/com/example/math/mqtt/SendMsg.java Normal file
View File

@ -0,0 +1,17 @@
package com.example.math.mqtt;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@Data
public class SendMsg {
private int qos;
private boolean retained;
private String topic;
private String message;
}

29
src/main/java/com/example/math/mqtt/SpringUtil.java Normal file
View File

@ -0,0 +1,29 @@
package com.example.math.mqtt;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
@Component
public class SpringUtil implements ApplicationContextAware {
private static ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
SpringUtil.applicationContext = applicationContext;
}
public static Object getBean(String beanName) {
return applicationContext.getBean(beanName);
}
public static <T> T getBean(Class<T> beanClass) {
return applicationContext.getBean(beanClass);
}
public static String getProperty(String key) {
return applicationContext.getEnvironment().getProperty(key);
}
}

102
src/main/java/com/example/math/mqtt/SpringUtils.java Normal file
View File

@ -0,0 +1,102 @@
package com.example.math.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.stereotype.Component;
/**
* spring工具类 方便在非spring管理环境中获取bean
*/
@Component
public class SpringUtils implements BeanFactoryPostProcessor {
/**
* Spring应用上下文环境
*/
private static ConfigurableListableBeanFactory beanFactory;
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
SpringUtils.beanFactory = beanFactory;
}
/**
* 获取对象
*
* @param name
* @return Object 一个以所给名字注册的bean的实例
* @throws BeansException
*/
@SuppressWarnings("unchecked")
public static <T> T getBean(String name) throws BeansException {
return (T) beanFactory.getBean(name);
}
/**
* 获取类型为requiredType的对象
*
* @param clz
* @return
* @throws BeansException
*/
public static <T> T getBean(Class<T> clz) throws BeansException {
T result = (T) beanFactory.getBean(clz);
return result;
}
/**
* 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
*
* @param name
* @return boolean
*/
public static boolean containsBean(String name) {
return beanFactory.containsBean(name);
}
/**
* 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
*
* @param name
* @return boolean
* @throws NoSuchBeanDefinitionException
*/
public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException {
return beanFactory.isSingleton(name);
}
/**
* @param name
* @return Class 注册对象的类型
* @throws NoSuchBeanDefinitionException
*/
public static Class<?> getType(String name) throws NoSuchBeanDefinitionException {
return beanFactory.getType(name);
}
/**
* 如果给定的bean名字在bean定义中有别名，则返回这些别名
*
* @param name
* @return
* @throws NoSuchBeanDefinitionException
*/
public static String[] getAliases(String name) throws NoSuchBeanDefinitionException {
return beanFactory.getAliases(name);
}
/**
* 获取aop代理对象
*
* @param invoker
* @return
*/
@SuppressWarnings("unchecked")
public static <T> T getAopProxy(T invoker) {
return (T) AopContext.currentProxy();
}
}

805
src/main/java/com/example/math/mqtt/consumer/MqttConsumerCallBack.java Normal file
View File

@ -0,0 +1,805 @@
package com.example.math.mqtt.consumer;
import com.alibaba.fastjson2.JSONObject;
import com.aliyun.sts20150401.Client;
import com.aliyun.sts20150401.models.AssumeRoleRequest;
import com.aliyun.sts20150401.models.AssumeRoleResponse;
import com.aliyun.tea.TeaException;
import com.aliyun.teautil.models.RuntimeOptions;
import com.example.math.constants.DJIMethonConstants;
import com.example.math.mqtt.MqttConfig;
import com.example.math.mqtt.SendMsg;
import com.example.math.mqtt.SpringUtil;
//import com.example.math.task.FirstTask;
import com.example.math.mqtt.SpringUtils;
import com.example.math.utils.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
@Slf4j
@Component
public class MqttConsumerCallBack implements MqttCallback {
public static List<SendMsg> sendList = new ArrayList<>();
@Value("${spring.mqtt.username}")
private String username;
@Value("${spring.mqtt.password}")
private String password;
@Value("${spring.mqtt.url}")
private String address;
// @Value("${spring.mqtt.newUrl}")
// private String newAddress;
@Value("${spring.mqtt.client.id}")
private String clientId;
@Autowired
private RedisUtils redisUtils;
//机场推流地址
private static String airportRtmpUrl = "rtmp://live.push.t-aaron.com/dji/hyslj002";
//无人机推流地址
private static String droneRtmpUrl = "rtmp://live.push.t-aaron.com/dji/hyslj001";
//topic固定参数
private static String someTopic = "thing/product/%s/services";
//机场sn号
private static String airPortSn = "7CTDM7H00BXZD1";
/**
* 客户端断开连接的回调
*/
@Override
public void connectionLost(Throwable throwable) {
log.info("与服务器断开连接，可重连");
MqttConfig client = SpringUtil.getBean(MqttConfig.class);
client.connect();
// if (!client.isconnect()) {
// client.connect();
// log.info("重连成功！");
// }
}
private static int qos = 2;
private static boolean retained = true;
@Autowired
private MqttConfig providerClient;
private static final String DJIAIRPORT_SN = "7CTDM7H00BXZD1";
private static final String DJIDRONE_SN = "1581F6Q8D243100C605L";
private static final String AIRLINE = "airline";
private static final String CAMERA_INDEX = "81-0-0";
/**
* 消息到达的回调
*/
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
//
// if (true){
// return;
// }
// System.out.println();
// System.out.println(String.format("接收消息主题 : %s", topic));
//// System.out.println(String.format("接收消息Qos : %d", message.getQos()));
// System.out.println(String.format("接收消息内容 : %s", new String(message.getPayload())));
//// System.out.println(String.format("接收消息retained : %b", message.isRetained()));
// System.out.println();
// if(topic.length() > 0){
//// return;
// }
// if(true){
// if (Objects.isNull(redisUtils)){
// redisUtils = SpringUtils.getBean(RedisUtils.class);
// }
// redisUtils.set(AIRLINE + DJIDRONE_SN, 11111, 60 * 60 * 24);
// Object o = redisUtils.get(AIRLINE + DJIDRONE_SN);
// }
String payload = new String(message.getPayload());
JSONObject jsonObject = JSONObject.parseObject(payload);
Object tid = jsonObject.getString("tid");
Object bid = jsonObject.getString("bid");
Object timestamp = jsonObject.getLong("timestamp");
Object method = jsonObject.getString("method");
Object gateway = jsonObject.getString("gateway");
//
String data = jsonObject.getString("data");
if (!topic.equals("thing/product/8UUXN2T00A01RU/events")){
return;
}
//DRC-飞行控制无效原因通知
if (method.equals("joystick_invalid_notify")){
JSONObject dataJson = JSONObject.parseObject(data);
String reason = dataJson.getString("reason");
System.out.printf("joystick_invalid_notify is {}, is {} " + reason);
}
//DRC 链路状态通知
if (method.equals("drc_status_notify")){
JSONObject dataJson = JSONObject.parseObject(data);
String drcState = dataJson.getString("drc_state");
String result = dataJson.getString("result");
System.out.printf("drc_status_notify is {}, is {} " + drcState + "result" + result);
}
//
// log.info("tid is {},bid is {},method is {}, gateway is {}, timestamp is {}",
// tid,bid,method,gateway,timestamp);
// if ("thing/product/7CTDM7H00BXZD1/events".equals(topic)){
// Map<String,Object> data = new HashMap<>();
// data.put("result",0);
// Map<String,Object> content = new HashMap<>();
// content.put("tid",tid);
// content.put("bid",bid);
// content.put("gateway", DJIAIRPORT_SN);
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","flighttask_progress");
// content.put("data",data);
// String msg = JSONObject.toJSONString(content);
// topic += "_reply";
// SendMsg sendMsg = new SendMsg();
// sendMsg.setQos(qos);
// sendMsg.setRetained(retained);
// sendMsg.setTopic(topic);
// sendMsg.setMessage(msg);
// sendList.add(sendMsg);
//
// }
// if ("file_upload_callback".equals(method)){
// System.out.printf("媒体文件上传结果上报" + topic);
// System.out.printf("媒体文件上传结果上报2" + method);
// }
//
// if ("thing/product/7CTDM7H00BXZD1/events".equals(topic) && "file_upload_callback".equals(method)){
// System.out.printf("媒体文件上传结果上报" + topic);
// System.out.printf("媒体文件上传结果上报2" + method);
// }
//
// try{
// if (topic.equals("thing/product/7CTDM7H00BXZD1/osd")){
// //此处处理 无人机osd 回调处理气象 状态等信息 airportStatus
// log.info("topic is {}", topic);
// airportStatus(jsonObject, "7CTDM7H00BXZD1", tid , bid);
//
// Object data = jsonObject.getString("data");
// JSONObject newJson = JSONObject.parseObject(data.toString());
// String string = newJson.getString("drc_state");
// if (Objects.nonNull(string)){
// System.out.printf("osd中的drc_state" + string);
//// if (string.equals("2") || string.equals(2)){
//// log.info("一键起飞命令 drc_mode_enter is start");
//// FirstTask.seq = 0;
////
//// Integer qos = 2;
//// Boolean retained = true;
//// //topic主题 sn是机场sn号 thing/product/%s/services
//// //TODO 此处用的是无人的SN号 待测试
//// String newtopic = String.format(someTopic, airPortSn);
//// //封装数据
//// Map<String,Object> content = new HashMap<>();
//// Map<String,Object> newdata = new HashMap<>();
//// //负载枚举值 镜头负载
//// newdata.put("hsi_frequency", 1);
////
//// //Broker 连接信息
//// Map<String,Object> mqttBroker = new HashMap<>();
//// //服务器连接地址
//// mqttBroker.put("address", newAddress);
//// //客户端ID
//// mqttBroker.put("client_id", clientId += "drc");
//// //enable_tls
//// mqttBroker.put("enable_tls", false);
//// //expire_time 当前时间+3000秒
//// mqttBroker.put("expire_time", System.currentTimeMillis()/1000 + 3600);
//// //密码
//// mqttBroker.put("password", password);
//// //用户名
//// mqttBroker.put("username", username);
//// newdata.put("mqtt_broker", mqttBroker);
//// //设置 OSD 上报频率 {"max":30,"min":1,"unit_name":"赫兹 / Hz"}
//// newdata.put("osd_frequency", 21);
////
//// content.put("tid", UUID.randomUUID().toString());
//// content.put("bid", UUID.randomUUID().toString());
//// content.put("timestamp",System.currentTimeMillis());
//// content.put("method", DJIMethonConstants.DRC_MODE_ENTER);
//// content.put("data", newdata);
//// String message1 = JSONObject.toJSONString(content);
//// System.out.println(message);
//// MqttDeliveryToken publish = providerClient.publish(qos, retained, newtopic, message1);
//// log.info("一键起飞命令 drc_mode_enter is end topic is {}, message is {}, token is {} ", newtopic, message1, publish);
//// }
//
//
// }
//
//
// }else if (topic.equals("thing/product/1581F6Q8D243100C605L/osd")){
// //此处处理 机场osd 回调处理气象 状态等信息
// log.info("处理机场osd回调信息 topic is {}", topic);
// droneStatus(jsonObject, "1581F6Q8D243100C605L", tid , bid);
// }else if (topic.equals("sys/product/7CTDM7H00BXZD1/status")){
// log.info("收到无人机osd失效 sys/product/7CTDM7H00BXZD1/status");
// Map<String,Object> data = new HashMap<>();
// data.put("result",0);
// Map<String,Object> content = new HashMap<>();
// content.put("tid",tid);
// content.put("bid",bid);
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","update_topo");
// content.put("data",data);
// MqttConfig providerClient = SpringUtil.getBean(MqttConfig.class);
// String msg = com.alibaba.fastjson.JSONObject.toJSONString(content);
// topic += "_reply";
// MqttDeliveryToken publish = providerClient.publish(qos, retained, topic, msg);
// log.info("处理无人机osd失效的回复 topic is {}, msg is {}, publish is {}", topic, msg, publish);
// }else if (topic.equals("sys/product/7CTDM7H00BXZD1/status")){
// log.info("收到无人机osd失效 sys/product/7CTDM7H00BXZD1/status");
// Map<String,Object> data = new HashMap<>();
// data.put("result",0);
// Map<String,Object> content = new HashMap<>();
// content.put("tid",tid);
// content.put("bid",bid);
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","update_topo");
// content.put("data",data);
// MqttConfig providerClient = SpringUtil.getBean(MqttConfig.class);
// String msg = com.alibaba.fastjson.JSONObject.toJSONString(content);
// topic += "_reply";
// MqttDeliveryToken publish = providerClient.publish(qos, retained, topic, msg);
// log.info("处理无人机osd失效的回复 topic is {}, msg is {}, publish is {}", topic, msg, publish);
// }else if (topic.equals("thing/product/7CTDM7H00BXZD1/events") && method.equals("drc_status_notify") ) {
// //此处drc链路通知状态
// Object data = jsonObject.getString("data");
// log.info("drc_status_notify topic is {}, data is {}", topic, data);
// JSONObject newJson = JSONObject.parseObject(data.toString());
// log.info("drc链路消息", newJson);
// }else if (topic.equals("thing/product/7CTDM7H00BXZD1/events") && method.equals("takeoff_to_point_progress") ) {
// //处理一键起飞结果
// log.info("topic is {}", topic);
// } else if ( topic.equals("thing/product/7CTDM7H00BXZD1/drc/up") && method.equals("drone_control") ){
// System.out.printf("控制指令报错了" + jsonObject);
// Object data = jsonObject.getString("data");
// JSONObject newJson = JSONObject.parseObject(data.toString());
// String string = newJson.getString("result");
// if (!string.equals(0)){
//// OperateController.seq = 0;
// }
// }
// }catch (Exception e){
//
// }
//
//
// if(topic.equals("thing/product/7CTDM7H00BXZD1/requests")){
// System.out.println("enter...............");
// if(method.equals("config")){
// System.out.println("enter config..........");
// //需要回复
//
// Map<String,Object> data = new HashMap<>();
// data.put("ntp_server_host","Google.mzr.me");
// data.put("app_id","148774");
// data.put("app_key","a3ba3e9934f43b82bd52428b26e695b");
// data.put("app_license","Cm+BDccW/auc1L5Fixy59JrudnnXxguksAfGNPRjaDDqf3ESlHJt4JYaYBlXJ/rkkSPREF/MIg5FZV3MSwPVcXQ2ioOahFH0OEHx0fhxLnrS2uUVXaXcDkDxLRfDuOTvkmojS3KFZIgPOjt6hdF9Madi/WC+EsnZzpA5RvTD/u4=");
//
// Map<String,Object> content = new HashMap<>();
// content.put("tid",tid);
// content.put("bid",bid);
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","config");
// content.put("data",data);
//
// String msg = JSONObject.toJSONString(content);
//// log.info("will send msg is {}" ,msg);
//
// topic += "_reply";
//
// SendMsg sendMsg = new SendMsg();
// sendMsg.setQos(qos);
// sendMsg.setRetained(retained);
// sendMsg.setTopic(topic);
// sendMsg.setMessage(msg);
// sendList.add(sendMsg);
//
//
//
//// if (Objects.isNull(providerClient)){
//// providerClient = SpringUtil.getBean(MqttProducerConfig.class);
//// }
//// providerClient.publish(qos, retained, topic, msg);
// return;
// }else if(method.equals("airport_bind_status")){
//// }else if(method.equals("storage_config_get")){
// System.out.println("enter airport_bind_status..........");
// //需要回复
//
// Map<String,Object> deviceInfo = new HashMap<>();
// deviceInfo.put("sn" ,"7CTDM7H00BXZD1");
// deviceInfo.put("device_callsign" ,"");
// deviceInfo.put("is_device_bind_organization" ,true);
// deviceInfo.put("organization_id" ,"w");
// deviceInfo.put("organization_name" ,"Test Group One");
//
//
//
// List<Map<String,Object>> bindStatusList = new ArrayList<>();
// bindStatusList.add(deviceInfo);
//
// Map<String,Object> bindItem = new HashMap<>();
// bindItem.put("bind_status" ,bindStatusList);
//
// Map<String,Object> data = new HashMap<>();
// data.put("output",bindItem);
// data.put("result",0);
//
// Map<String,Object> content = new HashMap<>();
// content.put("tid",tid);
// content.put("bid",bid);
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","airport_bind_status");
// content.put("data",data);
//
// String msg = JSONObject.toJSONString(content);
//// log.info("will send msg is {}" ,msg);
//
// topic += "_reply";
// SendMsg sendMsg = new SendMsg();
// sendMsg.setQos(qos);
// sendMsg.setRetained(retained);
// sendMsg.setTopic(topic);
// sendMsg.setMessage(msg);
//
// sendList.add(sendMsg);
//// if (Objects.isNull(providerClient)){
//// providerClient = SpringUtil.getBean(MqttConfig.class);
//// }
//// providerClient.publish(qos, retained, topic, msg);
// return;
// }else if(method.equals("airport_organization_get")){
//// }else if(method.equals("storage_config_get")){
// System.out.println("enter airport_organization_get..........");
// //需要回复
//
// Map<String,Object> outputItem = new HashMap<>();
// outputItem.put("organization_name" ,"Test Group One");
//
// Map<String,Object> data = new HashMap<>();
// data.put("output",outputItem);
// data.put("result",0);
//
// Map<String,Object> content = new HashMap<>();
// content.put("tid",tid);
// content.put("bid",bid);
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","airport_organization_get");
// content.put("data",data);
//
// String msg = JSONObject.toJSONString(content);
//// log.info("will send msg is {}" ,msg);
//
// topic += "_reply";
// SendMsg sendMsg = new SendMsg();
// sendMsg.setQos(qos);
// sendMsg.setRetained(retained);
// sendMsg.setTopic(topic);
// sendMsg.setMessage(msg);
//
// sendList.add(sendMsg);
//// if (Objects.isNull(providerClient)){
//// providerClient = SpringUtil.getBean(MqttConfig.class);
//// }
//// providerClient.publish(qos, retained, topic, msg);
// return;
// }else if(method.equals("airport_organization_bind")){
//// }else if(method.equals("storage_config_get")){
// System.out.println("enter airport_organization_get..........");
// //需要回复
//
//
// Map<String,Object> err_info = new HashMap<>();
// err_info.put("sn" ,"7CTDM7H00BXZD1");
// err_info.put("err_code" ,0);
//
// List<Map<String,Object>> err_infos = new ArrayList<>();
// err_infos.add(err_info);
//
// Map<String,Object> outputItem = new HashMap<>();
// outputItem.put("err_infos" ,err_infos);
//
// Map<String,Object> data = new HashMap<>();
// data.put("output",outputItem);
// data.put("result",0);
//
// Map<String,Object> content = new HashMap<>();
// content.put("tid",tid);
// content.put("bid",bid);
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","airport_organization_bind");
// content.put("data",data);
//
// String msg = JSONObject.toJSONString(content);
//// log.info("will send msg is {}" ,msg);
//
// topic += "_reply";
// SendMsg sendMsg = new SendMsg();
// sendMsg.setQos(qos);
// sendMsg.setRetained(retained);
// sendMsg.setTopic(topic);
// sendMsg.setMessage(msg);
//
// sendList.add(sendMsg);
//// if (Objects.isNull(providerClient)){
//// providerClient = SpringUtil.getBean(MqttConfig.class);
//// }
//// providerClient.publish(qos, retained, topic, msg);
// return;
// }else if(method.equals("storage_config_get")){
// System.out.println("enter storage_config_get..........");
// //需要回复
// System.out.printf("jsonObject" + jsonObject);
// Map<String,Object> credentials = new HashMap<>();
// //获取阿里云 OSS对应文件夹的token
// Map<String, String> ossMap = ossToken();
// if (!CollectionUtils.isEmpty(ossMap)){
// credentials.put("access_key_id" , ossMap.get("access_key_id"));
// credentials.put("access_key_secret" , ossMap.get("access_key_secret"));
// credentials.put("expire" ,3600);
// credentials.put("security_token" , ossMap.get("security_token"));
// log.info("获取阿里云上传凭证成功");
// }else {
// log.info("获取阿里云上传凭证失败");
// }
// Map<String,Object> outputItem = new HashMap<>();
// outputItem.put("bucket" ,"dji-dev");
// outputItem.put("endpoint" ,"https://oss-cn-shanghai.aliyuncs.com");
// outputItem.put("object_key_prefix" ,"hyslj");
// outputItem.put("provider" ,"ali");
// outputItem.put("region" ,"sh");
// outputItem.put("credentials" ,credentials);
//
// Map<String,Object> data = new HashMap<>();
// data.put("output",outputItem);
// data.put("result",0);
//
// Map<String,Object> content = new HashMap<>();
// content.put("tid",tid);
// content.put("bid",bid);
// content.put("timestamp",System.currentTimeMillis());
// content.put("method","storage_config_get");
// content.put("data",data);
//
// String msg = JSONObject.toJSONString(content);
//// log.info("will send msg is {}" ,msg);
//
// topic += "_reply";
// SendMsg sendMsg = new SendMsg();
// sendMsg.setQos(qos);
// sendMsg.setRetained(false);
// sendMsg.setTopic(topic);
// sendMsg.setMessage(msg);
//
// sendList.add(sendMsg);
//// //消息回复
//// if (Objects.isNull(providerClient)){
//// providerClient = SpringUtil.getBean(MqttConfig.class);
//// }
//// providerClient.publish(qos, retained, topic, msg);
// return;
// }else if(method.equals("flighttask_resource_get")){
//
//// System.out.println("enter flighttask_resource_get..........");
//// //需要回复
////
//// Map<String,Object> file = new HashMap<>();
//// file.put("fingerprint","AA9D2EF80BBE544B3C7B97C9087E640F");
//// file.put("url","xxxxxx");
////
//// Map<String,Object> output = new HashMap<>();
//// output.put("file" ,file);
////
//// Map<String,Object> data = new HashMap<>();
//// data.put("output",output);
//// data.put("result",0);
////
//// Map<String,Object> content = new HashMap<>();
//// content.put("tid",tid);
//// content.put("bid",bid);
//// content.put("timestamp",System.currentTimeMillis());
//// content.put("method","flighttask_resource_get");
//// content.put("data",data);
////
//// String msg = JSONObject.toJSONString(content);
////// log.info("will send msg is {}" ,msg);
////
//// topic += "_reply";
//// SendMsg sendMsg = new SendMsg();
//// sendMsg.setQos(qos);
//// sendMsg.setRetained(false);
//// sendMsg.setTopic(topic);
//// sendMsg.setMessage(msg);
////
//// sendList.add(sendMsg);
//// //消息回复
//// if (Objects.isNull(providerClient)){
//// providerClient = SpringUtil.getBean(MqttConfig.class);
//// }
//// providerClient.publish(qos, retained, topic, msg);
// return;
// }
// }
//
//
// System.out.println("finish...");
}
/**
* 消息发布成功的回调
*/
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
}
/**
* 气温 状态等数据
* @param oldjsonObject
*/
private void droneStatus(JSONObject oldjsonObject, String sn , Object tid, Object bid) {
Map map = new HashMap();
//发送的坐标信息
String string = oldjsonObject.getString("data");
com.alibaba.fastjson.JSONObject jsonObject = com.alibaba.fastjson.JSONObject.parseObject(string);
//当前位置经度 longitude
String longitude = jsonObject.getString("longitude");
map.put("longitude", longitude);
//当前位置纬度 latitude
String latitude = jsonObject.getString("latitude");
map.put("latitude", latitude);
//飞行器状态
String modeCode = jsonObject.getString("mode_code");
map.put("modeCode", modeCode);
//相对起飞点高度 elevation
String elevation = jsonObject.getString("elevation");
map.put("elevation", elevation);
//风速 wind_speed
String windSpeed = jsonObject.getString("wind_speed");
map.put("windSpeed", windSpeed);
//当前风向 wind_direction
String windDirection = jsonObject.getString("wind_direction");
map.put("windDirection", windDirection);
//低电量告警 low_battery_warning_threshold
String lowBatteryWarningThreshold = jsonObject.getString("low_battery_warning_threshold");
map.put("lowBatteryWarningThreshold", lowBatteryWarningThreshold);
//飞行器电池信息 battery {
// capacity_percent电池的总剩余电量 remain_flight_time剩余飞行时间 return_home_power返航所需电量百分比 landing_power强制降落电量百分比capacity_percent电池剩余电量
// voltage电压 temperature温度 wind_direction风向 }
String battery = jsonObject.getString("battery");
map.put("battery", battery);
RedisUtils redisUtils = SpringUtil.getBean(RedisUtils.class);
redisUtils.hmset("DJIDrone" + sn, map, 300);
Map<String,Object> content = new HashMap<>();
content.put("longitude",longitude);
content.put("latitude",latitude);
String msg = com.alibaba.fastjson.JSONObject.toJSONString(content);
}
/**
* 无人机气温 状态等数据
* @param oldjsonObject
*/
private void airportStatus(JSONObject oldjsonObject, String sn, Object tid, Object bid) {
Map map = new HashMap();
//发送的坐标信息
String string = oldjsonObject.getString("data");
com.alibaba.fastjson.JSONObject jsonObject = com.alibaba.fastjson.JSONObject.parseObject(string);
//当前位置经度 longitude
String longitude = jsonObject.getString("longitude");
if (Objects.isNull(longitude)){
return;
}
map.put("longitude", longitude);
//当前位置纬度 latitude
String latitude = jsonObject.getString("latitude");
map.put("latitude", latitude);
//工作电流 working_current
String workingCurrent = jsonObject.getString("working_current");
map.put("workingCurrent", workingCurrent);
//工作电压 working_voltage
String workingVoltage = jsonObject.getString("working_voltage");
map.put("workingVoltage", workingVoltage);
//舱内湿度 humidity
String humidity = jsonObject.getString("humidity");
map.put("humidity", humidity);
//舱内温度 temperature
String temperature = jsonObject.getString("temperature");
map.put("temperature", temperature);
//environment_temperature 环境温度
String environmentTemperature = jsonObject.getString("environment_temperature");
map.put("environmentTemperature", environmentTemperature);
//wind_speed 风速
String windSpeed = jsonObject.getString("wind_speed");
map.put("windSpeed", windSpeed);
//rainfall 降雨量
String rainfall = jsonObject.getString("rainfall");
map.put("rainfall", rainfall);
//drone_in_dock 飞行器是否在舱{"0":"舱外","1":"舱内"}
String droneInDock = jsonObject.getString("drone_in_dock");
map.put("droneInDock", droneInDock);
//cover_state 舱盖状态"0":"关闭","1":"打开","2":"半开","3":"舱盖状态异常"
String coverState = jsonObject.getString("cover_state");
map.put("coverState", coverState);
//flighttask_step_code 机场任务状态"0":"作业准备中","1":"飞行作业中","2":"作业后状态恢复","3":"自定义飞行区更新中","4":"地形障碍物更新中","5":"任务空闲","255":"飞行器异常","256":"未知状态"
String flighttaskStepCode = jsonObject.getString("flighttask_step_code");
map.put("flighttaskStepCode", flighttaskStepCode);
//mode_code 机场状态
String modeCode = jsonObject.getString("mode_code");
map.put("modeCode", modeCode);
//当前位置经度 longitude
String airPortLongitude = jsonObject.getString("longitude");
map.put("airPortLongitude", airPortLongitude);
//当前位置纬度 latitude
String airPortLatitude = jsonObject.getString("latitude");
map.put("airPortLatitude", airPortLatitude);
RedisUtils redisUtils = SpringUtil.getBean(RedisUtils.class);
redisUtils.hmset("DJIAirport"+ sn, map, 30);
}
/**
* 获取
*/
private Map<String, String> ossToken() {
Map<String, String> ossMap = new HashMap<>();
String policy = "{\n" +
" \"Version\": \"1\", \n" +
" \"Statement\": [\n" +
" {\n" +
" \"Action\": [\n" +
" \"oss:PutObject\"\n" +
" ], \n" +
" \"Resource\": [\n" +
" \"acs:oss:*:*:dji-dev/*\" \n" +
" ], \n" +
" \"Effect\": \"Allow\"\n" +
" }\n" +
" ]\n" +
"}";
String roleARN = "acs:ram::1399733914954856:role/ramosstest";
String sessionName = "SessionTest";
java.util.List<String> args = Arrays.asList();
Client client = null;
try {
client = MqttConsumerCallBack.createClient();
} catch (Exception e) {
log.info("获取阿里云token失败");
}
AssumeRoleRequest assumeRoleRequest = new com.aliyun.sts20150401.models.AssumeRoleRequest()
.setDurationSeconds(3600L)
.setPolicy(policy)
.setRoleArn(roleARN)
.setRoleSessionName(sessionName);
try {
// 复制代码运行请自行打印 API 的返回值
AssumeRoleResponse assumeRoleResponse = client.assumeRoleWithOptions(assumeRoleRequest, new RuntimeOptions());
String arn = assumeRoleResponse.getBody().getAssumedRoleUser().getArn();
String assumedRoleId = assumeRoleResponse.getBody().getAssumedRoleUser().getAssumedRoleId();
String accessKeyId = assumeRoleResponse.getBody().getCredentials().getAccessKeyId();
String accessKeySecret = assumeRoleResponse.getBody().getCredentials().getAccessKeySecret();
String securityToken = assumeRoleResponse.getBody().getCredentials().getSecurityToken();
String expiration = assumeRoleResponse.getBody().getCredentials().getExpiration();
log.info(arn);
log.info(assumedRoleId);
log.info(accessKeyId);
log.info(accessKeySecret);
log.info(securityToken);
log.info(expiration);
log.info("a");
ossMap.put("access_key_id", accessKeyId);
ossMap.put("access_key_secret", accessKeySecret);
ossMap.put("security_token", securityToken);
return ossMap;
} catch (TeaException error) {
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
log.info(error.getMessage());
// 诊断地址
log.info(error.getData().get("Recommend")+"");
com.aliyun.teautil.Common.assertAsString(error.message);
} catch (Exception _error) {
TeaException error = new TeaException(_error.getMessage(), _error);
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
log.info(error.getMessage());
// 诊断地址
log.info(error.getData().get("Recommend")+"");
com.aliyun.teautil.Common.assertAsString(error.message);
}
return ossMap;
}
/**
* <b>description</b> :
* <p>使用AK&amp;SK初始化账号Client</p>
* @return Client
*
* @throws Exception
*/
public static Client createClient() throws Exception {
// 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考。
// 建议使用更安全的 STS 方式，更多鉴权访问方式请参见：https://help.aliyun.com/document_detail/378657.html。
com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
// 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。
.setAccessKeyId("LTAI5tFmaJHU7ywvBH2j12Gp")
// 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。
.setAccessKeySecret("mX6rS1xYTgZgFAgpOrQHyc6jLXv0AV");
// Endpoint 请参考 https://api.aliyun.com/product/Sts
config.endpoint = "sts.cn-shanghai.aliyuncs.com";
return new Client(config);
}
}

42
src/main/java/com/example/math/mqtt/producer/MqttProviderCallBack.java Normal file
View File

@ -0,0 +1,42 @@
package com.example.math.mqtt.producer;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MqttProviderCallBack implements MqttCallback{
@Value("${spring.mqtt.client.id}")
private String clientId;
/**
* 与服务器断开的回调
*/
@Override
public void connectionLost(Throwable cause) {
System.out.println(clientId+"与服务器断开连接 product");
}
/**
* 消息到达的回调
*/
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
}
/**
* 消息发布成功的回调
*/
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
IMqttAsyncClient client = token.getClient();
System.out.println(client.getClientId()+"发布消息成功！");
}
}

0
src/main/java/com/example/math/service/IMqttService.java Normal file
View File

24
src/main/java/com/example/math/service/MqttServiceImpl.java Normal file
View File

@ -0,0 +1,24 @@
//package com.example.math.service;
//
//import com.example.math.mqtt.MqttConfig;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Service;
//
///**
// * <p>
// */
//@Service
//@Slf4j
//public class MqttServiceImpl implements IMqttService {
//
// @Autowired
// private MqttConfig client;
//
// @Override
// public void alarmSend(int qos, boolean retained, String topic, String message) {
// client.publish(qos, retained, topic, message);
// }
//
//
//}

64
src/main/java/com/example/math/task/FirstTask.java Normal file
View File

@ -0,0 +1,64 @@
package com.example.math.task;
import com.example.math.constants.DJIMethonConstants;
import com.example.math.mqtt.MqttConfig;
import com.example.math.mqtt.SendMsg;
import com.example.math.mqtt.consumer.MqttConsumerCallBack;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Slf4j
@Configuration //1.主要用于标记配置类，兼备Component的效果。
@EnableScheduling // 2.开启定时任务
public class FirstTask {
private static Integer seq = 0;
@Autowired
private MqttConfig providerClient;
// @Scheduled(cron = "0/1 * * * * ?")
// //或直接指定时间间隔，例如：5秒
// //@Scheduled(fixedRate=5000)
// private void configureTasks() {
// System.err.println("执行静态定时任务时间: " + System.currentTimeMillis());
//
// List<SendMsg> sendList = MqttConsumerCallBack.sendList;
//
// while(sendList.size()>0) {
// SendMsg sendMsg = sendList.remove(0);
// log.info("sendmsg qos is {},retained is {}, topic is {}, msg is {}",
// sendMsg.getQos(),sendMsg.isRetained(),sendMsg.getTopic(),sendMsg.getMessage());
// providerClient.publish(sendMsg.getQos(),sendMsg.isRetained(),sendMsg.getTopic(),sendMsg.getMessage());
//
// }
// }
@Scheduled(cron = "0/5 * * * * ?")
//或直接指定时间间隔，例如：5秒
//@Scheduled(fixedRate=5000)
private void aaaa() {
Map<String,Object> content = new HashMap<>();
HashMap<String, Long> data
= new HashMap<>();
data.put("timestamp", System.currentTimeMillis());
seq++;
content.put("method", "heart_beat");
content.put("data", data);
content.put("seq", seq);
String message = com.alibaba.fastjson.JSONObject.toJSONString(content);
System.out.println(message);
MqttDeliveryToken publish = providerClient.publish(1, false, "thing/product/7CTDM7H00BXZD1/drc/down", message);
}
}

289
src/main/java/com/example/math/utils/CommonUtils.java Normal file
View File

@ -0,0 +1,289 @@
package com.example.math.utils;
import com.alibaba.fastjson.JSONObject;
import org.springframework.util.CollectionUtils;
import java.lang.reflect.Field;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
/**
* 公共函数类
*/
public class CommonUtils {
/**
* 密码加密
*
* @param password 密码
* @return
*/
public static String password(String password) {
String md51 = md5(password.getBytes());
String pwd = md5((md51 + "IgtUdEQJyVevaCxQnY").getBytes());
return pwd;
}
/**
* 正则匹配富文本图片
*
* @param htmlStr 富文本内容
* @return
*/
public static List<String> getImgStr(String htmlStr) {
Pattern p_image = Pattern.compile("<img.*src\\s*=\\s*(.*?)[^>]*?>", Pattern.CASE_INSENSITIVE);
Pattern r_image = Pattern.compile("src\\s*=\\s*\"?(.*?)(\"|>|\\s+)");
List<String> list = new ArrayList<>();
Matcher m_image = p_image.matcher(htmlStr);
while (m_image.find()) {
// 得到<img />数据
String img = m_image.group();
System.out.println(img);
// 匹配<img>中的src数据
Matcher m = r_image.matcher(img);
while (m.find()) {
list.add(m.group(1));
}
}
return list;
}
/**
* 验证邮箱是否正确
*
* @param email
* @return
*/
public static boolean isEmail(String email) {
boolean flag = false;
try {
String check = "^([a-z0-9A-Z]+[-|\\.]?)+[a-z0-9A-Z]@([a-z0-9A-Z]+(-[a-z0-9A-Z]+)?\\.)+[a-zA-Z]{2,}$";
Pattern regex = Pattern.compile(check);
Matcher matcher = regex.matcher(email);
flag = matcher.matches();
} catch (Exception e) {
flag = false;
}
return flag;
}
/**
* 验证手机号是否正确
*
* @param mobile
* @return
*/
public static boolean isMobile(String mobile) {
boolean flag = false;
try {
Pattern p = Pattern.compile("^(1[0-9])\\d{9}$");
Matcher m = p.matcher(mobile);
flag = m.matches();
} catch (Exception e) {
flag = false;
}
return flag;
}
/**
* 生成指定位数的随机字符串
*
* @param isNum 是否是纯数字
* @param length 长度
* @return
*/
public static String getRandomStr(boolean isNum, int length) {
String resultStr = "";
String str = isNum ? "1234567890" : "1234567890abcdefghijkmnpqrstuvwxyz";
int len = str.length();
boolean isStop = true;
do {
resultStr = "";
int count = 0;
for (int i = 0; i < length; i++) {
double dblR = Math.random() * len;
int intR = (int) Math.floor(dblR);
char c = str.charAt(intR);
if (('0' <= c) && (c <= '9')) {
count++;
}
resultStr += str.charAt(intR);
}
if (count >= 2) {
isStop = false;
}
} while (isStop);
return resultStr;
}
/**
* 判断是否在数组中
*
* @param s
* @param array
* @return
*/
public static boolean inArray(final String s, final String[] array) {
for (String item : array) {
if (item != null && item.equals(s)) {
return true;
}
}
return false;
}
/**
* 从html中提取纯文本
*
* @param strHtml
* @return
*/
public static String stripHtml(String strHtml) {
String content = strHtml.replaceAll("</?[^>]+>", ""); //剔出<html>的标签
content = content.replaceAll("\\s*|\t|\r|\n", "");//去除字符串中的空格,回车,换行符,制表符
return content;
}
/**
* 去除字符串中的空格、回车、换行符、制表符等
*
* @param str 原始字符串
* @return
*/
public static String replaceSpecialStr(String str) {
String repl = "";
if (str != null) {
Pattern p = Pattern.compile("\\s*|\t|\r|\n");
Matcher m = p.matcher(str);
repl = m.replaceAll("");
}
return repl;
}
/**
* 判断某个元素是否在数组中
*
* @param key 元素
* @param map 数组
* @return
*/
public static boolean inArray(String key, Map<String, String> map) {
boolean flag = false;
for (String k : map.keySet()) {
if (k.equals(key)) {
flag = true;
}
}
return flag;
}
/**
* 对象转Map
*
* @param obj 对象
* @return
* @throws IllegalAccessException
*/
public static Map<String, Object> objectToMap(Object obj) throws IllegalAccessException {
Map<String, Object> map = new HashMap<>();
Class<?> clazz = obj.getClass();
for (Field field : clazz.getDeclaredFields()) {
field.setAccessible(true);
String fieldName = field.getName();
Object value = field.get(obj);
map.put(fieldName, value);
}
return map;
}
/**
* 判断是否是JSON格式
*
* @param str JSON字符串
* @return
*/
private boolean isJson(String str) {
try {
JSONObject jsonStr = JSONObject.parseObject(str);
return true;
} catch (Exception e) {
return false;
}
}
/**
* MD5方法
*
* @param source
* @return
*/
public static String md5(byte[] source) {
try {
MessageDigest md = MessageDigest.getInstance("MD5");
md.update(source);
StringBuffer buf = new StringBuffer();
for (byte b : md.digest()) {
buf.append(String.format("%02x", b & 0xff));
}
return buf.toString();
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 对数组进行分组
*
* @param list 数据源
* @param size 每组几个
* @param <T>
* @return
*/
public static <T> List<List<T>> split(List<T> list, Integer size) {
if (CollectionUtils.isEmpty(list)) {
return new ArrayList<>();
}
Integer count = list.size() / size;
List<List<T>> arrayList = new ArrayList<>();
for (int i = 0; i < count; i++) {
List<T> temp = list.subList(i * size, (i + 1) * size);
arrayList.add(temp);
}
Integer extra = list.size() % size;
if (extra != 0) {
List<T> temp = list.subList(count * size, count * size + extra);
arrayList.add(temp);
}
return arrayList;
}
/**
* 批量处理方法
*
* @param operater 回调方法，有入参无返回值
* @param sourceList 批量处理list对象
* @param threshold 阀值，比如有5000个对象，阀值设置1000，就是1000执行一次
* @param <T> 对象类型
*/
public static <T> void batchOperate(Consumer<List<T>> operater, List<T> sourceList, Integer threshold) {
int size = sourceList.size();
int fromIndex = 0;
List<T> list = null;
while (size > fromIndex){
list = sourceList.stream().skip(fromIndex).limit(threshold).collect(Collectors.toList());
operater.accept(list);
fromIndex += threshold;
}
}
}

64
src/main/java/com/example/math/utils/EncryptUtil.java Normal file
View File

@ -0,0 +1,64 @@
package com.example.math.utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
/**
* 加密工具类
* @Author: rosh
*/
public final class EncryptUtil {
private EncryptUtil() {
}
private static final Logger logger = LoggerFactory.getLogger(EncryptUtil.class);
public static String encodeBase64(byte[] bytes) {
return Base64.getEncoder().encodeToString(bytes);
}
public static byte[] decodeBase64(String str) {
return Base64.getDecoder().decode(str);
}
public static String encodeUTF8StringBase64(String str) {
return Base64.getEncoder().encodeToString(str.getBytes(StandardCharsets.UTF_8));
}
public static String decodeUTF8StringBase64(String str) {
byte[] bytes = Base64.getDecoder().decode(str);
return new String(bytes, StandardCharsets.UTF_8);
}
public static String encodeURL(String url) {
String encoded = null;
try {
encoded = URLEncoder.encode(url, String.valueOf(StandardCharsets.UTF_8));
} catch (UnsupportedEncodingException e) {
logger.warn("URLEncode失败", e);
}
return encoded;
}
public static String decodeURL(String url) {
String decoded = null;
try {
decoded = URLDecoder.decode(url, String.valueOf(StandardCharsets.UTF_8));
} catch (UnsupportedEncodingException e) {
logger.warn("URLDecode失败", e);
}
return decoded;
}
}

195
src/main/java/com/example/math/utils/HttpUtils.java Normal file
View File

@ -0,0 +1,195 @@
package com.example.math.utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSession;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.net.URL;
import java.net.URLConnection;
import java.security.cert.X509Certificate;
/**
* 通用http发送方法
*/
public class HttpUtils {
private static final Logger log = LoggerFactory.getLogger(HttpUtils.class);
/**
* 向指定 URL 发送GET方法的请求
*
* @param url 发送请求的 URL
* @param param 请求参数，请求参数应该是 name1=value1&name2=value2 的形式。
* @return 所代表远程资源的响应结果
*/
public static String sendGet(String url, String param) {
StringBuilder result = new StringBuilder();
BufferedReader in = null;
try {
String urlNameString = url + "?" + param;
log.info("sendGet - {}", urlNameString);
URL realUrl = new URL(urlNameString);
URLConnection connection = realUrl.openConnection();
connection.setRequestProperty("accept", "*/*");
connection.setRequestProperty("connection", "Keep-Alive");
connection.setRequestProperty("user-agent", "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");
connection.connect();
in = new BufferedReader(new InputStreamReader(connection.getInputStream()));
String line;
while ((line = in.readLine()) != null) {
result.append(line);
}
log.info("recv - {}", result);
} catch (ConnectException e) {
log.error("调用HttpUtils.sendGet ConnectException, url=" + url + ",param=" + param, e);
} catch (SocketTimeoutException e) {
log.error("调用HttpUtils.sendGet SocketTimeoutException, url=" + url + ",param=" + param, e);
} catch (IOException e) {
log.error("调用HttpUtils.sendGet IOException, url=" + url + ",param=" + param, e);
} catch (Exception e) {
log.error("调用HttpsUtil.sendGet Exception, url=" + url + ",param=" + param, e);
} finally {
try {
if (in != null) {
in.close();
}
} catch (Exception ex) {
log.error("调用in.close Exception, url=" + url + ",param=" + param, ex);
}
}
return result.toString();
}
/**
* 向指定 URL 发送POST方法的请求
*
* @param url 发送请求的 URL
* @param param 请求参数，请求参数应该是 name1=value1&name2=value2 的形式。
* @return 所代表远程资源的响应结果
*/
public static String sendPost(String url, String param) {
PrintWriter out = null;
BufferedReader in = null;
StringBuilder result = new StringBuilder();
try {
String urlNameString = url + "?" + param;
log.info("sendPost - {}", urlNameString);
URL realUrl = new URL(urlNameString);
URLConnection conn = realUrl.openConnection();
conn.setRequestProperty("accept", "*/*");
conn.setRequestProperty("connection", "Keep-Alive");
conn.setRequestProperty("user-agent", "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");
conn.setRequestProperty("Accept-Charset", "utf-8");
conn.setRequestProperty("contentType", "utf-8");
conn.setDoOutput(true);
conn.setDoInput(true);
out = new PrintWriter(conn.getOutputStream());
out.print(param);
out.flush();
in = new BufferedReader(new InputStreamReader(conn.getInputStream(), "utf-8"));
String line;
while ((line = in.readLine()) != null) {
result.append(line);
}
log.info("recv - {}", result);
} catch (ConnectException e) {
log.error("调用HttpUtils.sendPost ConnectException, url=" + url + ",param=" + param, e);
} catch (SocketTimeoutException e) {
log.error("调用HttpUtils.sendPost SocketTimeoutException, url=" + url + ",param=" + param, e);
} catch (IOException e) {
log.error("调用HttpUtils.sendPost IOException, url=" + url + ",param=" + param, e);
} catch (Exception e) {
log.error("调用HttpsUtil.sendPost Exception, url=" + url + ",param=" + param, e);
} finally {
try {
if (out != null) {
out.close();
}
if (in != null) {
in.close();
}
} catch (IOException ex) {
log.error("调用in.close Exception, url=" + url + ",param=" + param, ex);
}
}
return result.toString();
}
public static String sendSSLPost(String url, String param) {
StringBuilder result = new StringBuilder();
String urlNameString = url + "?" + param;
try {
log.info("sendSSLPost - {}", urlNameString);
SSLContext sc = SSLContext.getInstance("SSL");
sc.init(null, new TrustManager[]{new TrustAnyTrustManager()}, new java.security.SecureRandom());
URL console = new URL(urlNameString);
HttpsURLConnection conn = (HttpsURLConnection) console.openConnection();
conn.setRequestProperty("accept", "*/*");
conn.setRequestProperty("connection", "Keep-Alive");
conn.setRequestProperty("user-agent", "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1;SV1)");
conn.setRequestProperty("Accept-Charset", "utf-8");
conn.setRequestProperty("contentType", "utf-8");
conn.setDoOutput(true);
conn.setDoInput(true);
conn.setSSLSocketFactory(sc.getSocketFactory());
conn.setHostnameVerifier(new TrustAnyHostnameVerifier());
conn.connect();
InputStream is = conn.getInputStream();
BufferedReader br = new BufferedReader(new InputStreamReader(is));
String ret = "";
while ((ret = br.readLine()) != null) {
if (ret != null && !ret.trim().equals("")) {
result.append(new String(ret.getBytes("ISO-8859-1"), "utf-8"));
}
}
log.info("recv - {}", result);
conn.disconnect();
br.close();
} catch (ConnectException e) {
log.error("调用HttpUtils.sendSSLPost ConnectException, url=" + url + ",param=" + param, e);
} catch (SocketTimeoutException e) {
log.error("调用HttpUtils.sendSSLPost SocketTimeoutException, url=" + url + ",param=" + param, e);
} catch (IOException e) {
log.error("调用HttpUtils.sendSSLPost IOException, url=" + url + ",param=" + param, e);
} catch (Exception e) {
log.error("调用HttpsUtil.sendSSLPost Exception, url=" + url + ",param=" + param, e);
}
return result.toString();
}
private static class TrustAnyTrustManager implements X509TrustManager {
@Override
public void checkClientTrusted(X509Certificate[] chain, String authType) {
}
@Override
public void checkServerTrusted(X509Certificate[] chain, String authType) {
}
@Override
public X509Certificate[] getAcceptedIssuers() {
return new X509Certificate[]{};
}
}
private static class TrustAnyHostnameVerifier implements HostnameVerifier {
@Override
public boolean verify(String hostname, SSLSession session) {
return true;
}
}
}

300
src/main/java/com/example/math/utils/IpUtils.java Normal file
View File

@ -0,0 +1,300 @@
package com.example.math.utils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
import javax.servlet.http.HttpServletRequest;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.NetworkInterface;
import java.net.ProtocolException;
import java.net.URL;
import java.net.UnknownHostException;
/**
* 获取IP工具类
*/
public class IpUtils {
private static final Logger log = LoggerFactory.getLogger(IpUtils.class);
/**
* 通过HttpServletRequest返回IP地址
*
* @param request HttpServletRequest
* @return ip String
* @throws Exception
*/
public static String getIpAddr(HttpServletRequest request) {
if (request == null) {
return "unknown";
}
String ip = request.getHeader("x-forwarded-for");
if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getHeader("Proxy-Client-IP");
}
if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getHeader("X-Forwarded-For");
}
if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getHeader("WL-Proxy-Client-IP");
}
if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getHeader("X-Real-IP");
}
if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getRemoteAddr();
}
return "0:0:0:0:0:0:0:1".equals(ip) ? "127.0.0.1" : ip;
}
public static boolean internalIp(String ip) {
byte[] addr = textToNumericFormatV4(ip);
return internalIp(addr) || "127.0.0.1".equals(ip);
}
private static boolean internalIp(byte[] addr) {
final byte b0 = addr[0];
final byte b1 = addr[1];
// 10.x.x.x/8
final byte SECTION_1 = 0x0A;
// 172.16.x.x/12
final byte SECTION_2 = (byte) 0xAC;
final byte SECTION_3 = (byte) 0x10;
final byte SECTION_4 = (byte) 0x1F;
// 192.168.x.x/16
final byte SECTION_5 = (byte) 0xC0;
final byte SECTION_6 = (byte) 0xA8;
switch (b0) {
case SECTION_1:
return true;
case SECTION_2:
if (b1 >= SECTION_3 && b1 <= SECTION_4) {
return true;
}
case SECTION_5:
switch (b1) {
case SECTION_6:
return true;
}
default:
return false;
}
}
/**
* 将IPv4地址转换成字节
*
* @param text IPv4地址
* @return byte 字节
*/
public static byte[] textToNumericFormatV4(String text) {
if (text.length() == 0) {
return null;
}
byte[] bytes = new byte[4];
String[] elements = text.split("\\.", -1);
try {
long l;
int i;
switch (elements.length) {
case 1:
l = Long.parseLong(elements[0]);
if ((l < 0L) || (l > 4294967295L))
return null;
bytes[0] = (byte) (int) (l >> 24 & 0xFF);
bytes[1] = (byte) (int) ((l & 0xFFFFFF) >> 16 & 0xFF);
bytes[2] = (byte) (int) ((l & 0xFFFF) >> 8 & 0xFF);
bytes[3] = (byte) (int) (l & 0xFF);
break;
case 2:
l = Integer.parseInt(elements[0]);
if ((l < 0L) || (l > 255L))
return null;
bytes[0] = (byte) (int) (l & 0xFF);
l = Integer.parseInt(elements[1]);
if ((l < 0L) || (l > 16777215L))
return null;
bytes[1] = (byte) (int) (l >> 16 & 0xFF);
bytes[2] = (byte) (int) ((l & 0xFFFF) >> 8 & 0xFF);
bytes[3] = (byte) (int) (l & 0xFF);
break;
case 3:
for (i = 0; i < 2; ++i) {
l = Integer.parseInt(elements[i]);
if ((l < 0L) || (l > 255L))
return null;
bytes[i] = (byte) (int) (l & 0xFF);
}
l = Integer.parseInt(elements[2]);
if ((l < 0L) || (l > 65535L))
return null;
bytes[2] = (byte) (int) (l >> 8 & 0xFF);
bytes[3] = (byte) (int) (l & 0xFF);
break;
case 4:
for (i = 0; i < 4; ++i) {
l = Integer.parseInt(elements[i]);
if ((l < 0L) || (l > 255L))
return null;
bytes[i] = (byte) (int) (l & 0xFF);
}
break;
default:
return null;
}
} catch (NumberFormatException e) {
return null;
}
return bytes;
}
public static String getHostIp() {
try {
return InetAddress.getLocalHost().getHostAddress();
} catch (UnknownHostException e) {
}
return "127.0.0.1";
}
public static String getHostName() {
try {
return InetAddress.getLocalHost().getHostName();
} catch (UnknownHostException e) {
}
return "未知";
}
/**
* 通过IP地址获取MAC地址
*
* @param ip String,127.0.0.1格式
* @return mac String
* @throws Exception
*/
public String getMACAddress(String ip) throws Exception {
String line = "";
String macAddress = "";
final String MAC_ADDRESS_PREFIX = "MAC Address = ";
final String LOOPBACK_ADDRESS = "127.0.0.1";
//如果为127.0.0.1,则获取本地MAC地址。
if (LOOPBACK_ADDRESS.equals(ip)) {
InetAddress inetAddress = InetAddress.getLocalHost();
//貌似此方法需要JDK1.6。
byte[] mac = NetworkInterface.getByInetAddress(inetAddress).getHardwareAddress();
//下面代码是把mac地址拼装成String
StringBuilder sb = new StringBuilder();
for (int i = 0; i < mac.length; i++) {
if (i != 0) {
sb.append("-");
}
//mac[i] & 0xFF 是为了把byte转化为正整数
String s = Integer.toHexString(mac[i] & 0xFF);
sb.append(s.length() == 1 ? 0 + s : s);
}
//把字符串所有小写字母改为大写成为正规的mac地址并返回
macAddress = sb.toString().trim().toUpperCase();
return macAddress;
}
//获取非本地IP的MAC地址
try {
Process p = Runtime.getRuntime().exec("nbtstat -A " + ip);
InputStreamReader isr = new InputStreamReader(p.getInputStream());
BufferedReader br = new BufferedReader(isr);
while ((line = br.readLine()) != null) {
if (line != null) {
int index = line.indexOf(MAC_ADDRESS_PREFIX);
if (index != -1) {
macAddress = line.substring(index + MAC_ADDRESS_PREFIX.length()).trim().toUpperCase();
}
}
}
br.close();
} catch (IOException e) {
e.printStackTrace(System.out);
}
return macAddress;
}
/**
* 通过IP获取地址(需要联网，调用淘宝的IP库)
*
* @param ip IP地址
* @return
*/
public static String getIpInfo(String ip) {
if ("127.0.0.1".equals(ip)) {
ip = "127.0.0.1";
}
String info = "";
try {
URL url = new URL("http://ip.taobao.com/service/getIpInfo.php?ip=" + ip);
HttpURLConnection htpcon = (HttpURLConnection) url.openConnection();
htpcon.setRequestMethod("GET");
htpcon.setDoOutput(true);
htpcon.setDoInput(true);
htpcon.setUseCaches(false);
InputStream in = htpcon.getInputStream();
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(in));
StringBuffer temp = new StringBuffer();
String line = bufferedReader.readLine();
while (line != null) {
temp.append(line).append("\r\n");
line = bufferedReader.readLine();
}
bufferedReader.close();
JSONObject obj = (JSONObject) JSON.parse(temp.toString());
if (obj.getIntValue("code") == 0) {
JSONObject data = obj.getJSONObject("data");
info += data.getString("country") + " ";
info += data.getString("region") + " ";
info += data.getString("city") + " ";
info += data.getString("isp");
}
} catch (MalformedURLException e) {
e.printStackTrace();
} catch (ProtocolException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
return info;
}
/**
* 根据IP查询地区
*
* @param ip IP地址
* @return
*/
public static String getRealAddressByIP(String ip) {
String address = "XX XX";
// 内网不查询
if (IpUtils.internalIp(ip)) {
return "内网IP";
}
String rspStr = HttpUtils.sendPost("http://ip.taobao.com/service/getIpInfo.php", "ip=" + ip);
if (StringUtils.isEmpty(rspStr)) {
log.error("获取地理位置异常 {}", ip);
return address;
}
JSONObject obj = JSONObject.parseObject(rspStr);
JSONObject data = obj.getObject("data", JSONObject.class);
String region = data.getString("region");
String city = data.getString("city");
address = region + " " + city;
return address;
}
}

0
src/main/java/com/example/math/utils/JsonResult.java Normal file
View File

609
src/main/java/com/example/math/utils/RedisUtils.java Normal file
View File

@ -0,0 +1,609 @@
package com.example.math.utils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
public class RedisUtils {
private RedisTemplate<String, Object> redisTemplate;
public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
this.redisTemplate = redisTemplate;
}
// =============================common============================
/**
* 普通缓存获取
*
* @param
* @return 值
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public Set<String> keys(String pattern) {
return redisTemplate.keys(pattern);
}
/**
* 指定缓存失效时间
*
* @param key 键
* @param time 时间(秒)
* @return
*/
public boolean expire(String key, long time) {
try {
if (time > 0) {
redisTemplate.expire(key, time, TimeUnit.SECONDS);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 根据key 获取过期时间
*
* @param key 键 不能为null
* @return 时间(秒) 返回0代表为永久有效
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public long getExpire(String key) {
return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
/**
* 判断key是否存在
*
* @param key 键
* @return true 存在 false不存在
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public boolean hasKey(String key) {
try {
return redisTemplate.hasKey(key);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
// /**
// * 删除缓存
// *
// * @param key 可以传一个值 或多个
// */
// @SuppressWarnings("unchecked")
// public void del(String... key) {
// if (key != null && key.length > 0) {
// if (key.length == 1) {
// redisTemplate.delete(key[0]);
// } else {
// redisTemplate.delete(CollectionUtils.arrayToList(key));
// }
// }
// }
// ============================String=============================
/**
* 普通缓存获取
*
* @param key 键
* @return 值
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public Object get(String key) {
return key == null ? null : redisTemplate.opsForValue().get(key);
}
/**
* 普通缓存放入
*
* @param key 键
* @param value 值
* @return true成功 false失败
*/
public boolean set(String key, Object value) {
try {
redisTemplate.opsForValue().set(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 普通缓存放入并设置时间
*
* @param key 键
* @param value 值
* @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期
* @return true成功 false 失败
*/
public boolean set(String key, Object value, long time) {
try {
if (time > 0) {
redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
} else {
set(key, value);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 递增
*
* @param key 键
* @param delta 要增加几(大于0)
* @return
*/
public long incr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递增因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, delta);
}
/**
* 递减
*
* @param key 键
* @param delta 要减少几(小于0)
* @return
*/
public long decr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递减因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, -delta);
}
// ================================Map=================================
/**
* HashGet
*
* @param key 键 不能为null
* @param item 项 不能为null
* @return 值
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public Object hget(String key, String item) {
return redisTemplate.opsForHash().get(key, item);
}
/**
* 获取hashKey对应的所有键值
*
* @param key 键
* @return 对应的多个键值
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public Map<Object, Object> hmget(String key) {
return redisTemplate.opsForHash().entries(key);
}
/**
* HashSet
*
* @param key 键
* @param map 对应多个键值
* @return true 成功 false 失败
*/
public boolean hmset(String key, Map<String, Object> map) {
try {
redisTemplate.opsForHash().putAll(key, map);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* HashSet 并设置时间
*
* @param key 键
* @param map 对应多个键值
* @param time 时间(秒)
* @return true成功 false失败
*/
public boolean hmset(String key, Map<String, Object> map, long time) {
try {
redisTemplate.opsForHash().putAll(key, map);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
*
* @param key 键
* @param item 项
* @param value 值
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value) {
try {
redisTemplate.opsForHash().put(key, item, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
*
* @param key 键
* @param item 项
* @param value 值
* @param time 时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value, long time) {
try {
redisTemplate.opsForHash().put(key, item, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 删除hash表中的值
*
* @param key 键 不能为null
* @param item 项 可以使多个 不能为null
*/
public void hdel(String key, Object... item) {
redisTemplate.opsForHash().delete(key, item);
}
/**
* 判断hash表中是否有该项的值
*
* @param key 键 不能为null
* @param item 项 不能为null
* @return true 存在 false不存在
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public boolean hHasKey(String key, String item) {
return redisTemplate.opsForHash().hasKey(key, item);
}
/**
* hash递增 如果不存在,就会创建一个 并把新增后的值返回
*
* @param key 键
* @param item 项
* @param by 要增加几(大于0)
* @return
*/
public double hincr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, by);
}
/**
* hash递减
*
* @param key 键
* @param item 项
* @param by 要减少记(小于0)
* @return
*/
public double hdecr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, -by);
}
// ============================set=============================
/**
* 根据key获取Set中的所有值
*
* @param key 键
* @return
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public Set<Object> sGet(String key) {
try {
return redisTemplate.opsForSet().members(key);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 根据value从一个set中查询,是否存在
*
* @param key 键
* @param value 值
* @return true 存在 false不存在
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public boolean sHasKey(String key, Object value) {
try {
return redisTemplate.opsForSet().isMember(key, value);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将数据放入set缓存
*
* @param key 键
* @param values 值 可以是多个
* @return 成功个数
*/
public long sSet(String key, Object... values) {
try {
return redisTemplate.opsForSet().add(key, values);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 将set数据放入缓存
*
* @param key 键
* @param time 时间(秒)
* @param values 值 可以是多个
* @return 成功个数
*/
public long sSetAndTime(String key, long time, Object... values) {
try {
Long count = redisTemplate.opsForSet().add(key, values);
if (time > 0) {
expire(key, time);
}
return count;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 获取set缓存的长度
*
* @param key 键
* @return
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public long sGetSetSize(String key) {
try {
return redisTemplate.opsForSet().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 移除值为value的
*
* @param key 键
* @param values 值 可以是多个
* @return 移除的个数
*/
public long setRemove(String key, Object... values) {
try {
Long count = redisTemplate.opsForSet().remove(key, values);
return count;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
// ===============================list=================================
/**
* 获取list缓存的内容
*
* @param key 键
* @param start 开始
* @param end 结束 0 到 -1代表所有值
* @return
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public List<Object> lGet(String key, long start, long end) {
try {
return redisTemplate.opsForList().range(key, start, end);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 获取list缓存的长度
*
* @param key 键
* @return
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public long lGetListSize(String key) {
try {
return redisTemplate.opsForList().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 通过索引 获取list中的值
*
* @param key 键
* @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
* @return
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public Object lGetIndex(String key, long index) {
try {
return redisTemplate.opsForList().index(key, index);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, Object value) {
try {
redisTemplate.opsForList().rightPush(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, Object value, long time) {
try {
redisTemplate.opsForList().rightPush(key, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, List<Object> value) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, List<Object> value, long time) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 根据索引修改list中的某条数据
*
* @param key 键
* @param index 索引
* @param value 值
* @return
*/
public boolean lUpdateIndex(String key, long index, Object value) {
try {
redisTemplate.opsForList().set(key, index, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 移除N个值为value
*
* @param key 键
* @param count 移除多少个
* @param value 值
* @return 移除的个数
*/
public long lRemove(String key, long count, Object value) {
try {
Long remove = redisTemplate.opsForList().remove(key, count, value);
return remove;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 按键的排序获取对应的值
*
* @param keys 多个键
* @return List<Object>
*/
@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
public List<Object> mget(Collection<String> keys) {
return redisTemplate.opsForValue().multiGet(keys);
}
}

62
src/main/java/com/example/math/utils/SslUtils.java Normal file
View File

@ -0,0 +1,62 @@
package com.example.math.utils;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSession;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
public class SslUtils {
private static void trustAllHttpsCertificates() throws Exception {
TrustManager[] trustAllCerts = new TrustManager[1];
TrustManager tm = new miTM();
trustAllCerts[0] = tm;
SSLContext sc = SSLContext.getInstance("SSL");
sc.init(null, trustAllCerts, null);
HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory());
}
static class miTM implements TrustManager, X509TrustManager {
public X509Certificate[] getAcceptedIssuers() {
return null;
}
public boolean isServerTrusted(X509Certificate[] certs) {
return true;
}
public boolean isClientTrusted(X509Certificate[] certs) {
return true;
}
public void checkServerTrusted(X509Certificate[] certs, String authType)
throws CertificateException {
return;
}
public void checkClientTrusted(X509Certificate[] certs, String authType)
throws CertificateException {
return;
}
}
/**
* 忽略HTTPS请求的SSL证书，必须在openConnection之前调用
*
* @throws Exception
*/
public static void ignoreSsl() throws Exception {
HostnameVerifier hv = new HostnameVerifier() {
public boolean verify(String urlHostName, SSLSession session) {
System.out.println("Warning: URL Host: " + urlHostName + " vs. " + session.getPeerHost());
return true;
}
};
trustAllHttpsCertificates();
HttpsURLConnection.setDefaultHostnameVerifier(hv);
}
}

62
src/main/resources/application.yml Normal file
View File

@ -0,0 +1,62 @@
server:
servlet:
context-path: /
port: 9999
spring:
application:
name: provider
#MQTT配置信息
mqtt:
#MQTT服务地址，端口号默认11883，如果有多个，用逗号隔开
url: tcp://58.213.148.44:1883
#用户名
username: admin
#密码
password: admin##123
#客户端id(不能重复)
client:
id: airport_
#MQTT默认的消息推送主题，实际可在调用接口是指定
default:
topic: topic
newUrl: 58.213.148.44:1883
# djiClientId: airport_dji_mqtt
# mqtt:
# onlineUrl: http://101.133.163.127:18083
# #MQTT服务地址，端口号默认11883，如果有多个，用逗号隔开
# url: tcp://101.133.163.127:1883
# #用户名
# username: admin
# #密码
# password: admin##123
# #客户端id(不能重复)
# client:
# id: provider-id
# #MQTT默认的消息推送主题，实际可在调用接口是指定
# default:
# topic: topic
# newClient:
# id: DJIprovider-id
# newUrl: 106.15.64.139:1883
# djiClientId: airport_dji_mqtt
redis:
# 缓存库默认索引0
database: 9
# Redis服务器地址
host: 127.0.0.1
# Redis服务器连接端口
port: 6379
# Redis服务器连接密码（默认为空）
password:
# 连接超时时间（毫秒）
timeout: 30000
# 默认的数据过期时间，主要用于shiro权限管理
expire: 2592000
jedis:
pool:
max-active: 1000 # 连接池最大连接数（使用负值表示没有限制）
max-wait: -1 # 连接池最大阻塞等待时间（使用负值表示没有限制）
max-idle: 10 # 连接池中的最大空闲连接
min-idle: 1 # 连接池中的最小空闲连接