package com.dy.rtuMw;
|
|
import java.util.ArrayList;
|
import java.util.List;
|
|
import com.dy.common.util.ConfigProperties;
|
import com.dy.common.util.IPUtils;
|
import com.dy.rtuMw.server.*;
|
import com.dy.rtuMw.server.mqtt.MqttUnit;
|
import com.dy.rtuMw.server.mqtt.MqttUnitConfigVo;
|
import com.dy.rtuMw.server.msCenter.MsCenterConfigVo;
|
import com.dy.rtuMw.server.msCenter.MsCenterUnit;
|
import com.dy.rtuMw.server.rtuData.RtuDataUnit;
|
import com.dy.rtuMw.server.rtuData.RtuDataUnitConfigVo;
|
import com.dy.rtuMw.server.tasks.*;
|
import com.dy.common.mw.UnitInterface;
|
import com.dy.common.mw.channel.tcp.TcpConfigVo;
|
import com.dy.common.mw.channel.tcp.TcpUnit;
|
import com.dy.common.mw.core.CoreUnit;
|
import com.dy.common.mw.core.CoreUnitConfigVo;
|
import com.dy.common.mw.protocol.ProtocolConfigVo;
|
import com.dy.common.mw.protocol.ProtocolUnit;
|
import com.dy.common.mw.support.SupportUnit;
|
import com.dy.common.mw.support.SupportUnitConfigVo;
|
import com.dy.rtuMw.resource.ResourceUnit;
|
import com.dy.rtuMw.resource.ResourceUnitConfigVo;
|
import com.dy.common.springUtil.SpringContextUtil;
|
import com.dy.common.util.ConfigXml4Springboot;
|
import com.dy.common.util.IDLongGenerator;
|
|
import com.dy.rtuMw.server.upgrade.UpgradeUnit;
|
import com.dy.rtuMw.server.upgrade.UpgradeUnitConfigVo;
|
import org.jdom2.Document;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.core.io.ResourceLoader;
|
import org.springframework.stereotype.Component;
|
|
@Component
|
public class Server {
|
|
private ConfigXml4Springboot conf = null ;
|
private Document doc = null ;
|
private boolean showStartInfo = false ;
|
|
private String orgTag ;
|
private String RmiSvUrl ;
|
private String TcpSvUrl ;
|
@Value("${server.port}")
|
private String HttpSvPort ;
|
@Value("${server.servlet.context-path}")
|
private String HttpSvPath ;
|
|
@Autowired
|
protected ResourceLoader resourceLoader ;
|
|
private List<UnitInterface> units = new ArrayList<UnitInterface>() ;
|
|
/*
|
* @param args 参数
|
public static void main(String[] args) {
|
new Server().startServer();
|
}
|
*/
|
|
public void startServer(){
|
/**
|
try {
|
URL url = Server.class.getResource("/config/this.licence");
|
if(!new Lnp(null).parese(url.getPath())){
|
System.out.println("licence error!") ;
|
return ;
|
}
|
} catch (Exception e) {
|
System.out.println("licence error!") ;
|
return ;
|
}
|
*/
|
if(this.doStartServer()){
|
ServerShutDownHook.OnShutDown();
|
}
|
}
|
/**
|
* 启动服务
|
*/
|
private boolean doStartServer(){
|
boolean running = false ;
|
long start = System.currentTimeMillis() ;
|
try {
|
ConfigProperties.init(this.getClass().getResourceAsStream("/config.properties"), false);
|
|
this.conf = new ConfigXml4Springboot() ;
|
this.doc = this.conf.createDom(this.resourceLoader, "config.xml") ;
|
////////////////
|
//服务 配置
|
this.showStartInfo = this.conf.getSetAttrBoolean(this.doc, "config.server", "showStartInfo", null, null) ;
|
|
this.startUnits() ;
|
|
String svName ;
|
try{
|
svName = this.conf.getSetAttrTxt(this.doc, "config.server", "name", null, false, null) ;
|
}catch(Exception e){
|
svName = "" ;
|
}
|
svName += (ServerProperties.isLowPower?"（低功耗）": "") + "-" ;
|
|
String company ;
|
try{
|
company = this.conf.getSetAttrTxt(this.doc, "config.server", "company", null, true, null) ;
|
}catch(Exception e){
|
company = "" ;
|
}
|
System.out.println("$$$$$$$$$$$$ $$$$$$$$ $$$$$$$$") ;
|
System.out.println("@@@@@@@@@@@@@@@@#$ $@@@@@@@@& @@@@@@@@#") ;
|
System.out.println("@@@@@@@@@@@@@@@@@@@# @@@@@@@@# $@@@@@@@@&") ;
|
System.out.println("@@@@@@@@@@@@@@@@@@@@@# #@@@@@@@@@@@@@@@@$") ;
|
System.out.println("@@@@@@@@@@@@@@@@@@@@@@@ &@@@@@@@@@@@@@@") ;
|
System.out.println("@@@@@@$ $@@@@@@@@@& $@@@@@@@@@@@#") ;
|
System.out.println("@@@@@@$ @@@@@@@@@ @@@@@@@@@& " + this.orgTag + svName + "RtuMw 1.0.00" ) ;
|
if(this.HttpSvPath != null && this.HttpSvPort != null){
|
System.out.println("@@@@@@$ O@@@@@@@@@ &@@@@@@@@ HttpSv [ip]:" + this.HttpSvPort + this.HttpSvPath) ;
|
}else{
|
System.out.println("@@@@@@$ O@@@@@@@@@ &@@@@@@@@") ;
|
}
|
|
if(this.TcpSvUrl != null){
|
System.out.println("@@@@@@$ O@@@@@@@@@ &@@@@@@@@ TcpSv " + this.TcpSvUrl ) ;
|
}else{
|
System.out.println("@@@@@@$ O@@@@@@@@@ &@@@@@@@@") ;
|
}
|
if(this.RmiSvUrl != null){
|
System.out.println("@@@@@@$ #@@@@@@@@@$ &@@@@@@@@ RmiSv " + this.RmiSvUrl ) ;
|
}else{
|
System.out.println("@@@@@@$ #@@@@@@@@@$ &@@@@@@@@" ) ;
|
}
|
System.out.println("@@@@@@@@@@@@@@@@@@@@@@# &@@@@@@@@ Running in standalone mode" ) ;
|
System.out.println("@@@@@@@@@@@@@@@@@@@@@& &@@@@@@@@ Startup in " + (System.currentTimeMillis() - start) + " MS" ) ;
|
System.out.println("@@@@@@@@@@@@@@@@@@@# &@@@@@@@@ " + company) ;
|
System.out.println("@@@@@@@@@@@@@@@@#O &@@@@@@@@") ;
|
running = true ;
|
}catch(Exception e){
|
e.printStackTrace();
|
running = false ;
|
}
|
return running ;
|
}
|
|
private void startUnits(){
|
try {
|
///////////////
|
//基本配置
|
ServerProperties.orgTag = this.conf.getSetAttrTxt(this.doc, "config.base", "orgTag", null, false, null) ;
|
if(ServerProperties.orgTag == null || ServerProperties.orgTag.trim().equals("")){
|
throw new Exception("机构tag不能为空") ;
|
}
|
this.orgTag = ServerProperties.orgTag ;
|
ServerProperties.isLowPower = conf.getSetAttrBoolean(doc, "config.base", "isLowPower", null, null) ;
|
if(ServerProperties.isLowPower == null){
|
ServerProperties.isLowPower = false ;
|
}
|
|
String onlyDealRtus = conf.getSetAttrTxt(doc, "config.base", "onlyDealRtus", null, true, null) ;
|
if(onlyDealRtus == null || onlyDealRtus.trim().equals("")){
|
ServerProperties.onlyDealRtusTest = false ;
|
}else{
|
onlyDealRtus = onlyDealRtus.replaceAll("，", ",") ;
|
onlyDealRtus = onlyDealRtus.replaceAll(" ", "") ;
|
ServerProperties.onlyDealRtus = onlyDealRtus.split(",") ;
|
if(ServerProperties.onlyDealRtus != null && ServerProperties.onlyDealRtus.length > 0){
|
ServerProperties.onlyDealRtusTest = true ;
|
}else{
|
ServerProperties.onlyDealRtusTest = false ;
|
}
|
}
|
|
//在支持多通信协议时，只有RTU上线了，才能识别出该RTU实际用的通信协议，进而用此协议解析上行数据及构造下行命令。
|
//如果RTU未曾上线（通信中间件启动后该RTU未曾上线），那么在向它发送下行命令时，不能判断出其采用的协议，也不能构造命令，尤其是低功耗情况下，不上线是正常现象。
|
// 当只有一个协议情况下，在RTU未曾上线时，也能用此协议构造命令并缓存下来，尤其适合低功耗情况。
|
//onlyOne=true: 通信中间件当前只有一个协议
|
ServerProperties.onlyOneProtocol = conf.getSetAttrBoolean(doc, "config.base", "onlyOneProtocol", null, null) ;
|
if(ServerProperties.onlyOneProtocol == null){
|
ServerProperties.onlyOneProtocol = false ;
|
}
|
//下行命令允许发送的最大次数X，即允许重发X-1
|
ServerProperties.downComandMaxResendTimes = conf.getSetAttrPlusInt(doc, "config.base", "downComandMaxResendTimes", null, 1, 5, null).byteValue() ;
|
//针对一个RTU，下发命令的时间间隔
|
ServerProperties.commandSendInterval = conf.getSetAttrPlusInt(doc, "config.base", "commandSendInterval", null, 1, 40, null) * 1000L ;
|
//针对一个RTU，下发快速命令的时间间隔
|
ServerProperties.fastCommandSendInterval = conf.getSetAttrPlusInt(doc, "config.base", "fastCommandSendInterval", null, 1, 40000, null) * 1L ;
|
//命令已经发送达最大次数，仍未收到命令结果，需要在缓存继续等待，其等待最大时长
|
ServerProperties.cacheWaitResultTimeout = conf.getSetAttrPlusInt(doc, "config.base", "cacheWaitResultTimeout", null, 10, 360, null) * 1000L ;
|
//不在线缓存的命令最大缓存时长
|
ServerProperties.offLineCacheTimeout = conf.getSetAttrPlusInt(doc, "config.base", "offLineCacheTimeout", null, 15, 172800, null) * 1000L ;
|
//TCP上行数据时刻缓存时长，当达到时长时，TCP上行数据时刻被清空，采用TCP上行数据时刻目的是，阻止上数据同时下发数据，因为RTU处理不过来
|
ServerProperties.lastUpDataTimeLive = conf.getSetAttrPlusInt(doc, "config.base", "lastUpDataTimeLive", null, 0, 5000, null) * 1L ;
|
//数据库数据id生成器的id后缀，0是默认的后缀，一般web系统应用，数据中间件id后缀大于等于1
|
ServerProperties.dbDataIdSuffix = conf.getSetAttrInt(doc, "config.base", "dbDataIdSuffix", null, 0, 99, null);
|
//上下行数据缓存队列中缓存数据个数的报警量，这个与现实项目所接水表数相关
|
ServerProperties.cacheUpDownDataWarnCount = conf.getSetAttrPlusInt(doc, "config.base", "cacheUpDownDataWarnCount", null, 1, null, null) ;
|
//上下行数据缓存队列中缓存数据个数的最大值，这个与现实项目所接水表数相关
|
ServerProperties.cacheUpDownDataMaxCount = conf.getSetAttrPlusInt(doc, "config.base", "cacheUpDownDataMaxCount", null, 1, null, null) ;
|
if(ServerProperties.cacheUpDownDataMaxCount <= ServerProperties.cacheUpDownDataWarnCount){
|
throw new Exception("cacheUpDownDataMaxCount必须大于cacheUpDownDataWarnCount") ;
|
}
|
//没有上行数据的分钟数，达到这个分钟数，认为RTU断网了，取值范围是2-100
|
ServerProperties.disconnectedByNoUpDataMinutes = 0L + conf.getSetAttrPlusInt(doc, "config.base", "disconnectedByNoUpDataMinutes", null, 1, 100, null) ;
|
if(ServerProperties.disconnectedByNoUpDataMinutes < 1 || ServerProperties.disconnectedByNoUpDataMinutes > 100){
|
throw new Exception("disconnectedByNoUpDataMinutes取值必须是1~100") ;
|
}
|
ServerProperties.disconnectedByNoUpDataMinutes = ServerProperties.disconnectedByNoUpDataMinutes * 60 * 1000 ;
|
|
//工作报太频繁，N次上报处理1次，取值范围是1-100
|
ServerProperties.workReportDealOneByTimes = conf.getSetAttrPlusInt(doc, "config.base", "workReportDealOneByTimes", null, 1, 100, null) ;
|
|
//触发发送钉钉报警消息的取水口日漏损量的最小值（包括但除0.0外）
|
ServerProperties.intakeAlarmLossMinValue = conf.getSetAttrPlusDouble(doc, "config.base", "intakeAlarmLossMinValue", null, 0.0, 1000000.0, null) ;
|
|
//有报警发生时，向钉钉发送消息的间隔时长（分钟）
|
ServerProperties.sendDingDingAlarmMsInterval = conf.getSetAttrPlusInt(doc, "config.base", "sendDingDingAlarmMsInterval", null, 1, 600, null) ;
|
|
//设置ID生成器的后缀
|
IDLongGenerator.setSuffix(ServerProperties.dbDataIdSuffix.intValue());
|
|
|
////////////////
|
//协议模块
|
ProtocolConfigVo protoVo = new ProtocolConfigVo();
|
protoVo.centerAddr = conf.getSetAttrPlusInt(doc, "config.protocol", "centerAddr", null, 1, 4, null) ;
|
protoVo.synchroRtuClock = conf.getSetAttrBoolean(doc, "config.protocol", "synchroRtuClock", null, null) ;
|
//当RTU与服务器时钟相差一定秒钟后，进行校时
|
protoVo.synchroRtuClockTimepieces = 1000 * conf.getSetAttrPlusInt(doc, "config.protocol", "synchroRtuClockTimepieces", null, null, null, null) ;
|
protoVo.showStartInfo = showStartInfo ;
|
AdapterImp_ProtocolUnit protoAdap = new AdapterImp_ProtocolUnit();
|
protoAdap.setConfig(protoVo);
|
ProtocolUnit protoUnit = ProtocolUnit.getInstance();
|
protoUnit.setAdapter(protoAdap);
|
protoUnit.start(obj -> {
|
});
|
units.add(protoUnit) ;
|
|
////////////////////////////////////////////////////////
|
//支持模块： springHibernate和 线程池
|
SupportUnitConfigVo supVo = new SupportUnitConfigVo() ;
|
//短工作时长线程池，线程负责用时较短的工作任务
|
supVo.short_maxThread = conf.getSetAttrPlusInt(doc, "config.support", "short_maxThread", null, -1, 1000, null) ;//池中最大线程数为所有CPU核数+1
|
if(supVo.short_maxThread < 0){
|
supVo.short_maxThread = -1 ;
|
}
|
supVo.short_minThread = conf.getSetAttrPlusInt(doc, "config.support", "short_minThread", null, -1, 100, null) ;//池中最小线程数
|
if(supVo.short_minThread < 0){
|
supVo.short_minThread = -1 ;
|
}
|
supVo.short_freeTimeout = conf.getSetAttrPlusInt(doc, "config.support", "short_freeTimeout", null, 1, 90, null) * 1000 ;//线程数空闲时长，若池中线程数量大于minThread，且有的线程空闲时长超过freeTimeout，则清除该线程，为了不清除，把minThread与maxThread设置相等
|
supVo.short_busyTimeout = conf.getSetAttrPlusInt(doc, "config.support", "short_busyTimeout", null, 1, 10, null) * 1000 ;//线程不间断工作时长（单位为秒)超时限，认为线程已经了崩溃，将强制清除，短工作时长设置为5秒
|
if(supVo.short_maxThread == 0 || supVo.short_minThread == 0){
|
supVo.enableShortThreadPool = false ;
|
}else{
|
supVo.enableShortThreadPool = true ;
|
}
|
|
//长工作时长线程池，线程负责用时较长的工作任务
|
supVo.long_maxThread = conf.getSetAttrInt(doc, "config.support", "long_maxThread", null, -1, 1000, null) ;//池中最大线程数,若为-1，不受限制
|
if(supVo.long_maxThread < 0){
|
supVo.long_maxThread = -1 ;
|
}
|
supVo.long_minThread = conf.getSetAttrPlusInt(doc, "config.support", "long_minThread", null, -1, 100, null) ;//池中最小线程数
|
if(supVo.long_minThread < 0){
|
supVo.long_minThread = -1 ;
|
}
|
supVo.long_freeTimeout = conf.getSetAttrPlusInt(doc, "config.support", "long_freeTimeout", null, 1, 90, null) * 1000 ;//线程数空闲时长，若池中线程数量大于minThread，且有的线程空闲时长超过freeTimeout，则清除该线程
|
supVo.long_busyTimeout = conf.getSetAttrInt(doc, "config.support", "long_busyTimeout", null, -1, 10, null) ;//线程不间断工作时长（单位为秒)超时限，若为-1，不受限制
|
if(supVo.long_busyTimeout < 0){
|
supVo.long_busyTimeout = -1 ;
|
}
|
if(supVo.long_maxThread == 0 || supVo.long_minThread == 0){
|
supVo.enableLongThreadPool = false ;
|
}else{
|
supVo.enableLongThreadPool = true ;
|
}
|
|
supVo.showStartInfo = showStartInfo ;
|
|
AdapterImp_SupportUnit supAdap = new AdapterImp_SupportUnit() ;
|
supAdap.setConfig(supVo);
|
SupportUnit supUnit = SupportUnit.getInstance() ;
|
supUnit.setAdapter(supAdap);
|
supUnit.start(obj -> {
|
});
|
units.add(supUnit) ;
|
|
|
/////////////////////////
|
//资源模块
|
ResourceUnitConfigVo resVo = new ResourceUnitConfigVo() ;
|
//RTU日志文件存储目录(相对目录)
|
resVo.rtuLogDir = conf.getSetAttrTxt(doc, "config.resource", "rtuLogDir", null, false, null) ;
|
//RTU日志文件最大字节数(KB)
|
resVo.rtuLogFileMaxSize = conf.getSetAttrPlusInt(doc, "config.resource", "rtuLogFileMaxSize", null, 10, 2000000, null) ;
|
//RTU日志文件最大文件数
|
resVo.rtuLogFileMaxCount = conf.getSetAttrPlusInt(doc, "config.resource", "rtuLogFileMaxCount", null, 1, 10, null) ;
|
|
//Rtu状态监视间隔(分钟)
|
resVo.monitorInterval = conf.getSetAttrPlusInt(doc, "config.resource", "monitorInterval", null, 1, 5, null) ;
|
//Rtu状态存数据库间隔(分钟)
|
resVo.saveDbInterval = conf.getSetAttrPlusInt(doc, "config.resource", "saveDbInterval", null, 5, 20, null) ;
|
//Rtu无数据上传达到此值，则认为不在线了(分钟)
|
//注意：noUpDataLimitTime在元素config.protocol中
|
//resVo.noUpDataLimitTime = conf.getSetAttrPlusInt(doc, "config.protocol", "noUpDataLimitTime", null, 30, 60, null) ;
|
AdapterImp_ResourceUnit resAdap = new AdapterImp_ResourceUnit() ;
|
resAdap.setConfig(resVo);
|
ResourceUnit resUnit = ResourceUnit.getInstance() ;
|
resUnit.setAdapter(resAdap);
|
//当前支持spring + hibernate
|
resUnit.setSpringContext(SpringContextUtil.getApplicationContext());
|
|
resUnit.start(obj -> {
|
});
|
units.add(resUnit) ;
|
|
|
/////////////////
|
//RMI模块 暂时不应用
|
/*
|
RmiConfigVo rmiVo = new RmiConfigVo();
|
rmiVo.enable = conf.getSetAttrBoolean(doc, "config.rmi", "enable", null, null) ;
|
if(rmiVo.enable){
|
rmiVo.port = conf.getSetAttrPlusInt(doc, "config.rmi", "port", null, 100, 65535, null);
|
rmiVo.context = conf.getSetAttrTxt(doc, "config.rmi", "context", null, false, null);
|
rmiVo.showStartInfo = showStartInfo ;
|
AdapterImp_RmiUnit rmiAdap = new AdapterImp_RmiUnit();
|
rmiAdap.setConfig(rmiVo);
|
RmiUnit rmiUnit = RmiUnit.getInstance();
|
rmiUnit.setAdapter(rmiAdap);
|
rmiUnit.start(obj -> {
|
});
|
RmiSvUrl = "[ip]:" + rmiVo.port + "/" + rmiVo.context ;
|
units.add(rmiUnit) ;
|
}
|
*/
|
|
|
/////////////////
|
//消息中心模块
|
MsCenterConfigVo mscVo = new MsCenterConfigVo();
|
mscVo.enable = conf.getSetAttrBoolean(doc, "config.msCenter", "enable", null, null) ;
|
mscVo.notifyMsInterval = conf.getSetAttrPlusInt(doc, "config.msCenter", "notifyInterval", null, 1, 600, null) * 1000L ;
|
mscVo.showStartInfo = showStartInfo ;
|
AdapterImp_MsCenterUnit mscAdapt = new AdapterImp_MsCenterUnit();
|
mscAdapt.setConfig(mscVo);
|
MsCenterUnit mscUnit = MsCenterUnit.getInstance();
|
mscUnit.setAdapter(mscAdapt);
|
mscUnit.start(obj -> {
|
});
|
units.add(mscUnit) ;
|
|
|
/////////////////
|
//RTU远程升级模块
|
UpgradeUnitConfigVo ugVo = new UpgradeUnitConfigVo();
|
ugVo.enable = conf.getSetAttrBoolean(doc, "config.upgrade", "enable", null, null) ;
|
if(ugVo.enable){
|
ugVo.openNoUpgrade = conf.getSetAttrBoolean(doc, "config.upgrade", "openNoUpgrade", null, null) ;
|
ugVo.lastOpenMaxGoOn = conf.getSetAttrPlusInt(doc, "config.upgrade", "lastOpenMaxGoOn", null, 5, 360000, null);
|
ugVo.lastOpenMaxGoOn = ugVo.lastOpenMaxGoOn * 1000 ;//变成毫秒
|
ugVo.noOneRtuUpgradeMaxDuration = conf.getSetAttrPlusInt(doc, "config.upgrade", "noOneRtuUpgradeMaxDuration", null, 5, 360000, null);
|
ugVo.noOneRtuUpgradeMaxDuration = ugVo.noOneRtuUpgradeMaxDuration * 1000 ;//变成毫秒
|
ugVo.runningAndIdleDuration = conf.getSetAttrPlusInt(doc, "config.upgrade", "runningAndIdleDuration", null, 5, 360000, null);
|
ugVo.runningAndIdleDuration = ugVo.runningAndIdleDuration * 1000 ;//变成毫秒
|
ugVo.failTryTimes = conf.getSetAttrPlusInt(doc, "config.upgrade", "failTryTimes", null, 0, 100, null);
|
ugVo.ugMaxRtuAtOnce = conf.getSetAttrPlusInt(doc, "config.upgrade", "ugMaxRtuAtOnce", null, 0, 1000000, null);
|
ugVo.rtuOffLineWaitDuration = conf.getSetAttrPlusInt(doc, "config.upgrade", "rtuOffLineWaitDuration", null, 1, 3600000, null);
|
ugVo.rtuOffLineWaitDuration = ugVo.rtuOffLineWaitDuration * 1000;//变成毫秒
|
ugVo.notifyStateInterval = conf.getSetAttrPlusInt(doc, "config.upgrade", "notifyStateInterval", null, 1, 300, null);
|
ugVo.notifyStateInterval = ugVo.notifyStateInterval * 1000;//变成毫秒
|
ugVo.notifyTimesAfterOver = conf.getSetAttrPlusInt(doc, "config.upgrade", "notifyTimesAfterOver", null, 0, null, null);
|
ugVo.showStartInfo = showStartInfo ;
|
AdapterImp_UpgradeUnit ugAdap = new AdapterImp_UpgradeUnit();
|
ugAdap.setConfig(ugVo);
|
UpgradeUnit ugUnit = UpgradeUnit.getInstance();
|
ugUnit.setAdapter(ugAdap);
|
ugUnit.start(obj -> {
|
});
|
units.add(ugUnit) ;
|
}
|
|
|
/////////////////
|
//RTU上行数据处理模块（任务树）
|
RtuDataUnitConfigVo rducVo = new RtuDataUnitConfigVo();
|
rducVo.resourceLoader = this.resourceLoader ;
|
AdapterImp_RtuDataUnit rducAdap = new AdapterImp_RtuDataUnit();
|
rducAdap.setConfig(rducVo);
|
RtuDataUnit rducUnit = RtuDataUnit.getInstance();
|
rducUnit.setAdapter(rducAdap);
|
rducUnit.start(obj -> {
|
});
|
units.add(rducUnit) ;
|
|
|
|
// ///////////////
|
// 核心
|
CoreUnitConfigVo coreConfVo = new CoreUnitConfigVo();
|
coreConfVo.coreInterval = conf.getSetAttrPlusInt(doc, "config.core", "coreInterval", null, 1, 200, null).longValue() ;
|
coreConfVo.queueWarnSize = ServerProperties.cacheUpDownDataWarnCount ;
|
coreConfVo.queueMaxSize = ServerProperties.cacheUpDownDataMaxCount ;
|
coreConfVo.showStartInfo = showStartInfo ;
|
AdapterImp_CoreUnit coreAdap = new AdapterImp_CoreUnit();
|
coreAdap.setConfig(coreConfVo);
|
CoreUnit coreUnit = CoreUnit.getInstance();
|
coreUnit.setAdapter(coreAdap);
|
CoreUnit.addConstantTask(new RtuDownConstantTask());
|
CoreUnit.addConstantTask(new FromRtuDataConstantTask());
|
CoreUnit.addConstantTask(new FromRtuComResultConstantTask());
|
Boolean enableMq = conf.getSetAttrBoolean(doc, "config.mqtt", "enable", null, null) ;
|
if(enableMq != null && enableMq.booleanValue()){
|
CoreUnit.addConstantTask(new MqttSubMessageConstantTask());
|
CoreUnit.addConstantTask(new MqttPubMessageConstantTask());
|
CoreUnit.addConstantTask(new MqttComResultConstantTask());
|
}
|
CoreUnit.addConstantTask(new SendMsConstantTask());
|
coreUnit.start(obj -> {
|
});
|
units.add(coreUnit) ;
|
|
|
// ///////////////
|
// TCP 模块
|
TcpConfigVo tcpVo = new TcpConfigVo();
|
tcpVo.enable = conf.getSetAttrBoolean(doc, "config.tcp", "enable", null, null) ;
|
if(tcpVo.enable){
|
tcpVo.port = conf.getSetAttrPlusInt(doc, "config.tcp", "port", null, 100, 65535, null);
|
tcpVo.processors = conf.getSetAttrPlusInt(doc, "config.tcp", "processors", null, 1, 100, null);
|
tcpVo.idle = conf.getSetAttrPlusInt(doc, "config.tcp", "idle", null, 1, 20, null);// 10分钟
|
tcpVo.showStartInfo = showStartInfo ;
|
AdapterImp_TcpUnit tcpAdap = new AdapterImp_TcpUnit();
|
tcpAdap.setConfig(tcpVo);
|
TcpUnit tcpUnit = TcpUnit.getInstance();
|
tcpUnit.setAdapter(tcpAdap);
|
tcpUnit.start(obj -> {
|
});
|
TcpSvUrl = "[ip]:" + tcpVo.port ;
|
units.add(tcpUnit) ;
|
}
|
|
/////////////////
|
//MQTT模块
|
MqttUnitConfigVo mqVo = new MqttUnitConfigVo();
|
mqVo.enable = conf.getSetAttrBoolean(doc, "config.mqtt", "enable", null, null) ;
|
ServerProperties.mqttUnitEnable = mqVo.enable ;
|
if(mqVo.enable){
|
mqVo.svIp = conf.getSetAttrTxt(doc, "config.mqtt", "svIp", null, false, null) ;
|
if(!IPUtils.ipValid(mqVo.svIp)){
|
throw new Exception("config.mqtt.svIp配置的IP不合法") ;
|
}
|
mqVo.svPort = conf.getSetAttrPlusInt(doc, "config.mqtt", "svPort", null, 5, 360000, null);
|
if(mqVo.svPort < 0 || mqVo.svPort > 65535){
|
throw new Exception("config.mqtt.svPort配置的端口不合法") ;
|
}
|
mqVo.svUserName = conf.getSetAttrTxt(doc, "config.mqtt", "svUserName", null, false, null) ;
|
if(mqVo.svUserName == null || mqVo.svUserName.trim().equals("")){
|
throw new Exception("config.mqtt.svUserName配置的用户名不合法") ;
|
}else{
|
mqVo.svUserName = mqVo.svUserName.trim() ;
|
}
|
mqVo.svUserPassword = conf.getSetAttrTxt(doc, "config.mqtt", "svUserPassword", null, false, null) ;
|
if(mqVo.svUserPassword == null || mqVo.svUserPassword.trim().equals("")){
|
throw new Exception("config.mqtt.svUserName配置的用户密码不合法") ;
|
}else{
|
mqVo.svUserPassword = mqVo.svUserPassword.trim() ;
|
}
|
mqVo.poolMaxSize = conf.getSetAttrPlusInt(doc, "config.mqtt", "poolMaxSize", null, 1, 1000, null);
|
|
mqVo.sendInterval = conf.getSetAttrPlusInt(doc, "config.mqtt", "sendInterval", null, 1, 3600, null) * 1000L ;
|
|
mqVo.reSendTimesByNoResult = conf.getSetAttrPlusInt(doc, "config.mqtt", "reSendTimesByNoResult", null, 0, 100, null);
|
|
mqVo.comCacheTimeout = conf.getSetAttrPlusInt(doc, "config.mqtt", "comCacheTimeout", null, 1, 3600, null) * 1000L ;
|
|
ServerProperties.mqttAcceptDataMinInterval = conf.getSetAttrPlusInt(doc, "config.mqtt", "acceptDataMinInterval", null, 1, 720, null) * 60 * 1000L ;
|
|
mqVo.useMemoryPersistence = conf.getSetAttrBoolean(doc, "config.mqtt", "useMemoryPersistence", null, null) ;
|
String proAndDevIds = conf.getSetAttrTxt(doc, "config.mqtt", "protocolAndDeviceIds", null, true, null) ;
|
if(proAndDevIds != null && !proAndDevIds.trim().equals("")){
|
proAndDevIds = proAndDevIds.trim() ;
|
proAndDevIds = proAndDevIds.replaceAll("，", ",");
|
proAndDevIds = proAndDevIds.replaceAll("；", ";");
|
proAndDevIds = proAndDevIds.replaceAll("\\\\", "/");
|
mqVo.protocolAndDeviceIds = proAndDevIds.split(",") ;
|
mqVo.deviceIds = new String[mqVo.protocolAndDeviceIds.length] ;
|
int index = 0 ;
|
for(String topicAndQosStr : mqVo.protocolAndDeviceIds){
|
String[] pd = topicAndQosStr.split("/") ;
|
mqVo.deviceIds[index] = pd[1].trim() ;
|
index++ ;
|
}
|
}
|
|
String subTopicAndQos = conf.getSetAttrTxt(doc, "config.mqtt", "subTopicAndQos", null, true, null) ;
|
if(subTopicAndQos != null && !subTopicAndQos.trim().equals("")){
|
subTopicAndQos = subTopicAndQos.trim() ;
|
subTopicAndQos = subTopicAndQos.replaceAll("，", ",");
|
subTopicAndQos = subTopicAndQos.replaceAll("；", ";");
|
String[] topicAndQosArr = subTopicAndQos.split(";") ;
|
mqVo.subTopics = new String[topicAndQosArr.length] ;
|
mqVo.subTopicsQos = new int[topicAndQosArr.length] ;
|
int index = 0 ;
|
for(String topicAndQosStr : topicAndQosArr){
|
String[] tq = topicAndQosStr.split(",") ;
|
mqVo.subTopics[index] = tq[0].trim() ;
|
mqVo.subTopicsQos[index] = Integer.parseInt(tq[1].trim()) ;
|
index++ ;
|
}
|
}
|
mqVo.pubTopicQos = conf.getSetAttrPlusInt(doc, "config.mqtt", "pubTopicQos", null, 0, 3, null);
|
if(mqVo.pubTopicQos < 0 || mqVo.pubTopicQos > 3){
|
throw new Exception("config.mqtt.pubTopicQos配置不合法") ;
|
}
|
|
Integer intNoSubThenOff = conf.getSetAttrPlusInt(doc, "config.mqtt", "noSubThenOff", null, 1, 1440, null);
|
mqVo.noSubThenOff = intNoSubThenOff * 60 * 1000L ;
|
|
mqVo.showStartInfo = showStartInfo ;
|
AdapterImp_MqttUnit mqAdapt = new AdapterImp_MqttUnit();
|
mqAdapt.setConfig(mqVo);
|
MqttUnit mqUnit = MqttUnit.getInstance();
|
mqUnit.setAdapter(mqAdapt);
|
mqUnit.start(obj -> {
|
});
|
units.add(mqUnit) ;
|
}
|
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
|
|
}
|
