Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 0d8d24f4 by 黄准
Options

公共doris连接池

parent 8e5edfaf
Showing with 145 additions and 51 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/LaneNormProcessFunction.java
......@@ -2,10 +2,7 @@ package com.zhht.irn.functions;
import com.zhht.irn.entity.*;
import com.zhht.irn.enums.EventType;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.MySQLUtils;
import com.zhht.irn.utils.StringUtils;
import com.zhht.irn.utils.*;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
......@@ -53,9 +50,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
@Override
public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
processCycle.update(true);
Connection connection = getConnection();
//初始化字典数据
if (dict.get("phaseDirection") == null) {
if (dict.get("phaseDirection") == null || dict.get("laneList") == null || dict.get("dictResult") == null || dict.get("positionList") == null || dict.get("positionNameList") == null) {
Connection connection = getConnection();
//查询相位方向信息
dict.put("phaseDirection", getDictResult(connection, "SELECT\n" +
" a.ID id,\n" +
......@@ -63,11 +62,36 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
" b.Name traffic_flow_direction,\n" +
" c.PID phase_no,\n" +
" c.PName phase_name\n" +
"FROM tb_phase_movement a\n" +
"LEFT JOIN trafficflowdirection b\n" +
"FROM dim_phase_movement a\n" +
"LEFT JOIN dim_traffic_flow_direction b\n" +
" on a.TrafficFlowId = b.ID\n" +
"LEFT JOIN dim_dic_phase c\n" +
" on a.PhaseId = c.ID"));
//车道道路流向关系
dict.put("laneList", getDictResult(connection, "SELECT cross_id,lane_id,turn_direction as name,position,lane_length FROM dim_cnt_cross_lane_position\n" +
" where out_in_type='in'"));
//获取所有车道信息
dict.put("dictResult", getDictResult(connection, "select a.*,b.saturation_flow_rate from (\n" +
" SELECT cross_id,lane_id,position,lane_length,GROUP_CONCAT(turn_direction) as name FROM dim_cnt_cross_lane_position\n" +
" where out_in_type='in' group By lane_id,cross_id,position,lane_length,cross_id order by cross_id,lane_id)a left join app_atomic_lane_saturation_flow_rate b\n" +
" on a.lane_id=b.lane_id and a.cross_id=b.cross_id\t\t" +
" where 1=1\n" +
" and b.inout_name='in' order by b.lane_id asc"));
//查询路口所有的方向
dict.put("positionList", getDictResult(connection, "SELECT cross_id,position FROM dim_cnt_cross_lane_position\n" +
" where out_in_type='in' GROUP BY position,cross_id"));
//查询路口所有的流向
dict.put("positionNameList", getDictResult(connection, " SELECT position,turn_direction as name,cross_id FROM dim_cnt_cross_lane_position\n" +
" where out_in_type='in' GROUP BY position,turn_direction,cross_id"));
long currentProcessingTime = ctx.timerService().currentProcessingTime();
//12小时触发一次
ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+12*60*60*1000);
//ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+1*60*1000);
//关闭mysql连接
if (connection != null) {
connection.close();
}
}
Cycle f1 = cycle;
// 代表周期已到，需要进行相应的指标计算了
......@@ -76,23 +100,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
if (travelEvents == null) {
travelEvents = new ArrayList<>();
}
List<TravelEvent> car = getCar(travelEvents);
List<TravelEvent> car = getCar(travelEvents,cycle);
//根据车道分割过车数据
Map<String, List<TravelEvent>> carForLane = getCarForLane(car);
//车道道路流向关系
List<Map> laneList = getDictResult(connection, "SELECT lane_id,name,position,lane_length FROM dim_cnt_cross_lane_position\n" +
" where out_in_name='in' and cross_id='" + f1.getCrossCode() + "'");
//获取所有车道信息
List<Map> dictResult = getDictResult(connection, "select a.*,b.saturation_flow_rate from (\n" +
"SELECT lane_id,position,lane_length,GROUP_CONCAT(position,name) as positionName,GROUP_CONCAT(name) as name FROM dim_cnt_cross_lane_position\n" +
" where cross_id='" + f1.getCrossCode() + "' and out_in_type=1 group By lane_id,cross_id,position,lane_length)a left join app_atomic_lane_saturation_flow_rate b\n" +
" on a.lane_id=b.lane_id and b.cross_id='" + f1.getCrossCode() + "'");
//查询路口所有的方向
List<Map> positionList = getDictResult(connection, "SELECT position FROM dim_cnt_cross_lane_position\n" +
" where out_in_name='in' and cross_id='" + f1.getCrossCode() + "' GROUP BY position");
//查询路口所有的流向
List<Map> positionNameList = getDictResult(connection, " SELECT position,name FROM dim_cnt_cross_lane_position\n" +
" where out_in_name='in' and cross_id='" + f1.getCrossCode() + "' GROUP BY position,name");
List<Stage> stageList = f1.getStageList();
//获取相位关系
for (Stage stage : stageList) {
......@@ -100,6 +110,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
//获取所有车道的指标信息
List<LaneNorm> list = new ArrayList<LaneNorm>();
List<Map> laneList = dict.get("laneList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
List<Map> dictResult = dict.get("dictResult").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
List<Map> positionList = dict.get("positionList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
List<Map> positionNameList = dict.get("positionNameList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
for (Map map : dictResult) {
LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList);
list.add(laneNorm);
......@@ -121,10 +135,6 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
cycleTravel.put(f1.getCrossCode(),collect);
cycleTravelTemp.put(f1.getCrossCode(),new ArrayList<>());
processCycle.update(false);
//关闭mysql连接
if (connection != null) {
connection.close();
}
}
@Override
......@@ -147,6 +157,19 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<LaneNorm> out) throws Exception {
super.onTimer(timestamp, ctx, out);
System.out.println("触发定时器了。。。。");
//通过定时器清空字典表
dict.remove("phaseDirection");
dict.remove("laneList");
dict.remove("dictResult");
dict.remove("positionList");
dict.remove("positionNameList");
}
private static String gettEffectGreenTimeForPosition(List<Stage> stageList, String position) {
int AllGreenTime = 0;
for (Stage stage : stageList) {
......@@ -299,9 +322,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
LaneNorm laneNorm = new LaneNorm();
//车道属性 分别是车道编码 方向
String lane_id = String.valueOf(map.get("lane_id"));
String positionName = String.valueOf(map.get("positionName"));
String name = String.valueOf(map.get("name"));
//计算交通流量
Double trafficCapacity = getTrafficCapacity(cycle, car, lane_id);
//获取车道长度
Integer lane_length = Integer.parseInt(String.valueOf(map.get("lane_length")));
//车道基本信息
laneNorm.setOrder(0);
......@@ -325,22 +349,20 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
String[] effcetGreenTime = gettEffectGreenTime(cycle.getStageList(), lane_id, laneList);
//车道绿信比
double laneEffectGreenTime = Double.valueOf(effcetGreenTime[1]) / cycle.getDuration();
//车道级别指标计算
//最大排队长度
Integer maxQueueLength = getMaxQueueLength(car, lane_id);
Integer maxQueueLength = getMaxQueueLength(cycle,car, lane_id);
if (maxQueueLength > lane_length) {
maxQueueLength = lane_length;
}
double v = saturation_flow_rate * laneEffectGreenTime;
//车道级别指标计算
laneNorm.setMaxQueueLength(maxQueueLength);
laneNorm.setEffectGreenTime(effcetGreenTime[0]);
//通行能力
double v = saturation_flow_rate * laneEffectGreenTime;
BigDecimal bd2 = new BigDecimal(v);
Double d3 = bd2.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
laneNorm.setPassCapacity(d3);
//饱和流率
Double saturation = (trafficCapacity / (saturation_flow_rate * laneEffectGreenTime));
BigDecimal bd1 = new BigDecimal(laneEffectGreenTime==0d?0:saturation);
Double d2 = bd1.setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue();
......@@ -380,8 +402,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
/**
* 计算通行能力
*
* 计算交通流量
* 口径为 周期开始时间至周期结束时间内进入的车数量 单位为辆/h
* @param cycle
* @param laneCarMap
* @param laneId
......@@ -398,10 +420,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
} catch (Exception e) {
}
if (i == 0) {
if (i==0 || aLong == 0) {
return 0d;
}
BigDecimal spot = new BigDecimal((60 * 60) / (aLong / 1000));
BigDecimal spot = new BigDecimal(i * (60 * 60) / (aLong / 1000));
Double d1 = spot.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
return d1;
}
......@@ -441,23 +464,40 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
/**
* 根据过车查询车道的最大长度
*/
private static Integer getMaxQueueLength(Map<String, List<TravelEvent>> laneCarMap, String laneId) {
private static Integer getMaxQueueLength(Cycle cycle,Map<String, List<TravelEvent>> laneCarMap, String laneId) {
//排队车辆
int i = 0;
long cycleStart = Long.parseLong(cycle.getBeginDateTime());
long cycleEnd = Long.parseLong(cycle.getEndDateTime());
int maxQueneLength = 0;
try {
if (laneCarMap.containsKey(laneId)) {
i = laneCarMap.get(laneId).stream().
//获取这个车道所有的 过车数据
List<TravelEvent> travelEvents = laneCarMap.get(laneId);
//去周期内 每一秒的排队长度 取最大值
for(Long start = cycleStart; start <= cycleEnd; start += 1000) {
long start1 = start;
//通过每一秒分析出该车道停留的车数量
int queneLength = travelEvents.stream().filter(a-> Long.parseLong(a.getArrayTime())<=start1
&& (StringUtils.isEmpty(a.getInCorssTime()) || start1<Long.parseLong(a.getInCorssTime()))).collect(Collectors.toList()).size();
if(maxQueneLength<queneLength){
maxQueneLength=queneLength;
}
}
//旧版逻辑 根据有到达时间 没有进入时间的车辆取 最大排队长度
/*i = laneCarMap.get(laneId).stream().
filter(a -> StringUtils.isNotEmpty(a.getArrayTime()) &&
StringUtils.isEmpty(a.getInCorssTime())).collect(Collectors.toList()).size();
StringUtils.isEmpty(a.getInCorssTime())).collect(Collectors.toList()).size();*/
}
} catch (Exception e) {
}
return (i * 6);
return (maxQueneLength * 6);
}
private static Connection getConnection() throws Exception {
Connection connection = DorisUtils.getConnection();
//修改为从连接池中取连接
Connection connection = DruidConnectPoolUtils.getConncent();
//Connection connection = DorisUtils.getConnection();
return connection;
}
......@@ -488,10 +528,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
//处理过车数据 将其合并
private static List<TravelEvent> getCar(List<TravelEvent> cars) {
private static List<TravelEvent> getCar(List<TravelEvent> cars,Cycle cycle) {
List<TravelEvent> lists = new ArrayList<TravelEvent>();
Map<String, TravelEvent> map = new HashMap<String, TravelEvent>();
for (TravelEvent event : cars) {
//十分钟之前的数据 直接废弃
if (!DateUtils.compareDataIsBefore10Min(event.getEventTime())) {
continue;
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/LaneNormJob.java
......@@ -77,8 +77,8 @@ public class LaneNormJob {
DataStream<TravelEvent> carStream = carStringStream.map(getTravelEventFunction).assignTimestampsAndWatermarks(travelEventWatermarkStrategy).keyBy(carKeySelector).filter(filterFunction);
ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleDataStreamSource.connect(carStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
SingleOutputStreamOperator<LaneNorm> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction());
process.addSink(new LaneNormSink());
SingleOutputStreamOperator<LaneNorm> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4);
process.addSink(new LaneNormSink()).name("laneNormJobSink");
env.execute("laneNormJob");
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LaneNormSink.java
......@@ -3,6 +3,7 @@ package com.zhht.irn.sink;
import com.zhht.irn.entity.LaneNorm;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
......@@ -30,7 +31,7 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// path暂定为固定值
connection = DorisUtils.getConnection();
connection = DruidConnectPoolUtils.getConncent();
insertPstmt = connection.prepareStatement("insert into " +
"app_cross_dimension_norm(cross_id,lane_id,direction,cycle_order,max_queue_length,effect_green_time,pass_capacity,traffic_capacity,saturation,residual_capacity," +
"spot_space_share,dimension_name,`order`,statistic_time,`type`,spot_space_length,cycle_start_time,cycle_end_time,record_date) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
......@@ -70,5 +71,12 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
insertPstmt.setString(18,value.getCycleEndTime());
insertPstmt.setString(19,DateUtils.getTodayTime());
insertPstmt.execute();
/*if(counter < 10) {
counter ++;
insertPstmt.addBatch();
} else {
insertPstmt.executeBatch();
counter = 0;
}*/
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DruidConnectPoolUtils.java
package com.zhht.irn.utils;
import com.alibaba.druid.pool.DruidDataSource;
import org.apache.flink.api.java.utils.ParameterTool;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
......@@ -12,6 +14,9 @@ import java.sql.SQLException;
* @date 2022-09-05 15:11:00
*/
public class DruidConnectPoolUtils {
private static DruidDataSource pool=null;
/**
* 初始化连接池
* @param dbHosts
......@@ -25,17 +30,57 @@ public class DruidConnectPoolUtils {
dataSource.setUrl(dbHosts);
dataSource.setUsername(username);
dataSource.setPassword(password);
dataSource.setInitialSize(3); //初始连接数，默认0
dataSource.setMaxActive(30); //最大连接数，默认8
dataSource.setMaxActive(10); //最大连接数，默认8
dataSource.setMinIdle(5); //最小闲置数
dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除
dataSource.setTestOnBorrow(false); // 借出连接时不要测试，否则很影响性能
return dataSource;
}
public static void initDataSource(String dbHosts, String username, String password) {
DruidDataSource dataSource = new DruidDataSource();
dataSource.setDriverClassName("com.mysql.jdbc.Driver");
dataSource.setUrl(dbHosts);
dataSource.setUsername(username);
dataSource.setPassword(password);
dataSource.setInitialSize(10); //初始连接数，默认0
dataSource.setMaxActive(100); //最大连接数，默认8
dataSource.setMinIdle(5); //最小闲置数
dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除
dataSource.setTestOnBorrow(false); // 借出连接时不要测试，否则很影响性能
pool=dataSource;
}
public static Connection getConncent(){
try {
if(pool==null){
InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("doris.properties");
ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
// 根据路径的配置文件获取MySQL链接参数
String driver = tool.getRequired("driver");
String username = tool.getRequired("username");
String password = tool.getRequired("password");
String host = tool.getRequired("host");
String port = tool.getRequired("port");
String database = tool.get("database", "default");
String jdbcUrl = String.format("jdbc:mysql://%s:%s/%s", host, port, database);
initDataSource(jdbcUrl,username,password);
}
}catch (Exception e){
e.printStackTrace();
}
try {
return pool.getConnection();
} catch (SQLException e) {
e.printStackTrace();
}
return null;
}
/**
* CK close
* @param connection
......
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