钟来 / Luhui · 提交

转到一个项目

作者 钟来
提交 0ec3873c03ae1725beb1bec1d31c92367eb72e3b 1 个父辈 846e15c4

删除透传模块，迁移到单独项目

内嵌 并排对比
正在显示 13 个修改的文件 包含 0 行增加 和 462 行删除
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.zhonglai.luhui</groupId>
<artifactId>lh-neutrino-proxy</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
<artifactId>lh-neutrino-proxy-client</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>com.zhonglai.luhui</groupId>
<artifactId>lh-neutrino-proxy-common</artifactId>
</dependency>
</dependencies>
</project>
\ No newline at end of file
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.zhonglai.luhui</groupId>
<artifactId>lh-neutrino-proxy</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
<artifactId>lh-neutrino-proxy-common</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</dependency>
<!-- mqtt -->
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
</dependency>
</dependencies>
</project>
\ No newline at end of file
package com.zhonglai.luhui.neutrino.proxy.common.mqtt;
import com.alibaba.fastjson.JSONObject;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Set;
/**
* mqtt回调
*/
public class MqttCallback implements MqttCallbackExtended {
private final Logger log = LoggerFactory.getLogger(this.getClass());
private MqttService mqttclient;
public MqttCallback(MqttService mqttclient)
{
this.mqttclient = mqttclient;
}
@Override
public void connectComplete(boolean b, String s) {
log.info("连接成功");
Set<String> topics = mqttclient.subscribe();
log.info("-----------订阅成功:{}--------------------",topics);
}
@Override
public void connectionLost(Throwable cause) {
log.error("连接丢失重新链接",cause);
while (!mqttclient.connect())
{
try {
Thread.sleep(60000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
log.info("mqtt发来消息>>>>>>>：{} 《===============》 字符串:【{}】",topic,message.toString());
String str = new String(message.getPayload());
mqttclient.operate(new Topic(topic),JSONObject.parseObject(str));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
log.info("消息发送完成");
}
}
package com.zhonglai.luhui.neutrino.proxy.common.mqtt;
import java.io.FileReader;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
public class MqttConfig {
public static String mqttBroker;
public static String mqttUserName;
public static String mqttPassword;
public static String mqttClientId;
public static Set<String> subTopic;
public static void init()
{
Properties properties = loadProperties();
mqttBroker = properties.getProperty("broker");
mqttUserName = properties.getProperty("username");
mqttPassword = properties.getProperty("password");
mqttClientId = properties.getProperty("clientId");
subTopic = new HashSet<>();
String topicsStr = properties.getProperty("subTopic");
if(null != topicsStr && !"".equals(topicsStr))
{
if (topicsStr != null && !topicsStr.trim().isEmpty()) {
String[] topics = topicsStr.split(",");
for (String topic : topics) {
subTopic.add(topic.trim());
}
}
}
}
public static Properties loadProperties()
{
Properties properties = new Properties();
try {
if(null != System.getProperty("configPath") && !"".equals(System.getProperty("configPath")))
{
properties.load(new FileReader(System.getProperty("configPath")+"/mqtt.properties"));
}else{
properties.load(MqttConfig.class.getClassLoader().getResourceAsStream("mqtt.properties"));
}
} catch (Exception e) {
e.printStackTrace();
}
return properties;
}
}
package com.zhonglai.luhui.neutrino.proxy.common.mqtt;
import com.alibaba.fastjson.JSONObject;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.PreDestroy;
import java.util.Set;
/**
* 通过mqtt实现消息推送，用来处理长链接
*/
public class MqttService {
private final Logger log = LoggerFactory.getLogger(this.getClass());
private MqttClient mqttclient;
private MqttConnectOptions options;
private OperateService operateService;
// 使用 volatile 确保多线程内存可见性和禁止指令重排
private static volatile MqttService instance;
private MqttService(OperateService operateService) {
this.operateService = operateService;
// 初始化代码
try {
sart();
} catch (MqttException e) {
throw new RuntimeException(e);
}
}
public static MqttService getInstance( OperateService operateService) {
if (instance == null) {
synchronized (MqttService.class) {
if (instance == null) {
instance = new MqttService(operateService);
}
}
}
return instance;
}
private void init() throws MqttException {
if(null == mqttclient)
{
mqttclient = new MqttClient(MqttConfig.mqttBroker, MqttConfig.mqttClientId, new MemoryPersistence());
}
options = new MqttConnectOptions();
options.setCleanSession(true);
options.setConnectionTimeout(15);
options.setKeepAliveInterval(60); // 添加心跳设置，单位为秒
//设置断开后重新连接
options.setAutomaticReconnect(true);
mqttclient.setCallback(new MqttCallback(this));
}
public boolean connect() {
try {
options.setUserName(MqttConfig.mqttUserName);
options.setPassword(MqttConfig.mqttPassword.toCharArray());
mqttclient.connect(options);
return true;
} catch (MqttException e) {
log.error("-----------mqtt连接服务器失败--------------------",e);
}
return false;
}
public void sart() throws MqttException{
MqttConfig.init();
log.info("-----------开始启动mqtt监听服务--------------------");
init();
connect();
log.info("-----------mqtt连接服务器成功--------------------");
}
@PreDestroy
public void stop() throws MqttException {
if(null != mqttclient)
{
log.info("退出mqtt服务");
mqttclient.unsubscribe(MqttConfig.subTopic.toArray(new String[MqttConfig.subTopic.size()]));
mqttclient.disconnect();
mqttclient.close();
}
instance = null;
}
public void publish(String topic, String messageStr) throws MqttException {
MqttMessage message = new MqttMessage();
message.setPayload(messageStr.getBytes());
mqttclient.publish(topic,message);
}
public Set<String> subscribe()
{
try {
mqttclient.subscribe(MqttConfig.subTopic.toArray(new String[MqttConfig.subTopic.size()]));
} catch (MqttException e) {
e.printStackTrace();
}
return MqttConfig.subTopic;
}
public boolean isConnect()
{
return mqttclient != null && mqttclient.isConnected();
}
public void operate(Topic topic, JSONObject payloaddata)
{
operateService.operate(topic,payloaddata);
}
}
package com.zhonglai.luhui.neutrino.proxy.common.mqtt;
import com.alibaba.fastjson.JSONObject;
public interface OperateService {
/**
* 指令操作
* @param topic
* @param data
*/
void operate(Topic topic, JSONObject data);
}
package com.zhonglai.luhui.neutrino.proxy.common.mqtt;
public class Topic {
public static final String TOPIC_PUT = "PUT";
public static final String TOPIC_READ = "READ";
private String topicType;
private String time;
public Topic()
{
}
public Topic(String stopicStr)
{
String[] strs = stopicStr.split("/");
topicType = strs[0];
time = strs[1];
}
public String getTopicType() {
return topicType;
}
public void setTopicType(String topicType) {
this.topicType = topicType;
}
public String getTime() {
return time;
}
public void setTime(String time) {
this.time = time;
}
}
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.zhonglai.luhui</groupId>
<artifactId>lh-neutrino-proxy</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
<artifactId>lh-neutrino-proxy-server</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>
<dependency>
<groupId>com.zhonglai.luhui</groupId>
<artifactId>lh-neutrino-proxy-common</artifactId>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
</dependencies>
</project>
\ No newline at end of file
package com.zhonglai.luhui.neutrino.proxy.server;
import com.zhonglai.luhui.neutrino.proxy.common.mqtt.MqttConfig;
import com.zhonglai.luhui.neutrino.proxy.common.mqtt.MqttService;
import com.zhonglai.luhui.neutrino.proxy.server.operate.OperateServiceImpl;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ControlServer {
private static Logger logger = LoggerFactory.getLogger(ControlServer.class);
public static void main(String[] args) throws Exception {
MqttService mqttService = MqttService.getInstance(new OperateServiceImpl());
mqttService.sart();
// 使用Lambda表达式添加关闭钩子
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
logger.info("正在关闭应用程序...");
try {
mqttService.stop();
} catch (MqttException e) {
logger.error("MQTT服务停止异常", e);
}
logger.info("应用程序关闭成功");
}));
}
}
package com.zhonglai.luhui.neutrino.proxy.server.operate;
import com.alibaba.fastjson.JSONObject;
import com.zhonglai.luhui.neutrino.proxy.common.mqtt.OperateService;
import com.zhonglai.luhui.neutrino.proxy.common.mqtt.Topic;
public class OperateServiceImpl implements OperateService {
@Override
public void operate(Topic topic, JSONObject data) {
}
}
broker=
username=
password=
clientId=
subTopic=
\ No newline at end of file
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.zhonglai.luhui</groupId>
<artifactId>lh-modules</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
<artifactId>lh-neutrino-proxy</artifactId>
<modules>
<module>lh-neutrino-proxy-server</module>
<module>lh-neutrino-proxy-client</module>
<module>lh-neutrino-proxy-common</module>
</modules>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<packaging>pom</packaging>
<description>
透传代理
</description>
</project>
\ No newline at end of file
... ... @@ -38,7 +38,6 @@
<module>lh-ssh-service-lesten</module>
<module>lh-deviceInfo-sync</module>
<module>lh-camera</module>
<module>lh-neutrino-proxy</module>
</modules>
<properties>
... ...