Skip to content

O
operating-vehicle
Commit 09c081de by dufafei
Options

Merge remote-tracking branch 'origin/master' 

# Conflicts:
#	src/main/scala/com/hikcreate/data/client/IgniteClient.scala
#	src/main/scala/com/hikcreate/data/offline/FullSync.scala
parents d4d16490 09ab3bf6
Showing with 303 additions and 47 deletions
.gitignore
### IntelliJ IDEA ###
.idea/
lib/
target/
*.iml
*.ipr
......
pom.xml
......@@ -132,7 +132,7 @@
<build>
<plugins>
<plugin>
<!--<plugin>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
......@@ -146,12 +146,17 @@
<includeScope>runtime</includeScope>
<excludeTransitive>false</excludeTransitive>
</configuration>
</plugin>
</plugin>-->
<plugin>
<groupId>org.scala-tools</groupId>
<artifactId>maven-scala-plugin</artifactId>
<version>2.15.2</version>
<executions><execution><goals><goal>compile</goal></goals></execution>
<executions>
<execution>
<goals>
<goal>compile</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
......
src/main/resources/conf.properties
kafka.bootstrap.servers=39.100.49.76:9092
#kafka.zookerper.servers=172.26.111.183:2181,172.26.111.178:2181,172.26.111.186:2181/tbd_kafka
kafka.zookerper.servers=10.197.236.211:2181
application.kafka.topic=tbd-transport-data-gathering
basicsInfo.kafka.topic=transport_basedata_operation
hive.group.id=hive
ignite.group.id=ignite1
ignite.group.id=ignite
basics.group.id=basics
hive.driver=org.apache.hive.jdbc.HiveDriver
......
src/main/scala/com/hikcreate/data/client/IgniteClient.scala
package com.hikcreate.data.client
import java.util.concurrent.TimeUnit
import com.hikcreate.ignite.domain.PrimaryKey
import com.hikcreate.ignite.domain.alarm._
import com.hikcreate.ignite.domain.alarm.processor.{DailyAlarmDealUpdate, DailyAlarmUpdate}
......@@ -12,7 +11,6 @@ import org.apache.ignite.binary.BinaryObject
import org.apache.ignite.cache.CacheMode
import org.apache.ignite.configuration.CacheConfiguration
import org.apache.ignite.{Ignite, IgniteCache, Ignition}
import scala.collection.JavaConversions.mapAsJavaMap
import scala.collection.JavaConverters._
/**
......@@ -210,9 +208,7 @@ object IgniteClient {
)
def main(args: Array[String]): Unit = {
//ignite.cacheNames().asScala.foreach(x=>ignite.destroyCache(x))
ignite.cacheNames().asScala.foreach(x=>ignite.getOrCreateCache(x).clear())
ignite.close()
}
}
src/main/scala/com/hikcreate/data/common/Sparking.scala
package com.hikcreate.data.common
import org.apache.log4j.{Level, Logger}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
trait Sparking {
// 屏蔽不必要的日志 ,在终端上显示需要的日志
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.WARN)
Logger.getLogger("org.apache.kafka.clients.consumer").setLevel(Level.WARN)
val conf: SparkConf = new SparkConf()
.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.set("hive.exec.dynamici.partition","true")
.set("hive.exec.dynamic.partition.mode","nonstrict")
.setAppName("test")
.setMaster("local[*]")
//.setAppName("test")
//.setMaster("local[*]")
def getKafkaParams(servers:String,groupId: String):Map[String,Object] = {
Map[String,Object](
......
src/main/scala/com/hikcreate/data/offline/FullSync.scala
......@@ -27,8 +27,8 @@ object FullSync extends Sparking{
val social_credit_code = row.getString(row.fieldIndex("social_credit_code"))
val into_time = row.getDate(row.fieldIndex("into_time"))
val province = row.getString(row.fieldIndex("province"))
val city = row.getString(row.fieldIndex("city"))
val area = row.getString(row.fieldIndex("area"))
val city = row.getString(row.fieldIndex("city")).substring(2,4)
val area = row.getString(row.fieldIndex("area")).substring(4,6)
val into_status = row.getBoolean(row.fieldIndex("into_status"))
val status = row.getString(row.fieldIndex("status"))
val address = row.getString(row.fieldIndex("address"))
......@@ -61,23 +61,14 @@ object FullSync extends Sparking{
val status = row.getString(row.fieldIndex("status"))
val vehicle_brand = row.getString(row.fieldIndex("vehicle_brand"))
val enterpriseInfo = Tools.getEnterpriseInfo(enterprise_code)
/*val province = enterpriseInfo.map(x => x.getProvince).getOrElse("")
val city = enterpriseInfo.map(x => x.getCity).getOrElse("")*/
val code = enterpriseInfo.map(x => x.getArea).getOrElse("")
if(code.length == 6) {
val province = code.substring(0,2)
val city = code.substring(2,4)
val area = code.substring(4,6)
val province = enterpriseInfo.map(x => x.getProvince).getOrElse("")
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)
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)
}
}
//报警类型配置基础表
sparkSession.sqlContext.read.format("jdbc").options(Map(
......
src/main/scala/com/hikcreate/data/sync/SyncIgnite.scala
......@@ -13,7 +13,6 @@ import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.joda.time.DateTime
import org.joda.time.format.DateTimeFormat
import scala.collection.mutable.ArrayBuffer
object SyncIgnite extends Sparking with Logging{
......@@ -23,9 +22,9 @@ object SyncIgnite extends Sparking with Logging{
val kafkaParams = getKafkaParams(Const.bootstrap, Const.igniteGroupId)
val offsets = zkManager.getBeginOffset(Const.applicationTopic, Const.igniteGroupId)
val offsetRanges = new ArrayBuffer[OffsetRange]()
val ssc = new StreamingContext(conf, Seconds(1))
val inputStream = KafkaUtils.createDirectStream[String, String](
ssc,
val ssc = new StreamingContext(conf, Seconds(2))
//ssc.sparkContext.setLogLevel("WARN")
val inputStream = KafkaUtils.createDirectStream[String, String](ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](Const.applicationTopic,kafkaParams,offsets))
inputStream.transform { rdd =>
......@@ -45,7 +44,6 @@ object SyncIgnite extends Sparking with Logging{
def processRow(iterator:Iterator[String]): Unit = {
iterator.foreach{ x =>
try {
println(x)
val json = JSON.parseObject(x)
TableKey(Option(json.getString("msgId")), Option(json.getString("dataType"))) match {
//车辆定位消息
......@@ -69,8 +67,7 @@ object SyncIgnite extends Sparking with Logging{
IgniteClient.vehicleNumberCache.withKeepBinary().put(vehicleNumberKey,vehicleNumberValue)
//累计行驶 累计安全行驶里程 今日车辆在线情况
val mileageKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor))
//val mileageValue = new DailyTravel(vehicleProvince,vehicleCity,vehicleArea,useNature,date,lat,lon,time,0D,0L)
val mileageValue = new DailyTravel("33","01","02",useNature,date,lat,lon,time,
val mileageValue = new DailyTravel(vehicleProvince,vehicleCity,vehicleArea,useNature,date,lat,lon,time,
0D,0L,code._1,code._2,code._3)
if(!IgniteClient.dailyTravelCache.withKeepBinary().putIfAbsent(mileageKey,mileageValue)){
IgniteClient.updateMileageCache(mileageKey,lat,lon,time)
......@@ -140,7 +137,7 @@ object SyncIgnite extends Sparking with Logging{
val eventType = infoJson.getString("EVENT_TYPE")
val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
val alarmInfoOptional = Tools.getAlarmInfo(warnType,eventType)
if(alarmInfoOptional.isDefined){//是否关联到报警信息基础表
if(alarmInfoOptional.isDefined){ //是否关联到报警信息基础表
val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("没有关联到车辆性质")
val alarmKey = IgniteClient.getBinaryObject(new PrimaryKey(code._1,code._2,code._3,useNature))
//累计行驶报警数
......@@ -171,11 +168,13 @@ object SyncIgnite extends Sparking with Logging{
val timestamp = json.getLong("warnTime") * 1000
val time = new DateTime(timestamp)
val warnTime = time.toString("yyyy-MM-dd HH:mm:ss")
val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("无")
val key = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId))
val value = new DailyAlarmDeal(vehicleNo,vehicleColor,"",superVisionId,warnTime,false)
val value = new DailyAlarmDeal(vehicleNo,vehicleColor,useNature,superVisionId,warnTime,false)
if(!IgniteClient.dailyAlarmDealCache.withKeepBinary().putIfAbsent(key,value)){
IgniteClient.updateDailyAlarmDealCache(key,Map(
"useNature"->"",
"useNature"->useNature,
"superVisionId"->superVisionId,
"warnTime"->warnTime)
)
......@@ -184,7 +183,7 @@ object SyncIgnite extends Sparking with Logging{
case tableKey if tableKey == TableKey(Some("0x1400"),Some("0x1401")) =>
val vehicleNo = json.getString("vehicleNo")
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 value = new DailyAlarmDeal(true)
if(!IgniteClient.dailyAlarmDealCache.withKeepBinary().putIfAbsent(key,value)){
......@@ -194,6 +193,8 @@ object SyncIgnite extends Sparking with Logging{
}
}catch {
case e:Exception =>
println(x)
println(e.getMessage)
e.printStackTrace()
}
}
......
src/main/scala/com/hikcreate/data/sync/SyncIgnite1.scala 0 → 100644
package com.hikcreate.data.sync
import com.alibaba.fastjson.JSON
import com.hikcreate.data.client.IgniteClient
import com.hikcreate.data.common.{Logging, Sparking}
import com.hikcreate.data.constant.Const
import com.hikcreate.data.model.TableKey
import com.hikcreate.data.util.{Tools, ZkManager}
import com.hikcreate.ignite.domain.PrimaryKey
import com.hikcreate.ignite.domain.alarm.{AttachmentInfo, DailyAlarm, DailyAlarmDeal, DailyAlarmDetail}
import com.hikcreate.ignite.domain.vehicles.{AlarmNumber, DailyTravel, DriverNumber, VehicleNumber}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.joda.time.DateTime
import org.joda.time.format.DateTimeFormat
import scala.collection.mutable.ArrayBuffer
object SyncIgnite1 extends Sparking with Logging{
def main(args: Array[String]): Unit = {
val zkManager = ZkManager(Const.zkKafka)
val kafkaParams = getKafkaParams(Const.bootstrap, Const.igniteGroupId)
val offsets = zkManager.getBeginOffset(Const.applicationTopic, Const.igniteGroupId)
val offsetRanges = new ArrayBuffer[OffsetRange]()
val ssc = new StreamingContext(conf, Seconds(1))
val inputStream = KafkaUtils.createDirectStream[String, String](ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](Const.applicationTopic,kafkaParams,offsets))
inputStream.transform { rdd =>
offsetRanges.clear()
offsetRanges.append(rdd.asInstanceOf[HasOffsetRanges].offsetRanges: _*)
rdd
}.map(x => x.value()).foreachRDD{ rdd =>
if (!rdd.isEmpty()) {
rdd.map(JSON.parseObject)
.groupBy(json=>TableKey(Option(json.getString("msgId")),Option(json.getString("dataType"))))
.foreachPartition( x=>
x.foreach{ x =>
try{
x._1 match {
case TableKey(Some("0x1200"),Some("0x1202")) => //车辆定位消息
val value = x._2.toList.grouped(20)
value.foreach{ sub =>
val codes = Tools.getLocationCodes(sub.map(x=>(x.getDouble("lon")/1000000,x.getDouble("lat")/1000000)))
sub.zip(codes).foreach{ x=>
val json = x._1
val code = x._2
val vehicleNo = json.getString("vehicleNo")
val vehicleColor = json.getString("vehicleColor")
val lon = json.getDouble("lon")/1000000
val lat = json.getDouble("lat")/1000000
val dateTime = DateTime.parse(json.getString("dateTime"),DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss"))//定位时间
val time = dateTime.toString("yyyy-MM-dd HH:mm:ss")
val date = dateTime.toLocalDate.toString("yyyy-MM-dd")
val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
val vehicleProvince = vehicleInfoOptional.map(x=>x.getProvince).getOrElse("无")
val vehicleCity = vehicleInfoOptional.map(x=>x.getCity).getOrElse("无")
val vehicleArea = vehicleInfoOptional.map(x=>x.getArea).getOrElse("无")
val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("无")
//累计行驶车辆数
val vehicleNumberKey = IgniteClient.getBinaryObject(new PrimaryKey(code._1,code._2,code._3,vehicleNo,vehicleColor))
val vehicleNumberValue = new VehicleNumber(code._1,code._2,code._3,useNature)
IgniteClient.vehicleNumberCache.withKeepBinary().put(vehicleNumberKey,vehicleNumberValue)
//累计行驶 累计安全行驶里程 今日车辆在线情况
val mileageKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor))
val mileageValue = new DailyTravel(vehicleProvince,vehicleCity,vehicleArea,useNature,
date, lat,lon,time, 0D,0L,code._1,code._2,code._3)
if(!IgniteClient.dailyTravelCache.withKeepBinary().putIfAbsent(mileageKey,mileageValue)){
IgniteClient.updateMileageCache(mileageKey,lat,lon,time)
}
}
}
case TableKey(Some("0x1200"),Some("0x1203")) => //车辆定位消息补报
val flat = x._2.flatMap(x=>Tools.addLocation(x))
val value = flat.toList.grouped(20)
value.foreach{ sub =>
val codes = Tools.getLocationCodes(sub.map(x=>(x.getDouble("lon")/1000000,x.getDouble("lat")/1000000)))
sub.zip(codes).foreach{ x=>
val json = x._1
val code = x._2
val vehicleNo = json.getString("vehicleNo")
val vehicleColor = json.getString("vehicleColor")
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 vehicleNumberValue = new VehicleNumber(code._1,code._2,code._3,useNature)
IgniteClient.vehicleNumberCache.withKeepBinary().put(vehicleNumberKey,vehicleNumberValue)
}
}
case tableKey if tableKey == TableKey(Some("0x1D00"),Some("0x1d02")) => //定时上传驾驶员身份识别信息
x._2.foreach{ x=>
val vehicleNo = x.getString("vehicleNo")
val vehicleColor = x.getString("vehicleColor")
val driverId = x.getString("driverId")
val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
val vehicleProvince = vehicleInfoOptional.map(x=>x.getProvince).getOrElse("无")
val vehicleCity = vehicleInfoOptional.map(x=>x.getCity).getOrElse("无")
val vehicleArea = vehicleInfoOptional.map(x=>x.getArea).getOrElse("无")
val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("无")
val key = IgniteClient.getBinaryObject(new PrimaryKey(driverId))
val driverNumber = new DriverNumber(vehicleProvince,vehicleCity,vehicleArea,useNature)
IgniteClient.driverNumberCache.withKeepBinary().put(key,driverNumber)
}
case TableKey(Some("0x1C00"),Some("0x1c02")) => //智能视频报警附件上传结果上报
x._2.foreach{ x=>
x.remove("msgId")
x.remove("dataType")
val timestamp = x.getLong("warnTime") * 1000
val dateTime = new DateTime(timestamp)
val warnTime = dateTime.toString("yyyy-MM-dd HH:mm:ss")
x.put("warnTime",warnTime)
val key = IgniteClient.getBinaryObject(
new PrimaryKey(x.getString("vehicleNo"),
x.getString("vehicleColor"),
x.getString("deviceId"),
x.getString("warnTime"),
x.getString("fileIndex"))
)
val value = JSON.parseObject(x.toJSONString,classOf[AttachmentInfo])
IgniteClient.attachmentCache.withKeepBinary().put(key,value)
}
case TableKey(Some("0x1400"), Some("0x1402")) => //上报报警信息消息
val value = x._2.filter(x=>Const.warnTypes.contains(x.getString("warnType"))).toList.grouped(20)
value.foreach{ sub=>
val codes = Tools.getLocationCodes(sub
.map{ x=>
val infoStr = x.getString("infoContent")
val infoJson = Tools.getInfoContentJsonobj(infoStr)
(infoJson.getDouble("LONGITUDE")/1000000,infoJson.getDouble("LATITUDE")/1000000)
})
sub.zip(codes).foreach{ x=>
val json = x._1
val code = x._2
val vehicleNo = json.getString("vehicleNo")
val vehicleColor = json.getString("vehicleColor")
val warnType = json.getString("warnType")
val timestamp = json.getLong("warnTime") * 1000
val time = new DateTime(timestamp)
val warnTime = time.toString("yyyy-MM-dd HH:mm:ss")
val date = time.toString("yyyy-MM-dd")
val infoStr = json.getString("infoContent")
val infoJson = Tools.getInfoContentJsonobj(infoStr)
val eventType = infoJson.getString("EVENT_TYPE")
val deviceId = infoJson.getString("DEVICE_ID")
val lon = infoJson.getDouble("LONGITUDE")/1000000
val lat = infoJson.getDouble("LATITUDE")/1000000
val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
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 alarmContent = alarmInfoOptional.get.getWarningTypeName
val alarmDetailKey = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,warnTime))
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)
}
}
}
//报警督办请求信息
case TableKey(Some("0x9400"),Some("0x9401")) =>
x._2.foreach{ x=>
val vehicleNo = x.getString("vehicleNo")
val vehicleColor = x.getString("vehicleColor")
val superVisionId = x.getString("superVisionId")
val timestamp = x.getLong("warnTime") * 1000
val warnTime = new DateTime(timestamp).toString("yyyy-MM-dd HH:mm:ss")
val vehicleInfoOptional = Tools.getVehicleInfo(vehicleNo,vehicleColor)
val useNature = vehicleInfoOptional.map(x=>x.getUseNature).getOrElse("无")
val key = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId))
val value = new DailyAlarmDeal(vehicleNo,vehicleColor,useNature,superVisionId,warnTime,false)
if(!IgniteClient.dailyAlarmDealCache.withKeepBinary().putIfAbsent(key,value)){
IgniteClient.updateDailyAlarmDealCache(key,Map(
"useNature"->useNature,
"superVisionId"->superVisionId,
"warnTime"->warnTime)
)
}
}
//报警督办应答消息
case tableKey if tableKey == TableKey(Some("0x1400"),Some("0x1401")) =>
x._2.foreach{ x=>
val vehicleNo = x.getString("vehicleNo")
val vehicleColor = x.getString("vehicleColor")
val superVisionId = x.getString("supervisionId")
val key = IgniteClient.getBinaryObject(new PrimaryKey(vehicleNo,vehicleColor,superVisionId))
val value = new DailyAlarmDeal(true)
if(!IgniteClient.dailyAlarmDealCache.withKeepBinary().putIfAbsent(key,value)){
IgniteClient.updateDailyAlarmDealCache(key,Map("isDeal"-> (true:java.lang.Boolean)))
}
}
case _ =>
}
}catch {
case e:Exception => e.printStackTrace()
}
}
)
zkManager.saveEndOffset(offsetRanges,Const.igniteGroupId)
}
}
ssc.start()
ssc.awaitTermination()
}
}
src/main/scala/com/hikcreate/data/util/Tools.scala
......@@ -8,8 +8,11 @@ import com.hikcreate.data.common.Logging
import com.hikcreate.data.constant.Const
import com.hikcreate.ignite.domain.basic.{AlarmTypeInfo, EnterpriseInfo, VehicleInfo}
import org.apache.ignite.cache.query.SqlQuery
import org.joda.time.{DateTime, Duration}
import scalaj.http.Http
import scala.collection.mutable.ArrayBuffer
object Tools extends Logging{
def addLocation(json:JSONObject): Array[JSONObject] = {
......@@ -25,6 +28,7 @@ object Tools extends Logging{
}
def getAddressAndLocationCode(lon:Double,lat:Double):(String,String) = {
try{
val json = new JSONObject()
val arr = new JSONArray()
val lonAndLat = new JSONObject()
......@@ -32,20 +36,21 @@ object Tools extends Logging{
lonAndLat.put("latitude",lat)
arr.add(lonAndLat)
json.put("locations",arr)
val response = Http(Const.areaCodeAndAddressUrl)
.postData(json.toJSONString)
val starttime = DateTime.now()
val response = Http(Const.areaCodeAndAddressUrl).postData(json.toJSONString)
.header("content-type","application/json")
//.charset("ISO-8859-1")
.timeout(connTimeoutMs = 8000, readTimeoutMs = 8000)
//.timeout(connTimeoutMs = 1000,readTimeoutMs = 1000)
.asString
if(response.code == 200){
val endtime = DateTime.now()
println("http请求时间："+new Duration(starttime,endtime).getMillis)
val body = JSON.parseObject(response.body)
val item = body.getJSONObject("result").getJSONArray("regeoItems").getJSONObject(0)
val address = item.getString("formattedAddress")
val locationCode = item.getJSONObject("addressComponent").getString("adcode")
(address,locationCode)
}else{
throw new RuntimeException("http请求城市区编码出错")
}catch{
case e:Exception =>
throw new RuntimeException(e)
}
}
......@@ -58,6 +63,39 @@ object Tools extends Logging{
(provinceCode,cityCode,areaCode)
}
def getAddressAndLocationCodes(buffer:List[(Double,Double)]): IndexedSeq[(String,String)] = {
val json = new JSONObject()
val arr = new JSONArray()
buffer.foreach{x =>
val lonAndLat = new JSONObject()
lonAndLat.put("longitude",x._1)
lonAndLat.put("latitude",x._2)
arr.add(lonAndLat)
}
json.put("locations",arr)
val startTime = DateTime.now()
val response = Http(Const.areaCodeAndAddressUrl).postData(json.toJSONString).header("content-type","application/json").asString
val endTime = DateTime.now()
println("http请求时间："+new Duration(startTime,endTime).getMillis)
val body = JSON.parseObject(response.body)
val items = body.getJSONObject("result").getJSONArray("regeoItems")
(0 until items.size()).map{ index =>
val item = items.getJSONObject(index)
val address = item.getString("formattedAddress")
val locationCode = item.getJSONObject("addressComponent").getString("adcode")
(address,locationCode)
}
}
def getLocationCodes(buffer:List[(Double,Double)]): IndexedSeq[(String,String,String)] = {
getAddressAndLocationCodes(buffer).map(x=>x._2).map{ locationCode =>
val provinceCode = locationCode.substring(0,2)
val cityCode = locationCode.substring(2,4)
val areaCode = locationCode.substring(4,6)
(provinceCode,cityCode,areaCode)
}
}
def getInfoContentJsonobj(infoStr:String):JSONObject = {
val jsonStr=("{\""+infoStr.replace(":=","\":\"").replace(";","\",\"")+"\"}").replace(",\"\"}","}")
val jSONObject = JSON.parseObject(jsonStr)
......
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