Skip to content

O
operating-vehicle
Commit b937c9b0 by dufafei
Options

1

parent bfbd3eb8
Showing with 269 additions and 326 deletions
pom.xml
...@@ -16,6 +16,9 @@ ...@@ -16,6 +16,9 @@
</repositories> </repositories>
<properties> <properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<scala.binary.version>2.11</scala.binary.version> <scala.binary.version>2.11</scala.binary.version>
<spark.version>2.4.0</spark.version> <spark.version>2.4.0</spark.version>
<ignite.version>2.6.0</ignite.version> <ignite.version>2.6.0</ignite.version>
......
src/main/java/Test.java deleted 100644 → 0
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
public class Test {
public static void main(String[] args) {
DateTime eventTime = DateTime.parse("2019-10-17 8:51:36",DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")); //事件日期
String time = eventTime.toString("yyyy-MM-dd HH:mm:ss");
System.out.println(time);
}
}
src/main/java/com/hikcreate/ignite/pojo/PrimaryKey.java → src/main/java/com/hikcreate/ignite/domain/PrimaryKey.java
package com.hikcreate.ignite.pojo; package com.hikcreate.ignite.domain;
import java.io.Serializable; import java.io.Serializable;
......
src/main/java/com/hikcreate/ignite/pojo/basic/AttachmentInfo.java → src/main/java/com/hikcreate/ignite/domain/alarm/AttachmentInfo.java
package com.hikcreate.ignite.pojo.basic; package com.hikcreate.ignite.domain.alarm;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
......
src/main/java/com/hikcreate/ignite/pojo/alarm/DailyAlarm.java → src/main/java/com/hikcreate/ignite/domain/alarm/DailyAlarm.java
package com.hikcreate.ignite.pojo.alarm; package com.hikcreate.ignite.domain.alarm;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
......
src/main/java/com/hikcreate/ignite/pojo/alarm/DailyAlarmProcess.java → src/main/java/com/hikcreate/ignite/domain/alarm/DailyAlarmDeal.java
package com.hikcreate.ignite.pojo.alarm; package com.hikcreate.ignite.domain.alarm;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
public class DailyAlarmProcess implements Serializable { public class DailyAlarmDeal implements Serializable {
private static final long serialVersionUID = 1L; private static final long serialVersionUID = 1L;
...@@ -26,14 +26,16 @@ public class DailyAlarmProcess implements Serializable { ...@@ -26,14 +26,16 @@ public class DailyAlarmProcess implements Serializable {
@QuerySqlField @QuerySqlField
private Boolean isDeal; //是否处理 private Boolean isDeal; //是否处理
public DailyAlarmProcess(String useNature, String supervisionId, String warnTime,Boolean isDeal) { public DailyAlarmDeal(String vehicleNo, String vehicleColor, String useNature, String supervisionId, String warnTime, Boolean isDeal) {
this.vehicleNo = vehicleNo;
this.vehicleColor = vehicleColor;
this.useNature = useNature; this.useNature = useNature;
this.supervisionId = supervisionId; this.supervisionId = supervisionId;
this.warnTime = warnTime; this.warnTime = warnTime;
this.isDeal = isDeal; this.isDeal = isDeal;
} }
public DailyAlarmProcess(Boolean isDeal) { public DailyAlarmDeal(Boolean isDeal) {
this.isDeal = isDeal; this.isDeal = isDeal;
} }
} }
src/main/java/com/hikcreate/ignite/pojo/alarm/DailyAlarmDetail.java → src/main/java/com/hikcreate/ignite/domain/alarm/DailyAlarmDetail.java
package com.hikcreate.ignite.pojo.alarm; package com.hikcreate.ignite.domain.alarm;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
......
src/main/java/com/hikcreate/ignite/pojo/basic/AlarmInfo.java → src/main/java/com/hikcreate/ignite/domain/basic/AlarmTypeInfo.java
package com.hikcreate.ignite.pojo.basic; package com.hikcreate.ignite.domain.basic;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
public class AlarmInfo implements Serializable { public class AlarmTypeInfo implements Serializable {
private static final long serialVersionUID = 1L; private static final long serialVersionUID = 1L;
...@@ -25,7 +25,7 @@ public class AlarmInfo implements Serializable { ...@@ -25,7 +25,7 @@ public class AlarmInfo implements Serializable {
@QuerySqlField @QuerySqlField
private String businessTime; //业务发生时间,格式yyyy-MM-dd HH:mm:ss private String businessTime; //业务发生时间,格式yyyy-MM-dd HH:mm:ss
public AlarmInfo(Long id, String warningBigType, String warningSmallType, String warningTypeCode, Long warningLevelCode, String warningTypeName, String status, String businessTime) { public AlarmTypeInfo(Long id, String warningBigType, String warningSmallType, String warningTypeCode, Long warningLevelCode, String warningTypeName, String status, String businessTime) {
this.id = id; this.id = id;
this.warningBigType = warningBigType; this.warningBigType = warningBigType;
this.warningSmallType = warningSmallType; this.warningSmallType = warningSmallType;
......
src/main/java/com/hikcreate/ignite/pojo/basic/CompanyInfo.java → src/main/java/com/hikcreate/ignite/domain/basic/EnterpriseInfo.java
package com.hikcreate.ignite.pojo.basic; package com.hikcreate.ignite.domain.basic;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
import java.sql.Date; import java.sql.Date;
public class CompanyInfo implements Serializable { public class EnterpriseInfo implements Serializable {
private static final long serialVersionUID = 1L; private static final long serialVersionUID = 1L;
...@@ -33,7 +33,7 @@ public class CompanyInfo implements Serializable { ...@@ -33,7 +33,7 @@ public class CompanyInfo implements Serializable {
@QuerySqlField @QuerySqlField
private String businessTime; //业务发生时间,格式yyyy-MM-dd HH:mm:ss private String businessTime; //业务发生时间,格式yyyy-MM-dd HH:mm:ss
public CompanyInfo(String enterpriseCode, String enterpriseName, String socialCreditCode, Date intoTime, String province, String city, String area, Boolean intoStatus, String status, String address, String businessTime) { public EnterpriseInfo(String enterpriseCode, String enterpriseName, String socialCreditCode, Date intoTime, String province, String city, String area, Boolean intoStatus, String status, String address, String businessTime) {
this.enterpriseCode = enterpriseCode; this.enterpriseCode = enterpriseCode;
this.enterpriseName = enterpriseName; this.enterpriseName = enterpriseName;
this.socialCreditCode = socialCreditCode; this.socialCreditCode = socialCreditCode;
...@@ -47,18 +47,15 @@ public class CompanyInfo implements Serializable { ...@@ -47,18 +47,15 @@ public class CompanyInfo implements Serializable {
this.businessTime = businessTime; this.businessTime = businessTime;
} }
public CompanyInfo(Long id, String enterpriseCode, String enterpriseName, String socialCreditCode, Date intoTime, String province, String city, String area, Boolean intoStatus, String status, String address, String businessTime) { public String getProvince() {
this.id = id; return province;
this.enterpriseCode = enterpriseCode; }
this.enterpriseName = enterpriseName;
this.socialCreditCode = socialCreditCode; public String getCity() {
this.intoTime = intoTime; return city;
this.province = province; }
this.city = city;
this.area = area; public String getArea() {
this.intoStatus = intoStatus; return area;
this.status = status;
this.address = address;
this.businessTime = businessTime;
} }
} }
src/main/java/com/hikcreate/ignite/pojo/basic/PlatformInfo.java → src/main/java/com/hikcreate/ignite/domain/basic/PlatformInfo.java
package com.hikcreate.ignite.pojo.basic; package com.hikcreate.ignite.domain.basic;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
......
src/main/java/com/hikcreate/ignite/pojo/basic/VehicleInfo.java → src/main/java/com/hikcreate/ignite/domain/basic/VehicleInfo.java
package com.hikcreate.ignite.pojo.basic; package com.hikcreate.ignite.domain.basic;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
...@@ -9,7 +9,7 @@ public class VehicleInfo implements Serializable{ ...@@ -9,7 +9,7 @@ public class VehicleInfo implements Serializable{
private static final long serialVersionUID = 1L; private static final long serialVersionUID = 1L;
@QuerySqlField @QuerySqlField
private String id; //主键字段 private Long id; //主键字段
@QuerySqlField @QuerySqlField
private String vehicleCode; //车辆编号 private String vehicleCode; //车辆编号
@QuerySqlField @QuerySqlField
...@@ -50,7 +50,7 @@ public class VehicleInfo implements Serializable{ ...@@ -50,7 +50,7 @@ public class VehicleInfo implements Serializable{
@QuerySqlField @QuerySqlField
private String area; //区 private String area; //区
public VehicleInfo(String id, String vehicleCode, String plateNum, String plateColor, Date intoTime, String businessScope, String businessScopeDetail, String indentifier, Boolean intoStatus, String status, String operatingCertificateNo, String classLine, String useNature, String enterpriseCode, String drivingPermitNo, String vehicleBrand, String businessTime, String province, String city) { public VehicleInfo(Long id, String vehicleCode, String plateNum, String plateColor, Date intoTime, String businessScope, String businessScopeDetail, String indentifier, Boolean intoStatus, String status, String operatingCertificateNo, String classLine, String useNature, String enterpriseCode, String drivingPermitNo, String vehicleBrand, String businessTime, String province, String city, String area) {
this.id = id; this.id = id;
this.vehicleCode = vehicleCode; this.vehicleCode = vehicleCode;
this.plateNum = plateNum; this.plateNum = plateNum;
...@@ -70,6 +70,7 @@ public class VehicleInfo implements Serializable{ ...@@ -70,6 +70,7 @@ public class VehicleInfo implements Serializable{
this.businessTime = businessTime; this.businessTime = businessTime;
this.province = province; this.province = province;
this.city = city; this.city = city;
this.area = area;
} }
public String getUseNature() { public String getUseNature() {
...@@ -80,11 +81,11 @@ public class VehicleInfo implements Serializable{ ...@@ -80,11 +81,11 @@ public class VehicleInfo implements Serializable{
this.useNature = useNature; this.useNature = useNature;
} }
public String getId() { public Long getId() {
return id; return id;
} }
public void setId(String id) { public void setId(Long id) {
this.id = id; this.id = id;
} }
......
src/main/java/com/hikcreate/ignite/pojo/vehicles/AlarmNumber.java → src/main/java/com/hikcreate/ignite/domain/vehicles/AlarmNumber.java
package com.hikcreate.ignite.pojo.vehicles; package com.hikcreate.ignite.domain.vehicles;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
......
src/main/java/com/hikcreate/ignite/pojo/vehicles/DailyMileage.java → src/main/java/com/hikcreate/ignite/domain/vehicles/DailyMileage.java
package com.hikcreate.ignite.pojo.vehicles; package com.hikcreate.ignite.domain.vehicles;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable; import java.io.Serializable;
......
src/main/java/com/hikcreate/ignite/pojo/vehicles/DriverNumber.java → src/main/java/com/hikcreate/ignite/domain/vehicles/DriverNumber.java
package com.hikcreate.ignite.pojo.vehicles; package com.hikcreate.ignite.domain.vehicles;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
......
src/main/java/com/hikcreate/ignite/pojo/vehicles/VehicleNumber.java → src/main/java/com/hikcreate/ignite/domain/vehicles/VehicleNumber.java
package com.hikcreate.ignite.pojo.vehicles; package com.hikcreate.ignite.domain.vehicles;
import org.apache.ignite.cache.query.annotations.QuerySqlField; import org.apache.ignite.cache.query.annotations.QuerySqlField;
......
src/main/java/com/hikcreate/ignite/pojo/alarm/DailyAlarmDeal.java deleted 100644 → 0
package com.hikcreate.ignite.pojo.alarm;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
import java.io.Serializable;
public class DailyAlarmDeal implements Serializable {
private static final long serialVersionUID = 1L;
@QuerySqlField
private String useNature; //使用性质 公交
@QuerySqlField
private String supervisionId; //报警督办 ID
@QuerySqlField
private String warnTime;//报警时间
@QuerySqlField
private Boolean isDeal = false; //是否处理
public DailyAlarmDeal(String useNature, String supervisionId, String warnTime, Boolean isDeal) {
this.useNature = useNature;
this.supervisionId = supervisionId;
this.warnTime = warnTime;
this.isDeal = isDeal;
}
}
src/main/java/com/hikcreate/ignite/processor/DailyAlarmProcessUpdate.java → src/main/java/com/hikcreate/ignite/processor/DailyAlarmDealUpdate.java
...@@ -7,11 +7,11 @@ import javax.cache.processor.EntryProcessorException; ...@@ -7,11 +7,11 @@ import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.MutableEntry; import javax.cache.processor.MutableEntry;
import java.util.Map; import java.util.Map;
public class DailyAlarmProcessUpdate implements CacheEntryProcessor<BinaryObject,BinaryObject,Void> { public class DailyAlarmDealUpdate implements CacheEntryProcessor<BinaryObject,BinaryObject,Void> {
private Map<String,Object> parameter; private Map<String,Object> parameter;
public DailyAlarmProcessUpdate(Map<String,Object> parameter){ public DailyAlarmDealUpdate(Map<String,Object> parameter){
this.parameter = parameter; this.parameter = parameter;
} }
......
src/main/resources/ignite.xml
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd"> http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean class="org.apache.ignite.configuration.IgniteConfiguration"> <bean class="org.apache.ignite.configuration.IgniteConfiguration">
<property name="igniteInstanceName" value="dff"/>
<property name="clientMode" value="true"/> <property name="clientMode" value="true"/>
<property name="consistentId" value="node"/> <property name="consistentId" value="node"/>
<property name="peerClassLoadingEnabled" value="true"/> <property name="peerClassLoadingEnabled" value="true"/>
......
src/main/scala/com/hikcreate/data/client/IgniteClient.scala
package com.hikcreate.data.client package com.hikcreate.data.client
import java.util
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
import com.hikcreate.ignite.domain.PrimaryKey
import com.alibaba.fastjson.JSON import com.hikcreate.ignite.domain.alarm.{AttachmentInfo, DailyAlarm, DailyAlarmDeal, DailyAlarmDetail}
import com.hikcreate.ignite.pojo.PrimaryKey import com.hikcreate.ignite.domain.basic._
import com.hikcreate.ignite.pojo.alarm.{DailyAlarm, DailyAlarmDeal, DailyAlarmDetail, DailyAlarmProcess} import com.hikcreate.ignite.domain.vehicles.{AlarmNumber, DailyMileage, DriverNumber, VehicleNumber}
import com.hikcreate.ignite.pojo.basic._
import com.hikcreate.ignite.pojo.vehicles.{AlarmNumber, DailyMileage, DriverNumber, VehicleNumber}
import com.hikcreate.ignite.processor._ import com.hikcreate.ignite.processor._
import javax.cache.expiry.{CreatedExpiryPolicy, Duration} import javax.cache.expiry.{CreatedExpiryPolicy, Duration}
import org.apache.ignite.binary.{BinaryObject, BinaryTypeConfiguration} import org.apache.ignite.binary.BinaryObject
import org.apache.ignite.cache.{CacheMode, QueryEntity} import org.apache.ignite.cache.CacheMode
import org.apache.ignite.cache.query.SqlFieldsQuery import org.apache.ignite.configuration.CacheConfiguration
import org.apache.ignite.configuration.{CacheConfiguration, IgniteConfiguration}
import org.apache.ignite.{Ignite, IgniteCache, Ignition} import org.apache.ignite.{Ignite, IgniteCache, Ignition}
import org.joda.time.DateTime import scala.collection.JavaConversions.mapAsJavaMap
import scala.collection.JavaConverters._ import scala.collection.JavaConverters._
/** /**
* binary type无法自动更改 * binary type无法自动更改
...@@ -26,8 +21,6 @@ object IgniteClient { ...@@ -26,8 +21,6 @@ object IgniteClient {
lazy val ignite: Ignite = Ignition.start("ignite.xml") lazy val ignite: Ignite = Ignition.start("ignite.xml")
def active(): Unit = if(!ignite.cluster.active) ignite.cluster.active(true)
def getBinaryObject(key:Object):BinaryObject = { def getBinaryObject(key:Object):BinaryObject = {
val binary = ignite.binary().builder(key.getClass.getSimpleName) val binary = ignite.binary().builder(key.getClass.getSimpleName)
val fields = key.getClass.getDeclaredFields val fields = key.getClass.getDeclaredFields
...@@ -38,8 +31,9 @@ object IgniteClient { ...@@ -38,8 +31,9 @@ object IgniteClient {
binary.build() binary.build()
} }
/*********************************基础信息表****************************************************************/ ignite.destroyCache("AlarmInfo")
/*********************************基础信息表****************************************************************/
//平台基础信息表 //平台基础信息表
lazy val basicPlatformInfo: IgniteCache[Long, PlatformInfo] = ignite.getOrCreateCache( lazy val basicPlatformInfo: IgniteCache[Long, PlatformInfo] = ignite.getOrCreateCache(
new CacheConfiguration[Long,PlatformInfo]() new CacheConfiguration[Long,PlatformInfo]()
...@@ -49,16 +43,18 @@ object IgniteClient { ...@@ -49,16 +43,18 @@ object IgniteClient {
.setCacheMode(CacheMode.REPLICATED) .setCacheMode(CacheMode.REPLICATED)
.setIndexedTypes(classOf[Long],classOf[PlatformInfo]) .setIndexedTypes(classOf[Long],classOf[PlatformInfo])
) )
//企业基础信息表 -- 企业接入情况
lazy val basicCompanyInfo: IgniteCache[Long, CompanyInfo] = ignite.getOrCreateCache( //企业基础信息表--企业接入情况
new CacheConfiguration[Long,CompanyInfo]() lazy val basicEnterpriseInfo: IgniteCache[Long, EnterpriseInfo] = ignite.getOrCreateCache(
new CacheConfiguration[Long,EnterpriseInfo]()
.setSqlSchema("BasicInfo") .setSqlSchema("BasicInfo")
.setName("CompanyInfo") .setName("EnterpriseInfo")
.setDataRegionName("500MB_Region") .setDataRegionName("500MB_Region")
.setCacheMode(CacheMode.REPLICATED) .setCacheMode(CacheMode.REPLICATED)
.setIndexedTypes(classOf[Long],classOf[CompanyInfo]) .setIndexedTypes(classOf[Long],classOf[EnterpriseInfo])
) )
//车辆基础信息表 -- 接入车辆数
//车辆基础信息表--接入车辆数
lazy val basicVehicleInfo: IgniteCache[Long, VehicleInfo] = ignite.getOrCreateCache( lazy val basicVehicleInfo: IgniteCache[Long, VehicleInfo] = ignite.getOrCreateCache(
new CacheConfiguration[Long,VehicleInfo]() new CacheConfiguration[Long,VehicleInfo]()
.setSqlSchema("BasicInfo") .setSqlSchema("BasicInfo")
...@@ -67,15 +63,17 @@ object IgniteClient { ...@@ -67,15 +63,17 @@ object IgniteClient {
.setCacheMode(CacheMode.REPLICATED) .setCacheMode(CacheMode.REPLICATED)
.setIndexedTypes(classOf[Long],classOf[VehicleInfo]) .setIndexedTypes(classOf[Long],classOf[VehicleInfo])
) )
//报警类型配置基础表 //报警类型配置基础表
lazy val basicAlarmInfo: IgniteCache[Long, AlarmInfo] = ignite.getOrCreateCache( lazy val basicAlarmTypeInfo: IgniteCache[Long, AlarmTypeInfo] = ignite.getOrCreateCache(
new CacheConfiguration[Long,AlarmInfo]() new CacheConfiguration[Long,AlarmTypeInfo]()
.setSqlSchema("BasicInfo") .setSqlSchema("BasicInfo")
.setName("AlarmInfo") .setName("AlarmTypeInfo")
.setDataRegionName("500MB_Region") .setDataRegionName("500MB_Region")
.setCacheMode(CacheMode.REPLICATED) .setCacheMode(CacheMode.REPLICATED)
.setIndexedTypes(classOf[Long],classOf[AlarmInfo]) .setIndexedTypes(classOf[Long],classOf[AlarmTypeInfo])
) )
/*********************************营运车辆监测****************************************************************/ /*********************************营运车辆监测****************************************************************/
/** /**
...@@ -153,7 +151,7 @@ object IgniteClient { ...@@ -153,7 +151,7 @@ object IgniteClient {
/** /**
* 附件历史记录 * 附件历史记录
* 主键：车牌号 车牌颜色 报警时间 * 主键：车牌号 车牌颜色 设备ID 报警时间 序列号
*/ */
lazy val attachmentCache: IgniteCache[PrimaryKey, AttachmentInfo] = ignite.getOrCreateCache( lazy val attachmentCache: IgniteCache[PrimaryKey, AttachmentInfo] = ignite.getOrCreateCache(
new CacheConfiguration[PrimaryKey,AttachmentInfo]() new CacheConfiguration[PrimaryKey,AttachmentInfo]()
...@@ -162,7 +160,7 @@ object IgniteClient { ...@@ -162,7 +160,7 @@ object IgniteClient {
.setIndexedTypes(classOf[PrimaryKey],classOf[AttachmentInfo]) .setIndexedTypes(classOf[PrimaryKey],classOf[AttachmentInfo])
.setDataRegionName("500MB_Region") .setDataRegionName("500MB_Region")
.setCacheMode(CacheMode.REPLICATED) .setCacheMode(CacheMode.REPLICATED)
//.setExpiryPolicyFactory(CreatedExpiryPolicy.factoryOf(new Duration(TimeUnit.HOURS,24))) //.setExpiryPolicyFactory(CreatedExpiryPolicy.factoryOf(new Duration(TimeUnit.HOURS,24)))
) )
/** /**
...@@ -179,7 +177,7 @@ object IgniteClient { ...@@ -179,7 +177,7 @@ object IgniteClient {
.setCacheMode(CacheMode.REPLICATED) .setCacheMode(CacheMode.REPLICATED)
) )
def updateDailyAlarmCache(key:BinaryObject,time:String): Unit = { def updateDailyAlarmNumberCache(key:BinaryObject,time:String): Unit = {
dailyAlarmNumberCache dailyAlarmNumberCache
.withKeepBinary() .withKeepBinary()
.invoke(key,new DailyAlarmUpdate(time)) .invoke(key,new DailyAlarmUpdate(time))
...@@ -190,27 +188,26 @@ object IgniteClient { ...@@ -190,27 +188,26 @@ object IgniteClient {
* 主键：车牌号 车牌颜色 报警督办ID * 主键：车牌号 车牌颜色 报警督办ID
* 数据来源： 报警督办请求消息 报警督办应答消息 * 数据来源： 报警督办请求消息 报警督办应答消息
*/ */
lazy val dailyAlarmProcessCache: IgniteCache[PrimaryKey, DailyAlarmProcess] = ignite.getOrCreateCache( lazy val dailyAlarmDealCache: IgniteCache[PrimaryKey, DailyAlarmDeal] = ignite.getOrCreateCache(
new CacheConfiguration[PrimaryKey,DailyAlarmProcess]() new CacheConfiguration[PrimaryKey,DailyAlarmDeal]()
.setSqlSchema("Alarm") .setSqlSchema("Alarm")
.setName("DailyAlarmProcess") .setName("DailyAlarmProcess")
.setIndexedTypes(classOf[PrimaryKey],classOf[DailyAlarmProcess]) .setIndexedTypes(classOf[PrimaryKey],classOf[DailyAlarmDeal])
.setDataRegionName("500MB_Region") .setDataRegionName("500MB_Region")
.setCacheMode(CacheMode.REPLICATED) .setCacheMode(CacheMode.REPLICATED)
//.setExpiryPolicyFactory(CreatedExpiryPolicy.factoryOf(new Duration(TimeUnit.HOURS,24))) //.setExpiryPolicyFactory(CreatedExpiryPolicy.factoryOf(new Duration(TimeUnit.HOURS,24)))
) )
def updateDailyAlarmDealCache(key:BinaryObject,map:Map[String,AnyRef]): Unit = { def updateDailyAlarmDealCache(key:BinaryObject,map:Map[String,AnyRef]): Unit = {
dailyAlarmProcessCache dailyAlarmDealCache
.withKeepBinary() .withKeepBinary()
.invoke(key,new DailyAlarmProcessUpdate(map.asJava)) .invoke(key,new DailyAlarmDealUpdate(map))
} }
/** /**
* 今日报警详情 * 今日报警详情
* 主键：车牌号 车牌颜色 报警时间 信息 ID * 主键：车牌号 车牌颜色 报警时间
* 数据来源：上报报警信息消息，智能视频报警附件上传结果上报 * 数据来源：上报报警信息消息
* 根据 车牌号，车牌颜色，报警时间 关联 智能视频报警附件上传结果上报 获取视频地址信息
*/ */
lazy val dailyAlarmDetailCache: IgniteCache[PrimaryKey, DailyAlarmDetail] = ignite.getOrCreateCache( lazy val dailyAlarmDetailCache: IgniteCache[PrimaryKey, DailyAlarmDetail] = ignite.getOrCreateCache(
new CacheConfiguration[PrimaryKey,DailyAlarmDetail]() new CacheConfiguration[PrimaryKey,DailyAlarmDetail]()
...@@ -224,47 +221,22 @@ object IgniteClient { ...@@ -224,47 +221,22 @@ object IgniteClient {
def main(args: Array[String]): Unit = { def main(args: Array[String]): Unit = {
val a = ignite.binary.`type`("com.hikcreate.ignite.pojo.alarm.AlarmDetail") //ignite.cacheNames().asScala.foreach(println(_))
ignite.binary()
println(a.fieldTypeName("warnType")) //无法自动更新修改后的字段类型
dailyAlarmProcessCache.clear() /*basicPlatformInfo.destroy()
val json1 = JSON.parseObject("{\"appId\":\"10030\",\"businessTime\":\"2019-10-18 17:42:40\",\"dataType\":\"0x9401\",\"msgId\":\"0x9400\",\"superVisionEndTime\":\"1571391220\",\"superVisionId\":\"-566810\",\"superVisionLevel\":\"0\",\"superVisor\":\"supervisor\",\"superVisorEmail\":\"supervisor\",\"superVisorTel\":\"supervisor\",\"vehicleColor\":\"2\",\"vehicleNo\":\"浙D22717\",\"warnSrc\":\"1\",\"warnTime\":\"1571391217\",\"warnType\":\"100\"}") basicEnterpriseInfo.destroy()
val vehicleNo = json1.getString("vehicleNo") basicVehicleInfo.destroy()
val vehicleColor = json1.getString("vehicleColor") basicAlarmTypeInfo.destroy()*/
val superVisionId = json1.getString("superVisionId")
val timestamp = json1.getLong("warnTime") * 1000
val time = new DateTime(timestamp)
val warnTime = time.toString("yyyy-MM-dd HH:mm:ss")
val key = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId))
val value = new DailyAlarmProcess("",superVisionId,warnTime,false)
if(!IgniteClient.dailyAlarmProcessCache.withKeepBinary().putIfAbsent(key,value)){
updateDailyAlarmDealCache(key,Map(
"useNature"->"",
"superVisionId"->superVisionId,
"warnTime"->warnTime))
}
val key1 = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId))
val value1 = new DailyAlarmProcess(true)
if(!IgniteClient.dailyAlarmProcessCache.withKeepBinary().putIfAbsent(key1,value1)){
println("aaaaa")
updateDailyAlarmDealCache(key1,Map("isDeal"-> (true:java.lang.Boolean)))
}
//alarmDetailCache.destroy()
//basicCompanyInfo.clear()
//basicVehicleInfo.clear()
//mileageCache.clear() /*mileageCache.destroy()
//driverNumberCache.clear() driverNumberCache.destroy()
//vehicleNumberCache.clear() vehicleNumberCache.destroy()
//alarmNumberCache.clear() alarmNumberCache.destroy()*/
//dailyAlarmCache.clear() /*attachmentCache.destroy()
//alarmDetailCache.clear() dailyAlarmNumberCache.destroy()
//attachmentCache.clear() dailyAlarmDealCache.destroy()
//dailyAlarmDealCache.clear() dailyAlarmDetailCache.destroy()*/
ignite.close() ignite.close()
} }
......
src/main/scala/com/hikcreate/data/Etl.scala → src/main/scala/com/hikcreate/data/offline/FullSync.scala
package com.hikcreate.data package com.hikcreate.data.offline
import com.hikcreate.data.client.IgniteClient import com.hikcreate.data.client.IgniteClient
import com.hikcreate.data.common.Sparking import com.hikcreate.data.common.Sparking
import com.hikcreate.ignite.pojo.basic.{AlarmInfo, CompanyInfo, VehicleInfo} import com.hikcreate.data.util.Tools
import com.hikcreate.ignite.domain.basic.{AlarmTypeInfo, EnterpriseInfo, VehicleInfo}
import org.apache.log4j.{Level, Logger} import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession import org.apache.spark.sql.SparkSession
/**
*
* -DIGNITE_REST_START_ON_CLIENT=true
*/
object Etl extends Sparking{
def main(args: Array[String]): Unit = { object FullSync extends Sparking{
def main(args: Array[String]): Unit = {
Logger.getRootLogger.setLevel(Level.OFF) Logger.getRootLogger.setLevel(Level.OFF)
Logger.getLogger("org.apache.ignite").setLevel(Level.OFF) Logger.getLogger("org.apache.ignite").setLevel(Level.OFF)
val sparkSession = SparkSession.builder().config(conf).getOrCreate() val sparkSession = SparkSession.builder().config(conf).getOrCreate()
//基本企业信息表 //基本企业信息表
/*sparkSession.sqlContext.read.format("jdbc").options(Map( sparkSession.sqlContext.read.format("jdbc").options(Map(
"driver"->"com.mysql.jdbc.Driver", "driver"->"com.mysql.jdbc.Driver",
"url"->"jdbc:mysql://10.197.236.152:3306/db_tbd", "url"->"jdbc:mysql://10.197.236.152:3306/db_tbd",
"dbtable"->"base_into_enterprise_info", "dbtable"->"base_into_enterprise_info",
...@@ -36,12 +32,10 @@ object Etl extends Sparking{ ...@@ -36,12 +32,10 @@ object Etl extends Sparking{
val into_status = row.getBoolean(row.fieldIndex("into_status")) val into_status = row.getBoolean(row.fieldIndex("into_status"))
val status = row.getString(row.fieldIndex("status")) val status = row.getString(row.fieldIndex("status"))
val address = row.getString(row.fieldIndex("address")) val address = row.getString(row.fieldIndex("address"))
val value = new EnterpriseInfo(enterprise_code,enterprise_name,social_credit_code,
val value = new CompanyInfo(enterprise_code,enterprise_name,social_credit_code,into_time,province,city,area,into_status,status, into_time,province,city,area,into_status,status, address,"")
address,"") IgniteClient.basicEnterpriseInfo.withKeepBinary().put(id,value)
IgniteClient.basicCompanyInfo.withKeepBinary().put(id,value) }
}*/
//基本车辆信息表 //基本车辆信息表
sparkSession.sqlContext.read.format("jdbc").options(Map( sparkSession.sqlContext.read.format("jdbc").options(Map(
"driver"->"com.mysql.jdbc.Driver", "driver"->"com.mysql.jdbc.Driver",
...@@ -50,28 +44,33 @@ object Etl extends Sparking{ ...@@ -50,28 +44,33 @@ object Etl extends Sparking{
"user"->"root", "user"->"root",
"password"->"123456" "password"->"123456"
)).load().rdd.foreach{ row => )).load().rdd.foreach{ row =>
val id = row.getLong(row.fieldIndex("id")).toString val id = row.getLong(row.fieldIndex("id"))
val vehicle_code = row.getString(row.fieldIndex("vehicle_code")) val vehicle_code = row.getString(row.fieldIndex("vehicle_code"))
val plate_num = row.getString(row.fieldIndex("plate_num")) val plate_num = row.getString(row.fieldIndex("plate_num"))
val plate_color = row.getString(row.fieldIndex("plate_color")) val plate_color = row.getString(row.fieldIndex("plate_color"))
val into_time = row.getDate(row.fieldIndex("into_time")) val into_time = row.getDate(row.fieldIndex("into_time"))
val into_status = row.getBoolean(row.fieldIndex("into_status")) val into_status = row.getBoolean(row.fieldIndex("into_status"))
val business_scope = row.getString(row.fieldIndex("business_scope")) val business_scope = row.getString(row.fieldIndex("business_scope"))
val business_scope_detail = row.getString(row.fieldIndex("business_scope")) val business_scope_detail = row.getString(row.fieldIndex("business_scope_detail"))
val indentifier = row.getString(row.fieldIndex("business_scope")) val indentifier = row.getString(row.fieldIndex("indentifier"))
val operating_certificate_no = row.getString(row.fieldIndex("business_scope")) val operating_certificate_no = row.getString(row.fieldIndex("operating_certificate_no"))
val class_line = row.getString(row.fieldIndex("business_scope")) val class_line = row.getString(row.fieldIndex("class_line"))
val use_nature = row.getString(row.fieldIndex("business_scope")) val use_nature = row.getString(row.fieldIndex("use_nature"))
val enterprise_code = row.getString(row.fieldIndex("business_scope")) val enterprise_code = row.getString(row.fieldIndex("enterprise_code"))
val driving_permit_no = row.getString(row.fieldIndex("business_scope")) val driving_permit_no = row.getString(row.fieldIndex("driving_permit_no"))
val status = row.getString(row.fieldIndex("business_scope")) val status = row.getString(row.fieldIndex("status"))
val vehicle_brand = row.getString(row.fieldIndex("business_scope")) val vehicle_brand = row.getString(row.fieldIndex("vehicle_brand"))
val enterpriseInfo = Tools.getEnterpriseInfo(enterprise_code)
val value = new VehicleInfo(id,vehicle_code,plate_num,plate_color,into_time,business_scope,business_scope_detail,indentifier,into_status, val province = enterpriseInfo.map(x => x.getProvince).getOrElse("")
status,operating_certificate_no,class_line,use_nature,enterprise_code,driving_permit_no,vehicle_brand,"","","") val city = enterpriseInfo.map(x => x.getCity).getOrElse("")
val area = enterpriseInfo.map(x => x.getArea).getOrElse("")
val value = new VehicleInfo(
id,vehicle_code,plate_num,plate_color,into_time,business_scope,
business_scope_detail,indentifier,into_status,status,operating_certificate_no,
class_line,use_nature,enterprise_code,driving_permit_no,vehicle_brand,
"",province,city,area)
IgniteClient.basicVehicleInfo.withKeepBinary().put(id,value) IgniteClient.basicVehicleInfo.withKeepBinary().put(id,value)
} }
//报警类型配置基础表 //报警类型配置基础表
sparkSession.sqlContext.read.format("jdbc").options(Map( sparkSession.sqlContext.read.format("jdbc").options(Map(
"driver"->"com.mysql.jdbc.Driver", "driver"->"com.mysql.jdbc.Driver",
...@@ -87,9 +86,9 @@ object Etl extends Sparking{ ...@@ -87,9 +86,9 @@ object Etl extends Sparking{
val warning_big_type = row.getString(row.fieldIndex("warning_big_type")) val warning_big_type = row.getString(row.fieldIndex("warning_big_type"))
val warning_small_type = row.getString(row.fieldIndex("warning_small_type")) val warning_small_type = row.getString(row.fieldIndex("warning_small_type"))
val warning_type_code = row.getString(row.fieldIndex("warning_type_code")) val warning_type_code = row.getString(row.fieldIndex("warning_type_code"))
val value = new AlarmTypeInfo(id,warning_big_type,warning_small_type,warning_type_code,warning_level_id,warning_type_name,status,"")
val value = new AlarmInfo(id,warning_big_type,warning_small_type,warning_type_code,warning_level_id,warning_type_name,status,"") IgniteClient.basicAlarmTypeInfo.withKeepBinary().put(id,value)
IgniteClient.basicAlarmInfo.withKeepBinary().put(id,value)
} }
IgniteClient.ignite.close()
} }
} }
src/main/scala/com/hikcreate/data/SyncBasic.scala → src/main/scala/com/hikcreate/data/sync/SyncBasic.scala
package com.hikcreate.data package com.hikcreate.data.sync
import com.alibaba.fastjson.serializer.SerializerFeature
import com.alibaba.fastjson.JSON import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializerFeature
import com.hikcreate.data.client.IgniteClient import com.hikcreate.data.client.IgniteClient
import com.hikcreate.data.common.{Logging, Sparking} import com.hikcreate.data.common.{Logging, Sparking}
import com.hikcreate.data.constant.Const import com.hikcreate.data.constant.Const
import com.hikcreate.data.util.ZkManager import com.hikcreate.data.util.{Tools, ZkManager}
import com.hikcreate.ignite.pojo.basic.{CompanyInfo, PlatformInfo, VehicleInfo} import com.hikcreate.ignite.domain.basic.{EnterpriseInfo, VehicleInfo, PlatformInfo}
import org.apache.ignite.cache.query.SqlQuery
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._ import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scala.collection.mutable.ArrayBuffer import scala.collection.mutable.ArrayBuffer
object SyncBasic extends Sparking with Logging{ object SyncBasic extends Sparking with Logging{
def main(args: Array[String]): Unit = { def main(args: Array[String]): Unit = {
IgniteClient.active()
val zkManager = ZkManager(Const.zkKafka) val zkManager = ZkManager(Const.zkKafka)
val kafkaParams = getKafkaParams(Const.bootstrap,Const.basicsGroupId) val kafkaParams = getKafkaParams(Const.bootstrap,Const.basicsGroupId)
val offsets = zkManager.getBeginOffset(Const.basicsInfoTopic,Const.basicsGroupId) val offsets = zkManager.getBeginOffset(Const.basicsInfoTopic,Const.basicsGroupId)
...@@ -60,17 +59,17 @@ object SyncBasic extends Sparking with Logging{ ...@@ -60,17 +59,17 @@ object SyncBasic extends Sparking with Logging{
if(tableName.equals("baseIntoEnterpriseInfo")){ //企业 if(tableName.equals("baseIntoEnterpriseInfo")){ //企业
operation match { operation match {
case "add" => case "add" =>
IgniteClient.basicCompanyInfo.withKeepBinary().put(json.getLong("id"),JSON.parseObject(str,classOf[CompanyInfo])) IgniteClient.basicEnterpriseInfo.withKeepBinary().put(json.getLong("id"),JSON.parseObject(str,classOf[EnterpriseInfo]))
case "update" => case "update" =>
IgniteClient.basicCompanyInfo.withKeepBinary().put(json.getLong("id"),JSON.parseObject(str,classOf[CompanyInfo])) IgniteClient.basicEnterpriseInfo.withKeepBinary().put(json.getLong("id"),JSON.parseObject(str,classOf[EnterpriseInfo]))
case "delete" => case "delete" =>
IgniteClient.basicCompanyInfo.withKeepBinary().remove(json.getLong("id")) IgniteClient.basicEnterpriseInfo.withKeepBinary().remove(json.getLong("id"))
} }
} }
if(tableName.equals("baseIntoVehicleInfo")){ //车辆 if(tableName.equals("baseIntoVehicleInfo")){ //车辆
operation match { operation match {
case "add" => case "add" =>
val companyInfo = getCompanyInfo(json.getString("enterpriseCode")) val companyInfo = Tools.getEnterpriseInfo(json.getString("enterpriseCode"))
if(companyInfo.isDefined){ if(companyInfo.isDefined){
//json.put("province",companyInfo.get.getProvince) //json.put("province",companyInfo.get.getProvince)
//json.put("city",companyInfo.get.getCity) //json.put("city",companyInfo.get.getCity)
...@@ -85,25 +84,13 @@ object SyncBasic extends Sparking with Logging{ ...@@ -85,25 +84,13 @@ object SyncBasic extends Sparking with Logging{
if(tableName.equals("baseWarningType")){ if(tableName.equals("baseWarningType")){
operation match { operation match {
case "add" => case "add" =>
IgniteClient.basicAlarmInfo.withKeepBinary().put(json.getLong("id"),JSON.parseObject(str,classOf[VehicleInfo])) IgniteClient.basicAlarmTypeInfo.withKeepBinary().put(json.getLong("id"),JSON.parseObject(str,classOf[VehicleInfo]))
case "update" => case "update" =>
IgniteClient.basicAlarmInfo.withKeepBinary().put(json.getLong("id"),JSON.parseObject(str,classOf[VehicleInfo])) IgniteClient.basicAlarmTypeInfo.withKeepBinary().put(json.getLong("id"),JSON.parseObject(str,classOf[VehicleInfo]))
case "delete" => case "delete" =>
IgniteClient.basicAlarmInfo.withKeepBinary().remove(json.getLong("id")) IgniteClient.basicAlarmTypeInfo.withKeepBinary().remove(json.getLong("id"))
} }
} }
} }
} }
//根据车辆信息表的单位编码关联企业基础信息表得到省 市等信息
def getCompanyInfo(enterpriseCode:String): Option[CompanyInfo] = {
val companyInfoSql = new SqlQuery[Long,CompanyInfo](classOf[CompanyInfo],"ENTERPRISECODE = ?")
val companyInfos = IgniteClient.basicCompanyInfo.query(companyInfoSql.setArgs(enterpriseCode)).getAll
if(companyInfos.size() == 1){
val companyInfo = companyInfos.get(0).getValue
Some(companyInfo)
}else{
None
}
}
} }
src/main/scala/com/hikcreate/data/SyncHive.scala → src/main/scala/com/hikcreate/data/sync/SyncHive.scala
package com.hikcreate.data package com.hikcreate.data.sync
import java.sql.Connection import java.sql.Connection
import java.util.Locale import java.util.Locale
import scala.collection.mutable.ArrayBuffer
import com.alibaba.fastjson.{JSON, JSONObject} import com.alibaba.fastjson.{JSON, JSONObject}
import com.hikcreate.data.client.DbClient import com.hikcreate.data.client.DbClient
import com.hikcreate.data.common.{Logging, Sparking} import com.hikcreate.data.common.{Logging, Sparking}
...@@ -14,6 +14,8 @@ import org.apache.spark.streaming.{Seconds, StreamingContext} ...@@ -14,6 +14,8 @@ import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.joda.time.DateTime import org.joda.time.DateTime
import org.joda.time.format.DateTimeFormat import org.joda.time.format.DateTimeFormat
import scala.collection.mutable.ArrayBuffer
object SyncHive extends Sparking with Logging { object SyncHive extends Sparking with Logging {
def main(args: Array[String]): Unit = { def main(args: Array[String]): Unit = {
......
src/main/scala/com/hikcreate/data/SyncIgnite.scala → src/main/scala/com/hikcreate/data/sync/SyncIgnite.scala
package com.hikcreate.data package com.hikcreate.data.sync
import com.hikcreate.data.common.{Logging, Sparking}
import com.hikcreate.data.constant.Const
import com.hikcreate.data.util.{Tools, ZkManager}
import com.alibaba.fastjson.JSON import com.alibaba.fastjson.JSON
import com.hikcreate.data.client.IgniteClient import com.hikcreate.data.client.IgniteClient
import com.hikcreate.data.client.IgniteClient.updateDailyAlarmDealCache import com.hikcreate.data.common.{Logging, Sparking}
import com.hikcreate.data.constant.Const
import com.hikcreate.data.model.TableKey import com.hikcreate.data.model.TableKey
import com.hikcreate.ignite.pojo.PrimaryKey import com.hikcreate.data.util.{Tools, ZkManager}
import com.hikcreate.ignite.pojo.alarm.{DailyAlarm, DailyAlarmDetail, DailyAlarmProcess} import com.hikcreate.ignite.domain.PrimaryKey
import com.hikcreate.ignite.pojo.basic.{AlarmInfo, AttachmentInfo, VehicleInfo} import com.hikcreate.ignite.domain.alarm.{AttachmentInfo, DailyAlarm, DailyAlarmDeal, DailyAlarmDetail}
import com.hikcreate.ignite.pojo.vehicles.{AlarmNumber, DailyMileage, DriverNumber, VehicleNumber} import com.hikcreate.ignite.domain.vehicles.{AlarmNumber, DailyMileage, DriverNumber, VehicleNumber}
import org.apache.ignite.cache.query.SqlQuery
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._ import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.joda.time.DateTime import org.joda.time.DateTime
import org.joda.time.format.DateTimeFormat import org.joda.time.format.DateTimeFormat
import scala.collection.mutable.ArrayBuffer import scala.collection.mutable.ArrayBuffer
object SyncIgnite extends Sparking with Logging{ object SyncIgnite extends Sparking with Logging{
...@@ -60,53 +56,71 @@ object SyncIgnite extends Sparking with Logging{ ...@@ -60,53 +56,71 @@ object SyncIgnite extends Sparking with Logging{
val code = Tools.getLocationCode(lon,lat) val code = Tools.getLocationCode(lon,lat)
val time = json.getString("dateTime")//定位时间 val time = json.getString("dateTime")//定位时间
val date = DateTime.parse(time,DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")).toLocalDate.toString("yyyy-MM-dd") val date = DateTime.parse(time,DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")).toLocalDate.toString("yyyy-MM-dd")
val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("无")
//累计行驶车辆数
val vehicleNumberKey = IgniteClient.getBinaryObject(new PrimaryKey(code._1,code._2,code._3,vehicleNo,vehicleColor)) val vehicleNumberKey = IgniteClient.getBinaryObject(new PrimaryKey(code._1,code._2,code._3,vehicleNo,vehicleColor))
val vehicleNumberValue = new VehicleNumber(code._1,code._2,code._3,"") val vehicleNumberValue = new VehicleNumber(code._1,code._2,code._3,useNature)
IgniteClient.vehicleNumberCache.withKeepBinary().put(vehicleNumberKey,vehicleNumberValue) IgniteClient.vehicleNumberCache.withKeepBinary().put(vehicleNumberKey,vehicleNumberValue)
//当前在线车辆 今日车辆在线情况 累计行驶 累计安全行驶里程
val mileageKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor)) val mileageKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor))
if(IgniteClient.mileageCache.withKeepBinary().containsKey(mileageKey)){ val mileageValue = new DailyMileage("33","01","22",useNature,date,lat,lon,time,0D,0L)
if(!IgniteClient.mileageCache.withKeepBinary().putIfAbsent(mileageKey,mileageValue)){
IgniteClient.updateMileageCache(mileageKey,lat,lon,time) IgniteClient.updateMileageCache(mileageKey,lat,lon,time)
}else{
val value = new DailyMileage("33","01","22","",date,lat,lon,time,0D,0L)
IgniteClient.mileageCache.withKeepBinary().put(mileageKey,value)
} }
/*val vehicleInfoOptional = getVehicleInfo(vehicleNo,vehicleColor) /*
if(vehicleInfoOptional.isDefined){ if(vehicleInfoOptional.isDefined){
val vehicleInfo = vehicleInfoOptional.get val vehicleInfo = vehicleInfoOptional.get
val vehicleProvince = vehicleInfo.getProvince val vehicleProvince = vehicleInfo.getProvince
val vehicleCity = vehicleInfo.getCity val vehicleCity = vehicleInfo.getCity
val vehicleArea = vehicleInfo.getArea val vehicleArea = vehicleInfo.getArea
val useNature = vehicleInfo.getUseNature val useNature = vehicleInfo.getUseNature
val vehicleNumberKey = new PrimaryKey(code._1,code._2,code._3,vehicleNo,vehicleColor) val vehicleNumberKey = new PrimaryKey(code._1,code._2,code._3,vehicleNo,vehicleColor)
val vehicleNumberValue = new VehicleNumber(code._1,code._2,code._3,useNature) val vehicleNumberValue = new VehicleNumber(code._1,code._2,code._3,useNature)
IgniteClient.vehicleNumberCache.put(vehicleNumberKey,vehicleNumberValue)//累计行驶车辆数 IgniteClient.vehicleNumberCache.put(vehicleNumberKey,vehicleNumberValue)//累计行驶车辆数
val mileageKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor)) val mileageKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor))
if(!IgniteClient.mileageCache.withKeepBinary().containsKey(mileageKey)){ if(!IgniteClient.mileageCache.withKeepBinary().containsKey(mileageKey)){
val mileageValue = new DailyMileage(vehicleProvince,vehicleCity,vehicleArea,useNature,date,lat,lon,time,0D,0L) val mileageValue = new DailyMileage(vehicleProvince,vehicleCity,vehicleArea,useNature,date,lat,lon,time,0D,0L)
IgniteClient.mileageCache.withKeepBinary().put(mileageKey,mileageValue) IgniteClient.mileageCache.withKeepBinary().put(mileageKey,mileageValue)
}else{ }else{
IgniteClient.updateMileageCache(mileageKey,lat,lon,time)//今日车辆在线情况 累计行驶 累计安全行驶里程 当前在线车辆 IgniteClient.updateMileageCache(mileageKey,lat,lon,time)
} }
}*/ }*/
//车辆定位消息补报 //车辆定位消息补报
case tableKey if tableKey == TableKey(Some("0x1200"),Some("0x1203")) => case tableKey if tableKey == TableKey(Some("0x1200"),Some("0x1203")) =>
//定时上传驾驶员身份识别信息 //定时上传驾驶员身份识别信息
case tableKey if tableKey == TableKey(Some("0x1D00"),Some("0x1d02")) => case tableKey if tableKey == TableKey(Some("0x1D00"),Some("0x1d02")) =>
val vehicleNo = json.getString("vehicleNo") val vehicleNo = json.getString("vehicleNo")
val vehicleColor = json.getString("vehicleColor") val vehicleColor = json.getString("vehicleColor")
val driverId = json.getString("driverId") val driverId = json.getString("driverId")
val vehicleInfoOptional = getVehicleInfo(vehicleNo,vehicleColor) val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
if(vehicleInfoOptional.isDefined){ val vehicleProvince = vehicleInfoOptional.map(x=>x.getProvince).getOrElse("无")
val vehicleInfo = vehicleInfoOptional.get val vehicleCity = vehicleInfoOptional.map(x=>x.getCity).getOrElse("无")
val vehicleProvince = vehicleInfo.getProvince val vehicleArea = vehicleInfoOptional.map(x=>x.getArea).getOrElse("无")
val vehicleCity = vehicleInfo.getCity val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("无")
val vehicleArea = vehicleInfo.getArea val key = IgniteClient.getBinaryObject(new PrimaryKey(driverId))
val useNature = vehicleInfo.getUseNature val driverNumber = new DriverNumber(vehicleProvince,vehicleCity,vehicleArea,useNature)
val key = IgniteClient.getBinaryObject(new PrimaryKey(driverId)) IgniteClient.driverNumberCache.withKeepBinary().put(key,driverNumber)//接入驾驶人数
val driverNumber = new DriverNumber(vehicleProvince,vehicleCity,vehicleArea,useNature) //智能视频报警附件上传结果上报
IgniteClient.driverNumberCache.withKeepBinary().put(key,driverNumber)//接入驾驶人数 case tableKey if tableKey == TableKey(Some("0x1C00"),Some("0x1c02")) =>
} json.remove("msgId")
json.remove("dataType")
val timestamp = json.getLong("warnTime") * 1000
val dateTime = new DateTime(timestamp)
val warnTime = dateTime.toString("yyyy-MM-dd HH:mm:ss")
json.put("warnTime",warnTime)
val key = IgniteClient.getBinaryObject(
new PrimaryKey(
json.getString("vehicleNo"),
json.getString("vehicleColor"),
json.getString("deviceId"),
json.getString("warnTime"),
json.getString("fileIndex"))
)
val value = JSON.parseObject(json.toJSONString,classOf[AttachmentInfo])
IgniteClient.attachmentCache.withKeepBinary().put(key,value)
//上报报警信息消息 //上报报警信息消息
case tableKey if tableKey == TableKey(Some("0x1400"), Some("0x1402")) => case tableKey if tableKey == TableKey(Some("0x1400"), Some("0x1402")) =>
val vehicleNo = json.getString("vehicleNo") val vehicleNo = json.getString("vehicleNo")
...@@ -120,47 +134,35 @@ object SyncIgnite extends Sparking with Logging{ ...@@ -120,47 +134,35 @@ object SyncIgnite extends Sparking with Logging{
val lon = infoJson.getDouble("LONGITUDE")/1000000 val lon = infoJson.getDouble("LONGITUDE")/1000000
val lat = infoJson.getDouble("LATITUDE")/1000000 val lat = infoJson.getDouble("LATITUDE")/1000000
val code = Tools.getLocationCode(lon,lat) val code = Tools.getLocationCode(lon,lat)
val time = new DateTime(timestamp) //com.alibaba.fastjson.JSONException: can not cast to long, value 0x000000005DA83E33 val time = new DateTime(timestamp)
val warnTime = time.toString("yyyy-MM-dd HH:mm:ss") val warnTime = time.toString("yyyy-MM-dd HH:mm:ss")
val date = time.toString("yyyy-MM-dd") val date = time.toString("yyyy-MM-dd")
//val vehicleInfoOptional = getVehicleInfo(vehicleNo,vehicleColor)
//val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("没有关联到车辆性质")
val key = IgniteClient.getBinaryObject(new PrimaryKey(code._1,code._2,code._3,""))
//累计行驶报警数
val alarmNumber = new AlarmNumber(code._1,code._2,code._3,"",1L)
if(!IgniteClient.alarmNumberCache.withKeepBinary().putIfAbsent(key,alarmNumber)){
IgniteClient.updateAlarmNumberCache(key)
}
//今日报警数
val dailyAlarmNumber = new DailyAlarm(code._1,code._2,code._3,"",date,1L)
if(!IgniteClient.dailyAlarmNumberCache.withKeepBinary().putIfAbsent(key,dailyAlarmNumber)){
IgniteClient.updateDailyAlarmCache(key,warnTime)
}
//今日报警详情
val eventType = infoJson.getString("EVENT_TYPE") val eventType = infoJson.getString("EVENT_TYPE")
val alarmInfoOptional = getAlarmInfo(warnType,eventType) val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
if(alarmInfoOptional.isDefined){ val alarmInfoOptional = Tools.getAlarmInfo(warnType,eventType)
if(alarmInfoOptional.isDefined){//是否关联到报警信息基础表
val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("没有关联到车辆性质")
val alarmKey = IgniteClient.getBinaryObject(new PrimaryKey(code._1,code._2,code._3,useNature))
//累计行驶报警数
val alarmNumber = new AlarmNumber(code._1,code._2,code._3,useNature,1L)
if(!IgniteClient.alarmNumberCache.withKeepBinary().putIfAbsent(alarmKey,alarmNumber)){
IgniteClient.updateAlarmNumberCache(alarmKey)
}
//今日报警数
val dailyAlarmNumber = new DailyAlarm(code._1,code._2,code._3,useNature,date,1L)
if(!IgniteClient.dailyAlarmNumberCache.withKeepBinary().putIfAbsent(alarmKey,dailyAlarmNumber)){
IgniteClient.updateDailyAlarmNumberCache(alarmKey,warnTime)
}
//今日报警详情
val alarmType = alarmInfoOptional.get.getWarningLevelCode val alarmType = alarmInfoOptional.get.getWarningLevelCode
val alarmContent = alarmInfoOptional.get.getWarningTypeName val alarmContent = alarmInfoOptional.get.getWarningTypeName
val alarmDetailKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,warnTime)) val alarmDetailKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,warnTime))
val dailyAlarmDetail = new DailyAlarmDetail(code._1,code._2,code._3,vehicleNo,vehicleColor,"",deviceId,warnTime,lon,lat,alarmType,alarmContent) val dailyAlarmDetail = new DailyAlarmDetail(
code._1,code._2,code._3,vehicleNo,vehicleColor,useNature,
deviceId,warnTime,lon,lat,alarmType,alarmContent)
IgniteClient.dailyAlarmDetailCache.withKeepBinary().put(alarmDetailKey,dailyAlarmDetail) IgniteClient.dailyAlarmDetailCache.withKeepBinary().put(alarmDetailKey,dailyAlarmDetail)
} }
} }
//智能视频报警附件上传结果上报
case tableKey if tableKey == TableKey(Some("0x1C00"),Some("0x1c02")) =>
json.remove("msgId")
json.remove("dataType")
val timestamp = json.getLong("warnTime") * 1000
val dateTime = new DateTime(timestamp)
val warnTime = dateTime.toString("yyyy-MM-dd HH:mm:ss")
json.put("warnTime",warnTime)
val key = IgniteClient.getBinaryObject(
new PrimaryKey(json.getString("vehicleNo"),json.getString("vehicleColor"),
json.getString("deviceId"),json.getString("warnTime"),json.getString("fileIndex"))
)
val value = JSON.parseObject(json.toJSONString,classOf[AttachmentInfo])
IgniteClient.attachmentCache.withKeepBinary().put(key,value)
//报警督办请求信息 //报警督办请求信息
case tableKey if tableKey == TableKey(Some("0x9400"),Some("0x9401")) => case tableKey if tableKey == TableKey(Some("0x9400"),Some("0x9401")) =>
val vehicleNo = json.getString("vehicleNo") val vehicleNo = json.getString("vehicleNo")
...@@ -170,12 +172,13 @@ object SyncIgnite extends Sparking with Logging{ ...@@ -170,12 +172,13 @@ object SyncIgnite extends Sparking with Logging{
val time = new DateTime(timestamp) val time = new DateTime(timestamp)
val warnTime = time.toString("yyyy-MM-dd HH:mm:ss") val warnTime = time.toString("yyyy-MM-dd HH:mm:ss")
val key = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId)) val key = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId))
val value = new DailyAlarmProcess("",superVisionId,warnTime,false) val value = new DailyAlarmDeal(vehicleNo,vehicleColor,"",superVisionId,warnTime,false)
if(!IgniteClient.dailyAlarmProcessCache.withKeepBinary().putIfAbsent(key,value)){ if(!IgniteClient.dailyAlarmDealCache.withKeepBinary().putIfAbsent(key,value)){
updateDailyAlarmDealCache(key,Map( IgniteClient.updateDailyAlarmDealCache(key,Map(
"useNature"->"", "useNature"->"",
"superVisionId"->superVisionId, "superVisionId"->superVisionId,
"warnTime"->warnTime)) "warnTime"->warnTime)
)
} }
//报警督办应答消息 //报警督办应答消息
case tableKey if tableKey == TableKey(Some("0x1400"),Some("0x1401")) => case tableKey if tableKey == TableKey(Some("0x1400"),Some("0x1401")) =>
...@@ -183,9 +186,9 @@ object SyncIgnite extends Sparking with Logging{ ...@@ -183,9 +186,9 @@ object SyncIgnite extends Sparking with Logging{
val vehicleColor = json.getString("vehicleColor") val vehicleColor = json.getString("vehicleColor")
val superVisionId = json.getString("superVisionId") val superVisionId = json.getString("superVisionId")
val key = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId)) val key = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId))
val value = new DailyAlarmProcess(true) val value = new DailyAlarmDeal(true)
if(!IgniteClient.dailyAlarmProcessCache.withKeepBinary().putIfAbsent(key,value)){ if(!IgniteClient.dailyAlarmDealCache.withKeepBinary().putIfAbsent(key,value)){
updateDailyAlarmDealCache(key,Map("isDeal"-> (true:java.lang.Boolean))) IgniteClient.updateDailyAlarmDealCache(key,Map("isDeal"-> (true:java.lang.Boolean)))
} }
case _ => case _ =>
} }
...@@ -195,28 +198,4 @@ object SyncIgnite extends Sparking with Logging{ ...@@ -195,28 +198,4 @@ object SyncIgnite extends Sparking with Logging{
} }
} }
} }
//根据车牌号和车牌号码得到车辆信息
def getVehicleInfo(plateNum:String,plateColor:String): Option[VehicleInfo] = {
val vehicleInfoSql = new SqlQuery[Long,VehicleInfo](classOf[VehicleInfo],"PLATENUM = ? AND PLATECOLOR= ? ")
val vehicleInfos = IgniteClient.basicVehicleInfo.query(vehicleInfoSql.setArgs(plateNum,plateColor)).getAll
if(vehicleInfos.size() == 1 ){
val vehicleInfo = vehicleInfos.get(0).getValue
Some(vehicleInfo)
}else{
None
}
}
//根据报警类型和事件类型 得到 平台报警等级code 和 平台报警类型名称
def getAlarmInfo(warnType:String,eventType:String): Option[AlarmInfo] = {
val alarmInfoSql = new SqlQuery[Long,AlarmInfo](classOf[AlarmInfo],"warningBigType = ? AND warningSmallType = ? ")
val alarmInfos = IgniteClient.basicAlarmInfo.query(alarmInfoSql.setArgs(warnType,eventType)).getAll
if(alarmInfos.size() == 1 ){
val alarmInfo = alarmInfos.get(0).getValue
Some(alarmInfo)
}else{
None
}
}
} }
src/main/scala/com/hikcreate/data/util/Tools.scala
...@@ -3,8 +3,11 @@ package com.hikcreate.data.util ...@@ -3,8 +3,11 @@ package com.hikcreate.data.util
import scala.collection.mutable import scala.collection.mutable
import scala.collection.JavaConverters._ import scala.collection.JavaConverters._
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject} import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.hikcreate.data.client.IgniteClient
import com.hikcreate.data.common.Logging import com.hikcreate.data.common.Logging
import com.hikcreate.data.constant.Const import com.hikcreate.data.constant.Const
import com.hikcreate.ignite.domain.basic.{AlarmTypeInfo, EnterpriseInfo, VehicleInfo}
import org.apache.ignite.cache.query.SqlQuery
import scalaj.http.Http import scalaj.http.Http
object Tools extends Logging{ object Tools extends Logging{
...@@ -46,6 +49,7 @@ object Tools extends Logging{ ...@@ -46,6 +49,7 @@ object Tools extends Logging{
} }
} }
//根据经纬度获取地区编码
def getLocationCode(lon:Double,lat:Double): (String,String,String) = { def getLocationCode(lon:Double,lat:Double): (String,String,String) = {
val locationCode = getAddressAndLocationCode(lon,lat)._2 val locationCode = getAddressAndLocationCode(lon,lat)._2
val provinceCode = locationCode.substring(0,2) val provinceCode = locationCode.substring(0,2)
...@@ -59,4 +63,40 @@ object Tools extends Logging{ ...@@ -59,4 +63,40 @@ object Tools extends Logging{
val jSONObject = JSON.parseObject(jsonStr) val jSONObject = JSON.parseObject(jsonStr)
jSONObject jSONObject
} }
//根据企业编码得到企业信息
def getEnterpriseInfo(enterpriseCode:String): Option[EnterpriseInfo] = {
val vehicleInfoSql = new SqlQuery[Long,EnterpriseInfo](classOf[EnterpriseInfo],"enterpriseCode = ?")
val vehicleInfos = IgniteClient.basicEnterpriseInfo.query(vehicleInfoSql.setArgs(enterpriseCode)).getAll
if(vehicleInfos.size() == 1 ){
val vehicleInfo = vehicleInfos.get(0).getValue
Some(vehicleInfo)
}else{
None
}
}
//根据车牌号和车牌号码得到车辆信息
def getVehicleInfo(plateNum:String,plateColor:String): Option[VehicleInfo] = {
val vehicleInfoSql = new SqlQuery[Long,VehicleInfo](classOf[VehicleInfo],"PLATENUM = ? AND PLATECOLOR= ? ")
val vehicleInfos = IgniteClient.basicVehicleInfo.query(vehicleInfoSql.setArgs(plateNum,plateColor)).getAll
if(vehicleInfos.size() == 1 ){
val vehicleInfo = vehicleInfos.get(0).getValue
Some(vehicleInfo)
}else{
None
}
}
//根据报警类型和事件类型 得到 平台报警等级code 和 平台报警类型名称
def getAlarmInfo(warnType:String,eventType:String): Option[AlarmTypeInfo] = {
val alarmInfoSql = new SqlQuery[Long,AlarmTypeInfo](classOf[AlarmTypeInfo],"warningBigType = ? AND warningSmallType = ? ")
val alarmInfos = IgniteClient.basicAlarmTypeInfo.query(alarmInfoSql.setArgs(warnType,eventType)).getAll
if(alarmInfos.size() == 1 ){
val alarmInfo = alarmInfos.get(0).getValue
Some(alarmInfo)
}else{
None
}
}
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment