基于Java语言的公路运输协议数据处理设计源码

package com.gzyj.flink.gps.map.beidoualarm; import cn.hutool.core.collection.CollectionUtil; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer; import com.gzyj.flink.dto.alarm.AlarmALLObject; import com.gzyj.flink.dto.alarm.Kafka_VehicleNormalAlarm; import com.gzyj.flink.dto.alarm.MonitorClientAlarm; import com.gzyj.flink.dto.utils.ToolUtils; import com.gzyj.flink.gps.config.RedisKeyPrefixConfig; import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory; import com.gzyj.flink.gps.factory.RedisSessionFactory; import com.gzyj.flink.gps.mapper.mysql.MonYanzhongweiguiSetMapper; import com.gzyj.flink.gps.utils.TimeUtils; import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary; import com.gzyj.flink.pojo.mysql.set.MonYanzhongweiguiSet; import com.gzyj.flink.veh.VehicleIncrement; import org.apache.commons.lang3.StringUtils; import org.apache.flink.api.common.functions.RichFlatMapFunction; import org.apache.flink.configuration.Configuration; import org.apache.flink.util.Collector; import org.apache.ibatis.session.SqlSession; import org.apache.ibatis.session.SqlSessionFactory; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; import java.math.BigDecimal; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; /** * 北斗终端报警数据处理 */ public class BeidouZongduanDataProcess extends RichFlatMapFunction<Kafka_VehicleNormalAlarm, Object> { SqlSessionFactory factory; LettuceConnectionFactory redisFactory; @Override public void open(Configuration parameters) throws Exception { factory = MySQLMybatisSessionFactory.getSqlSessionFactory(); redisFactory = RedisSessionFactory.getRedisFactory(); } private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) { RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(redisSessionFactory); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class)); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class)); redisTemplate.afterPropertiesSet(); return redisTemplate; } public VehicleIncrement getVehicleInfo(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) { Set<String> keys = clusterRedisTemplate.keys(RedisKeyPrefixConfig.RedisKey_VehicleInfo + "*:" + vehId + "*"); VehicleIncrement vehicleInfo = null; if (CollectionUtil.isNotEmpty(keys)) { Object obj = clusterRedisTemplate.opsForValue().get(new ArrayList<String>(keys).get(0)); if (obj == null) { return null; } else { vehicleInfo = JSONObject.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), VehicleIncrement.class); } } else { return null; } return vehicleInfo; } public String trimend(String inStr, String suffix) { while (inStr.endsWith(suffix)) { inStr = inStr.substring(0, inStr.length() - suffix.length()); } return inStr; } @Override public void flatMap(Kafka_VehicleNormalAlarm VehAlarm, Collector<Object> collector) throws Exception { SqlSession session = factory.openSession(true); List<Object> list=new ArrayList<>(); try { MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper = session.getMapper(MonYanzhongweiguiSetMapper.class); MonYanzhongweiguiSet AlarmDeploy = monYanzhongweiguiSetMapper.selectByVehId(VehAlarm.getVehicleId()); session.commit(); RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory); VehicleIncrement vehinfo = getVehicleInfo(VehAlarm.getVehicleId(), clusterRedisTemplate); Map<Object, Object> tmp = clusterRedisTemplate.opsForHash().entries(RedisKeyPrefixConfig.RedisKey_CmAlarm + VehAlarm.getVehicleId()); // 获取开始报警数据 ConcurrentHashMap<Object, Object> VehicleAlarm = new ConcurrentHashMap<>(tmp); String[] sNote = trimend(VehAlarm.Note, ",").split(","); if (VehAlarm.Note.length() > 0) { //开始报警 Iterator<Map.Entry<Object, Object>> mapiter = VehicleAlarm.entrySet().iterator(); while (mapiter.hasNext()) { Map.Entry<Object, Object> entry = mapiter.next(); String sAlarmNote = (String) entry.getKey(); if (sAlarmNote.equalsIgnoreCase("疲劳驾驶")) sAlarmNote = "疲劳驾驶报警"; if (!VehAlarm.Note.contains(sAlarmNote)) { String sLastalarm = (String) entry.getValue(); AlarmALLObject Lastalarm = JSON.parseObject(sLastalarm, AlarmALLObject.class); if (VehAlarm.Velocity > Lastalarm.MaxSpeed) Lastalarm.MaxSpeed = VehAlarm.Velocity; if (VehAlarm.Latitude > 0 && VehAlarm.Longitude > 0) { Lastalarm.Distance += ToolUtils.GetDistance(Lastalarm.Lon, Lastalarm.Lat, VehAlarm.Longitude, VehAlarm.Latitude); Lastalarm.Lat = VehAlarm.Latitude; Lastalarm.Lon = VehAlarm.Longitude; } // 结束报警 double KeepTime = TimeUtils.getMinusSecond(VehAlarm.Time, Lastalarm.BeginTime); int ChaoSuBi = 0; int status = 1; if (KeepTime < 0) KeepTime = 0; // 疲劳驾驶严重违规计算 if (sAlarmNote.equalsIgnoreCase("疲劳驾驶报警")) { int plsj = -1; int plsj1 = -1; int plsj2 = -1; if (AlarmDeploy != null) { plsj = AlarmDeploy.getPlcxsj() == null ? -1 : AlarmDeploy.getPlcxsj(); plsj1 = AlarmDeploy.getPlcxsj1() == null ? -1 : AlarmDeploy.getPlcxsj1(); plsj2 = AlarmDeploy.getPlcxsj2() == null ? -1 : AlarmDeploy.getPlcxsj2(); if (plsj1 >= 0) { plsj1 = plsj1 * 60; if (KeepTime >= plsj1) status = 1; } if (plsj2 >= 0) { plsj2 = plsj2 * 60; if (KeepTime >= plsj2) status = 2; } if (plsj >= 0) { plsj = plsj * 60; if (KeepTime >= plsj) status = 3; } } } // 超速报警严重违规计算 if (sAlarmNote.equalsIgnoreCase("超速报警")) { int cssj = -1; int cssj1 = -1; int cssj2 = -1; double csbfb = -1; double csbfb1 = -1; double csbfb2 = -1; if (AlarmDeploy != null) { // 计算严重违�