实时即未来，车联网项目之远程诊断实时故障分析【七】

文章目录

• ​​远程诊断实时故障业务​​

• ​​什么是远程诊断实时故障​​
• ​​应用场景介绍​​
• ​​常用故障分析指标与含义​​
• ​​业务中间表数据结构​​
• ​​分析结果表数据结构​​
• ​​高德地图解决逆地理坐标问题​​

• ​​实时故障分析任务​​
• ​​远程诊断实时故障分析​​
• ​​车型车系销售信息广播流​​
• ​​获取地理位置信息​​

• ​​基于geohash编码的地理位置计算​​
• ​​定义redis操作的工具类​​

远程诊断实时故障业务

什么是远程诊断实时故障

监管部门或者车企通过判断实时上报的车辆数据，从而研判当前车辆故障诊断信息，给驾驶员发送预警告警信息等。

应用场景介绍

① 内部管理系统针对车辆的故障查询统计信息

② 实时监控大屏

常用故障分析指标与含义

• 19项车辆故障指标和车辆报警、故障信息属性50+

业务中间表数据结构

• 涉及到8张表和1章分析结果表
  [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5GOAHsGl-1664278460628)(assets/image-20210927092525648.png)]
• 表字段介绍
  分析的结果表 online_data 分为三类：

1. 实时上报的车辆数据；
2. 静态的车辆车型车系等数据；
3. 通过经纬度获取到的国家、省市区、地址等信息。

分析结果表数据结构

create table online_data ( vin varchar(17) not null comment '车架号' primary key, process_time datetime null comment '数据更新时间', lat double null comment '纬度', lng double null comment '经度', mileage double null comment '里程表读数', is_alarm int(1) null comment '故障标志(0正常，1故障)', alarm_name varchar(1000) null comment '故障名称(多个故障用~分割)', terminal_time datetime null comment '终端时间', earliest_time datetime null comment '最早数据接收时间', max_voltage_battery double null comment '单体电池最高电压', min_voltage_battery double null comment '单体电池最低电压', max_temperature_value double null comment '电池最高温度', min_temperature_value double null comment '电池最低温度', speed double null comment '车速', soc int(3) null comment 'SOC', charge_flag int(1) null comment '充电标识 0：未充电 1：充电 2：异常', total_voltage double null comment '总电压,单位：V，实际取值0.1~100V', total_current double null comment '总电流,单位：A,实际取值为-1000~1000A', battery_voltage varchar(1000) null comment '单体电池电压列表', probe_temperatures varchar(1000) null comment '电池模块温度列表', series_name varchar(255) null comment '车系', model_name varchar(255) null comment '车型', live_time int null comment '年限(单位:月，未查到数据显示-1)', sales_date varchar(20) null comment '销售日期', car_type varchar(20) null comment '车辆类型', province varchar(255) null comment '省份', city varchar(255) null comment '城市', county varchar(20) null comment '区（县）' );

高德地图解决逆地理坐标问题

• 拟地理编码含义
  输入位置信息（经度和维度）获取地球上位置。
• 高德等第三方Api支持拟地理演示地址

https://developer.amap.com/demo/javascript-api/example/geocoder/regeocoding/

• 如果使用高德Api的步骤

1. 获取key
2. 将key和经纬度参数封装为 url
3. 异步请求 httpGet 获取位置数据
4. 返回位置数据

• 远程实时诊断地理位置查询实现思路

实时故障分析任务

• 分析任务流程分析步骤

1. 消费数据，转换json对象
2. 过滤数据
3. 根据vin分组，创建timeWindow
4. 自定义window funtion，设置输出对象
5. 加载车型、车型、销售等数据并广播
6. 窗口数据与广播数据连接
7. 获得地理位置信息数据并与窗口数据连接
8. 结果落地数据到mysql中

• 实时故障分析流程

远程诊断实时故障分析

• 创建远程诊断实时故障分析任务主类—— OnlineStatisticsTask
• 开发步骤

1）初始化flink流处理的运行环境（事件时间、checkpoint、hadoop name）
2）接入kafka数据源，消费kafka数据
3）将消费到的json字符串转换成ItcastDataPartObj对象
4）过滤掉异常数据，保留正常数据
5）与redis维度表进行关联拉宽地理位置信息，对拉宽后的流数据关联redis，根据geohash找到地理位置信息，进行拉宽操作
6）过滤出来redis拉宽成功的地理位置数据
7）过滤出来redis拉宽失败的地理位置数据
8）对redis拉宽失败的地理位置数据使用异步io访问高德地图逆地理位置查询地理位置信息，并将返回结果写入到redis中
9）将reids拉宽的地理位置数据与高德api拉宽的地理位置数据进行合并
10）创建原始数据的30s的滚动窗口，根据vin进行分流操作
11）对原始数据的窗口流数据进行实时故障分析（区分出来告警数据和非告警数据19个告警字段）
12）加载业务中间表（7张表：车辆表、车辆类型表、车辆销售记录表，车俩用途表4张），并进行广播
13）将第11步和第12步的广播流结果进行关联，并应用拉宽操作
14）将拉宽后的结果数据写入到mysql数据库中
15）启动作业

• 需要获取地理位置对象，可以作为ItcastDataPartObj的父类

@Data
@NoArgsConstructor
@AllArgsConstructor
public class VehicleLocationModel implements Serializable {
    //省份
    private String province;
    //城市
    private String city;
    //国家
    private String country;
    //区县
    private String district;
    //详细地址
    private String address;
    //纬度
    private Double lat = -999999D;
    //经度
    private Double lng = -999999D;
}

• 实现合并流数据在redis存储的地理位置数据拉宽操作——LocationInfoRedisFunction

//继承 RichMapFunction<ItcastDataPartObj, ItcastDataPartObj>
//1.重写 map 方法
//1.1.获取车辆数据的经度和维度生成 geohash
//1.2.根据geohash 从redis中获取value值（geohash在redis中是作为主键存在）
//1.3.如果查询出来的值不为空，将其通过JSON对象转换成 VehicleLocationModel 对象，否则置为 null
//1.4.如果当前对象不为空，将国家，省市区地址赋值给 itcastDataPartObj，否则置为 null
//1.5.返回数据

• 对在redis获取失败的经纬度使用异步io流请求高德Api——AsyncHttpQueryFunction

//1.重写open方法
//1.1.创建请求配置
//1.2.创建Http异步的客户端
//1.3.开启client
//2.重写close方法
//3.重写timeout方法
//3.1.打印输出超时
//4.重写asyncInvoke方法
//4.1.获取当前车辆的经纬度
//4.2.通过GaoDeMapUtils工具类根据参数获取请求的url
//4.3.创建 http get请求对象
//4.4.使用刚创建的http异步客户端执行 http请求对象
//4.5.从执行完成的future中获取数据，返回ItcastDataPartObj对象
//4.5.1.重写get方法
//4.5.1.1.使用future获取到返回的值
//判断如果返回值的状态是正常值 200
//获取到响应的实体对象 entity
//将实体对象使用EntityUtils转换成string字符串
//因为返回的是json，需要使用JSON转换成JSONObject对象
//通过regeocode获取JSON对象，然后解析对象封装国家，省市区，地址
//封装成 VehicleLocationModel 对象
//4.5.1.2.通过RedisUtil将数据写入到redis，
//key=geohash，value=封装的对象的JSON字符串toJSONString
//4.5.1.3.将国家，省市区，地址进行封装并返回
//4.6.从future的thenAccept
//4.6.1.重写accept方法，使用集合中只放一个对象

• 引入高德Api 访问的工具类

public class GaoDeMapUtils {
    //指定高德地图请求的密钥
    private static final String KEY = ConfigLoader.getProperty("gaode.key");
    //指定返回值类型
    private static final String OUTPUT = "json";
    //请求的地址
    private static final String GET_ADDRESS_URL = ConfigLoader.getProperty("gaode.address.url");

    /**
     * 传递经纬度返回逆地理位置查询的请求地址
     * @param longitude
     * @param latitude
     * @return
     */
    public static String getUrlByLonLat(double longitude, double latitude) {
        //拼接经纬度的字符串参数
        String location = longitude + "," + latitude;
        //定义参数的集合对象
        Map<String, String> params = new HashMap<>();
        params.put("location", location);

        //根据请求base地址和参数集合列表拼接出来请求的完整地址
        String url = joinUrl(params, GET_ADDRESS_URL);
        return url;
    }

    /**
     * 拼接请求的参数和请求地址
     * @param params
     */
    private static String joinUrl(Map<String, String> params, String url) {
        StringBuilder baseUrl = new StringBuilder();
        baseUrl.append(url);

        try {
            //指定参数的索引
            int index = 0;
            Set<Map.Entry<String, String>> entries = params.entrySet();
            for (Map.Entry<String, String> param : entries) {
                if (index == 0) {
                    baseUrl.append("?");
                } else {
                    baseUrl.append("&");
                }
                //拼接所有的参数
                baseUrl.append(param.getKey()).append("=").append(URLEncoder.encode(param.getValue(), "utf-8"));
            }
            baseUrl.append("&output=").append(OUTPUT).append("&key=").append(KEY);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return baseUrl.toString();
    }
}

• 导入实时在线故障分析对象，用于存储在线故障对象——OnlineDataObj

/**
 * 实时在线故障分析javaBean对象
 */
@Data
public class OnlineDataObj extends VehicleLocationModel  {
    //车架号
    private String vin;
    //数据更新时间
    private String processTime;
    //里程表读数
    private double mileage;
    //故障标志(0正常，1故障)
    private int isAlarm;
    //故障名称(多个故障用~分割)
    private String alarmName;
    //终端时间
    private String terminalTime;
    //最早数据接收时间
    private String earliestTime;
    //单体电池最高电压
    private double maxVoltageBattery;
    //单体电池最低电压
    private double minVoltageBattery;
    //电池最高温度
    private double maxTemperatureValue;
    //电池最低温度
    private double minTemperatureValue;
    //车速
    private double speed;
    //SOC
    private int soc;
    //充电标识 0：未充电 1：充电 2：异常
    private int chargeFlag;
    //总电压,单位：V，实际取值0.1~100V
    private double totalVoltage;
    //总电流,单位：A,实际取值为-1000~1000A
    private double totalCurrent;
    //单体电池电压列表
    private String batteryVoltage;
    //电池模块温度列表
    private String probeTemperatures;
    //车系
    private String seriesName;
    //车型
    private String modelName;
    //年限(单位:月，未查到数据显示-1)
    private String liveTime;
    //销售日期
    private String salesDate;
    //车辆类型
    private String carType;
    //省份
    private String province;
    //城市
    private String city;
    //国家
    private String county;
    //区县
    private String district;
    //详细地址
    private String address;
}

• 窗口自定义远程故障诊断自定义窗口实现——OnlineStatisticsWindowFunction

//实现WindowFunction<ItcastDataPartObj, OnlineDataObj, String, TimeWindow>接口
//1.对当前的数据集合进行升序排列
//2.获取集合中第一条数据
//3.循环遍历每条数据，将集合中存在异常的数据拼接到指定属性中
//30s窗口最多6条数据，每条数据需要检测19个字段，如果出现异常字段就进行  //字符串拼接
//3.1.过滤没有各种告警的信息，调用setOnlineDataObj 将第一条对象和每条对象和标识0 返回到OnlineDataObj，并收集这个对象
// 否则 调用setOnlineDataObj 将第一条对象和每条对象和标识1 返回到OnlineDataObj，并收集这个对象
//4.实现setOnlineDataObj 方法
//4.1.定义OnlineDataObj
//4.2.将每条的对象属性拷贝到定义OnlineDataObj
//4.3.将每条对象中表显里程赋值给mileage
//4.4.将告警信号赋值给isAlarm
//4.5.将每个对象通过addAlarmNameList生成告警list，拼接成字符串赋值给alarmName，通过字符串join
//4.6.将窗口内第一条数据告警时间赋值给 earliestTime
//4.7.将获取每条记录的充电状态通过getChargeState返回充电标识赋值给充电标记
//4.8.将当前时间赋值给处理时间
//引入-判断是否存在报警的字段，addAlarmNameList，getChargeState

车型车系销售信息广播流

• 涉及到的字段

车型、车系、车辆销售信息数据，主要获得9个字段信息：

vin、series_name、model_name、series_code、model_code、nick_name、sales_date、product_date、car_type

• 数据源模型
• 从MySQL中读取车型车系销售信息

select t12.vin,t12.series_name,t12.model_name,t12.series_code,t12.model_code,t12.nick_name,t3.sales_date product_date,t4.car_type
 from (select t1.vin, t1.series_name, t2.show_name as model_name, t1.series_code,t2.model_code,t2.nick_name,t1.vehicle_id
 from vehicle_networking.dcs_vehicles t1 left join vehicle_networking.t_car_type_code t2 on t1.model_code = t2.model_code) t12
 left join  (select vehicle_id, max(sales_date) sales_date from vehicle_networking.dcs_sales group by vehicle_id) t3
 on t12.vehicle_id = t3.vehicle_id
 left join
 (select tc.vin,'net_cat' car_type from vehicle_networking.t_net_car tc
 union all select tt.vin,'taxi' car_type from vehicle_networking.t_taxi tt
 union all select tp.vin,'private_car' car_type from vehicle_networking.t_private_car tp
 union all select tm.vin,'model_car' car_type from vehicle_networking.t_model_car tm) t4
 on t12.vin = t4.vin

• [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eqD9pRLV-1664278460629)(assets/image-20210927174106033.png)]
• 导入车辆车型车系结果对象

/**
 * 定义车辆基础信息表的javaBean对象
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class VehicleInfoModel {
    //车架号
    private String vin;
    //车型编码
    private String modelCode;
    //车型名称
    private String modelName;
    //车系编码
    private String seriesCode;
    //车系名称
    private String seriesName;
    //出售日期
    private String salesDate;
    //车型
    private String carType;
    //车辆类型简称
    private String nickName;
    //年限
    private String liveTime;
}

• 创建读取MySQL的Flink的数据源Source——VehicleInfoMysqlSource

• 实现RichSourceFunction<HashMap<String, VehicleInfoModel>>
• 将数据源广播出去

//自定义实现车辆基础信息表的加载
//加载车辆基础信息表（车辆类型、车辆、销售记录表、车辆用途表）
//重写open方法
//重写run方法
//重写close方法
//重写cancel方法

• 窗口流数据与广播数据connect再flatMap——VehicleInfoMapMysqlFunction

//继承 RichCoFlatMapFunction<OnlineDataObj, HashMap<String, VehicleInfoModel>, OnlineDataObj>
//1.重写flatMap1
//1.1.通过 vin 获取到车辆基础信息
//1.2.如果车辆基础信息不为空
//1.2.1.将车系seriesName，车型modelName,年限LiveTime,销售日期saleDate，车辆类型carType封装到onlineDataObj对象中
//1.2.2.将onlineDataObj收集返回
//1.3.打印输出，基础信息不存在
//2.重写flatMap2
//赋值基本配置给变量

获取地理位置信息

基于geohash编码的地理位置计算

• geohash的概念介绍（高效的多维空间点索引算法.html）

geohash 就是将地图上位置（经纬度）转换成偶数位是经度、奇数数是维度，新的二进制字节，转换成字符串，用字符串代表某一个地理位置。

• geohash编码实现

/** * @Description TODO geohash算法实现工具类 */ public class GeoHashUtil { // todo 经纬度编码长度 private static int numbits = 6 * 5; // todo 数字字母编码数组 final static char[] digits = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'j', 'k', 'm', 'n', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z'}; // todo 存储编码字符循环存储的HashMap对象 final static HashMap<Character, Integer> lookup = new HashMap<>(); // todo 静态代码块，执行设置HashMap的key，value static { int i = 0; for (char c : digits) lookup.put(c, i++); } /** * @desc todo 根据编码后的geohash字符串值，进行解码，得到经纬度数组 * @param geoHash * @return [lat, lon] */ public static double[] decode(String geoHash) { StringBuilder buffer = new StringBuilder(); for (char c : geoHash.toCharArray()) { int i = lookup.get(c) + 32; buffer.append(Integer.toString(i, 2).substring(1)); } BitSet lonset = new BitSet(); BitSet latset = new BitSet(); // todo 经度，偶数位 int j = 0; for (int i = 0; i < numbits * 2; i += 2) { boolean isSet = false; if (i < buffer.length()) isSet = buffer.charAt(i) == '1'; lonset.set(j++, isSet); } // todo 纬度，奇数位 j = 0; for (int i = 1; i < numbits * 2; i += 2) { boolean isSet = false; if (i < buffer.length()) isSet = buffer.charAt(i) == '1'; latset.set(j++, isSet); } // todo 根据位编码、经度最小值、经度最大值计算出经度 double lon = decode(lonset, -180, 180); // todo 根据位编码、纬度最小值、纬度最大值计算出经度 double lat = decode(latset, -90, 90); // todo 返回纬度、经度数组 return new double[]{lat, lon}; } /** * @desc todo 编码方法，根据编码、维度[-90,90],经度[-180,180] * @param bs * @param floor * @param ceiling * @return */ private static double decode(BitSet bs, double floor, double ceiling) { double mid = 0; for (int i = 0; i < bs.length(); i++) { mid = (floor + ceiling) / 2; if (bs.get(i)) floor = mid; else ceiling = mid; } return mid; } /** * @desc todo 解码方法，根据纬度、经度，返回32编码字符串 * @param lat 纬度 * @param lon 经度 * @return base32的字符串 */ public static String encode(double lat, double lon) { BitSet latbits = getBits(lat, -90, 90); BitSet lonbits = getBits(lon, -180, 180); StringBuilder buffer = new StringBuilder(); for (int i = 0; i < numbits; i++) { buffer.append(lonbits.get(i) ? '1' : '0'); buffer.append(latbits.get(i) ? '1' : '0'); } return base32(Long.parseLong(buffer.toString(), 2)); } /** * @desc todo 根据经纬度和范围，获取对应的二进制值 * @param d 经度 | 纬度 * @param floor 最小值 * @param ceiling 最大值 * @return 返回BitSet，java.util工具类，用于位移操作工具类 */ private static BitSet getBits(double d, double floor, double ceiling) { BitSet buffer = new BitSet(numbits); for (int i = 0; i < numbits; i++) { double mid = (floor + ceiling) / 2; if (d >= mid) { buffer.set(i); floor = mid; } else { ceiling = mid; } } return buffer; } /** * @desc todo 将经纬度合并后二二进制进行指定32位编码 * @param i 被32编码的long值 * @return 32编码字符串 */ private static String base32(long i) { char[] buf = new char[65]; int charPos = 64; boolean negative = (i < 0); if (!negative) i = -i; while (i <= -32) { buf[charPos--] = digits[(int) (-(i % 32))]; i /= 32; } buf[charPos] = digits[(int) (-i)]; if (negative) buf[--charPos] = '-'; return new String(buf, charPos, (65 - charPos)); } } /** * @Description TODO 测试geohashutil 验证：http://www.geohash.cn/ */ public class TestGeoHashUtil { public static void main(String[] args) { // todo 根据经纬度编码 String geohash = GeoHashUtil.encode(25.050583, 121.559322); System.out.println(geohash); // todo geohash值解码 double[] geo = GeoHashUtil.decode("wsqqw0kf1x0h"); System.out.println(geo[0]+" "+geo[1]); } }

• 引入geohash工具类引入

/**
 * @Description TODO geohash算法实现工具类
 */
public class GeoHashUtil {
    // todo 经纬度编码长度
    private static int numbits =  6 * 5;
    // todo 数字字母编码数组
    final static char[] digits = {'0', '1', '2', '3', '4', '5', '6', '7',
            '8', '9', 'b', 'c', 'd', 'e', 'f', 'g',
            'h', 'j', 'k', 'm', 'n', 'p', 'q', 'r',
            's', 't', 'u', 'v', 'w', 'x', 'y', 'z'};
    // todo 存储编码字符循环存储的HashMap对象
    final static HashMap<Character, Integer> lookup = new HashMap<>();

    // todo 静态代码块，执行设置HashMap的key，value
    static {
        int i = 0;
        for (char c : digits)
            lookup.put(c, i++);
    }

    /**
     * @desc todo 根据编码后的geohash字符串值，进行解码，得到经纬度数组
     * @param geoHash
     * @return [lat, lon]
     */
    public static double[] decode(String geoHash) {
        StringBuilder buffer = new StringBuilder();
        for (char c : geoHash.toCharArray()) {
            int i = lookup.get(c) + 32;
            buffer.append(Integer.toString(i, 2).substring(1));
        }

        BitSet lonset = new BitSet();
        BitSet latset = new BitSet();

        // todo 经度，偶数位
        int j = 0;
        for (int i = 0; i < numbits * 2; i += 2) {
            boolean isSet = false;
            if (i < buffer.length())
                isSet = buffer.charAt(i) == '1';
            lonset.set(j++, isSet);
        }

        // todo 纬度，奇数位
        j = 0;
        for (int i = 1; i < numbits * 2; i += 2) {
            boolean isSet = false;
            if (i < buffer.length())
                isSet = buffer.charAt(i) == '1';
            latset.set(j++, isSet);
        }

        // todo 根据位编码、经度最小值、经度最大值计算出经度
        double lon = decode(lonset, -180, 180);
        // todo 根据位编码、纬度最小值、纬度最大值计算出经度
        double lat = decode(latset, -90, 90);
        // todo 返回纬度、经度数组
        return new double[]{lat, lon};
    }

    /**
     * @desc todo 编码方法，根据编码、维度[-90,90],经度[-180,180]
     * @param bs
     * @param floor
     * @param ceiling
     * @return
     */
    private static double decode(BitSet bs, double floor, double ceiling) {
        double mid = 0;
        for (int i = 0; i < bs.length(); i++) {
            mid = (floor + ceiling) / 2;
            if (bs.get(i))
                floor = mid;
            else
                ceiling = mid;
        }
        return mid;
    }

    /**
     * @desc todo 解码方法，根据纬度、经度，返回32编码字符串
     * @param lat 纬度
     * @param lon 经度
     * @return base32的字符串
     */
    public static String encode(double lat, double lon) {
        BitSet latbits = getBits(lat, -90, 90);
        BitSet lonbits = getBits(lon, -180, 180);
        StringBuilder buffer = new StringBuilder();
        for (int i = 0; i < numbits; i++) {
            buffer.append(lonbits.get(i) ? '1' : '0');
            buffer.append(latbits.get(i) ? '1' : '0');
        }
        return base32(Long.parseLong(buffer.toString(), 2));
    }

    /**
     * @desc todo 根据经纬度和范围，获取对应的二进制值
     * @param d 经度 | 纬度
     * @param floor 最小值
     * @param ceiling 最大值
     * @return 返回BitSet，java.util工具类，用于位移操作工具类
     */
    private static BitSet getBits(double d, double floor, double ceiling) {
        BitSet buffer = new BitSet(numbits);
        for (int i = 0; i < numbits; i++) {
            double mid = (floor + ceiling) / 2;
            if (d >= mid) {
                buffer.set(i);
                floor = mid;
            } else {
                ceiling = mid;
            }
        }
        return buffer;
    }

    /**
     * @desc todo 将经纬度合并后二二进制进行指定32位编码
     * @param i 被32编码的long值
     * @return 32编码字符串
     */
    private static String base32(long i) {
        char[] buf = new char[65];
        int charPos = 64;
        boolean negative = (i < 0);
        if (!negative)
            i = -i;
        while (i <= -32) {
            buf[charPos--] = digits[(int) (-(i % 32))];
            i /= 32;
        }
        buf[charPos] = digits[(int) (-i)];

        if (negative)
            buf[--charPos] = '-';
        return new String(buf, charPos, (65 - charPos));
    }
}
/**
 * @Description TODO 测试geohashutil 验证：http://www.geohash.cn/
 */
public class TestGeoHashUtil {
    public static void main(String[] args) {
        // todo 根据经纬度编码
        String geohash = GeoHashUtil.encode(25.050583, 121.559322);
        System.out.println(geohash);
        // todo geohash值解码
        double[] geo = GeoHashUtil.decode("wsqqw0kf1x0h");
        System.out.println(geo[0]+" "+geo[1]);
    }
}

定义redis操作的工具类

• 导入 redis 操作工具类——RedisUtil

public class RedisUtil {
    private static Pool<Jedis> jedisPool = null;
    private static ReentrantLock lock = new ReentrantLock();
    private static String HOST = ConfigLoader.getProperty("redis.host");
    private static int PORT = Integer.valueOf(ConfigLoader.getProperty("redis.port"));
    private static int TIMEOUT = Integer.valueOf(ConfigLoader.getProperty("redis.session.timeout"));
    private static int DATABASE = Integer.valueOf(ConfigLoader.getProperty("redis.database"));
    private static String PASSWORD = ConfigLoader.getProperty("redis.password");

    /**
     * 初始化连接池
     */
    static {
        if ("null".equals(PASSWORD)) PASSWORD = null;
        if (jedisPool == null) {
            jedisPool = new JedisPool(new GenericObjectPoolConfig(), HOST, PORT, TIMEOUT, PASSWORD, DATABASE, "");
        }
    }

    /**
     * @desc:获得jedis客户端
     * @return Jedis客户端
     */
    public static Jedis getJedis() {
        if (jedisPool == null) {
            lock.lock(); //防止吃初始化时多线程竞争问题
            jedisPool = new JedisPool(new GenericObjectPoolConfig(), HOST, PORT, TIMEOUT, PASSWORD, DATABASE, "");
            lock.unlock();
        }
        return jedisPool.getResource();
    }

    /**
     * @desc:根据redis中存在的key获得value
     * @param key
     * @return value的字节数组
     */
    public static byte[] get(byte[] key) {
        Jedis jedis = getJedis();
        byte[] result = "".getBytes();
        if (jedis.exists(key)) {
            result = jedis.get(key);
        }
        jedis.close();
        return result;
    }

    /**
     * @desc:插入数据到redis中，并设置key的存活时间（seconds）
     * @param key
     * @param value
     * @param keyTimeout
     * @return
     */
    public static Boolean set(byte[] key, byte[] value, int keyTimeout) {
        try {
            Jedis jedis = getJedis();
            jedis.setex(key, keyTimeout, value);
            jedis.close();
            return true;
        } catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }

    public static void set(byte[] key, byte[] value) {
        try {
            Jedis jedis = getJedis();
            jedis.set(key, value);
            jedis.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * @desc:释放资源，关闭连接池
     */
    public static void releaseSource() {
        if (jedisPool != null) jedisPool.close();
    }
}