数据转换

base/common-data-es/src/main/java/com/yihu/base/es/config/ElastricSearchHelper.java

@@ -59,6 +59,37 @@ public class ElastricSearchHelper {
        return null;
    }
    public Boolean save(String index, String type, String source) {
        JestClient jestClient = null;
        try {
            //得到链接elasticFactory.getJestClient();
            int success = 0;
            int error = 0;
            Bulk.Builder bulk = new Bulk.Builder().defaultIndex(index).defaultType(type);
            try {
                Index indexObj = new Index.Builder(source).build();
                success++;
                bulk.addAction(indexObj);
            } catch (Exception e) {
                logger.error(e.getMessage());
                error++;
            }
            BulkResult br = jestClient.execute(bulk.build());
            logger.info("save flag:" + br.isSucceeded());
            logger.info("save success:" + success);
            logger.info("save error:" + error);
            return br.isSucceeded();
        } catch (Exception e) {
            logger.error(" save error ：" + e.getMessage());
        } finally {
            if (jestClient != null) {
                jestClient.shutdownClient();
            }
        }
        return null;
    }
    public Boolean update(String index, String type, List<SaveModel> sms) {
        JestClient jestClient = null;

common/common-entity/src/main/java/com/yihu/jw/iot/data_input/DataProcessLogDO.java

@@ -0,0 +1,116 @@
package com.yihu.jw.iot.data_input;
import com.yihu.jw.IdEntityWithOperation;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Table;
import java.io.Serializable;
@Entity
@Table(name = "iot_data_process_log")
public class DataProcessLogDO extends IdEntityWithOperation implements Serializable {
    @Column(name = "file_name")
    private String fileName;
    @Column(name = "file_absoulte_path")
    private String fileAbsoultePath;
    @Column(name = "data_source")
    private String dataSource;
    @Column(name = "receive_time")
    private String receiveTime;
    @Column(name = "upload_time")
    private String uploadTime;
    @Column(name = "process_type")
    private String processType;
    @Column(name = "process_status")
    private String processStatus;
    @Column(name = "process_interface")
    private String processInterface;
    @Column(name = "process_des")
    private String processDes;
    @Column(name = "fail_count")
    private int fileCount;
    public String getFileName() {
        return fileName;
    }
    public void setFileName(String fileName) {
        this.fileName = fileName;
    }
    public String getFileAbsoultePath() {
        return fileAbsoultePath;
    }
    public void setFileAbsoultePath(String fileAbsoultePath) {
        this.fileAbsoultePath = fileAbsoultePath;
    }
    public String getDataSource() {
        return dataSource;
    }
    public void setDataSource(String dataSource) {
        this.dataSource = dataSource;
    }
    public String getReceiveTime() {
        return receiveTime;
    }
    public void setReceiveTime(String receiveTime) {
        this.receiveTime = receiveTime;
    }
    public String getUploadTime() {
        return uploadTime;
    }
    public void setUploadTime(String uploadTime) {
        this.uploadTime = uploadTime;
    }
    public String getProcessType() {
        return processType;
    }
    public void setProcessType(String processType) {
        this.processType = processType;
    }
    public String getProcessStatus() {
        return processStatus;
    }
    public void setProcessStatus(String processStatus) {
        this.processStatus = processStatus;
    }
    public String getProcessDes() {
        return processDes;
    }
    public void setProcessDes(String processDes) {
        this.processDes = processDes;
    }
    public int getFileCount() {
        return fileCount;
    }
    public void setFileCount(int fileCount) {
        this.fileCount = fileCount;
    }
}

common/common-entity/src/main/java/com/yihu/jw/iot/data_input/DataStandardDO.java

@@ -8,7 +8,7 @@ import javax.persistence.Table;
import java.io.Serializable;
@Entity
@Table("iot_base_data_def")
@Table(name = "iot_base_data_def")
public class DataStandardDO extends IdEntityWithOperation implements Serializable {
    @Column(name = "base_name")
@@ -32,4 +32,59 @@ public class DataStandardDO extends IdEntityWithOperation implements Serializabl
    @Column(name = "item_value_max")
    private String itemValueMax;
    public String getBaseName() {
        return baseName;
    }
    public void setBaseName(String baseName) {
        this.baseName = baseName;
    }
    public String getItemCode() {
        return itemCode;
    }
    public void setItemCode(String itemCode) {
        this.itemCode = itemCode;
    }
    public String getItemName() {
        return itemName;
    }
    public void setItemName(String itemName) {
        this.itemName = itemName;
    }
    public String getItemType() {
        return itemType;
    }
    public void setItemType(String itemType) {
        this.itemType = itemType;
    }
    public String getRequired() {
        return required;
    }
    public void setRequired(String required) {
        this.required = required;
    }
    public String getItemValueMin() {
        return itemValueMin;
    }
    public void setItemValueMin(String itemValueMin) {
        this.itemValueMin = itemValueMin;
    }
    public String getItemValueMax() {
        return itemValueMax;
    }
    public void setItemValueMax(String itemValueMax) {
        this.itemValueMax = itemValueMax;
    }
}

common/common-entity/src/main/java/com/yihu/jw/iot/device/IotDeviceDO.java

@@ -54,6 +54,9 @@ public class IotDeviceDO extends IdEntityWithOperation implements Serializable {
    @Column(name = "manufacturer_name")
    private String manufacturerName;//厂商名称
    @Column(name = "manufacturer_tel")
    private String manufactureTel;//厂商名称
    @Column(name = "order_code")
    private String orderCode;//订单code
@@ -75,6 +78,12 @@ public class IotDeviceDO extends IdEntityWithOperation implements Serializable {
    @Column(name = "del")
    private Integer del;//删除标志(1有效，0删除)
    @Column(name = "sale_org_code")
    private String saleOrgCode; //经销商单位代码
    @Column(name = "sale_org_name")
    private String saleOrgName; //经销商单位名称
    public String getSaasId() {
        return saasId;
    }
@@ -179,6 +188,14 @@ public class IotDeviceDO extends IdEntityWithOperation implements Serializable {
        this.manufacturerName = manufacturerName;
    }
    public String getManufactureTel() {
        return manufactureTel;
    }
    public void setManufactureTel(String manufactureTel) {
        this.manufactureTel = manufactureTel;
    }
    public String getOrderCode() {
        return orderCode;
    }
@@ -234,4 +251,20 @@ public class IotDeviceDO extends IdEntityWithOperation implements Serializable {
    public void setDel(Integer del) {
        this.del = del;
    }
    public String getSaleOrgCode() {
        return saleOrgCode;
    }
    public void setSaleOrgCode(String saleOrgCode) {
        this.saleOrgCode = saleOrgCode;
    }
    public String getSaleOrgName() {
        return saleOrgName;
    }
    public void setSaleOrgName(String saleOrgName) {
        this.saleOrgName = saleOrgName;
    }
}

common/common-request-mapping/src/main/java/com/yihu/jw/rm/iot/DataRequestMapping.java

@@ -0,0 +1,31 @@
package com.yihu.jw.rm.iot;
public class DataRequestMapping {
    public static final String api_iot_common = "svr-iot";
    public static final Integer api_iot_fail = -1;
    /**
     * 数据标准转换
     */
    public static class DataStandardConvert{
        public static final String api_convert = "dataConvert";//文件流上传
        public static final String message_success_convert = "转换成功";
        public static final String message_fail_convert = "转换失败";
        public static final String message_fail_jsonData_is_null = "jsonData is null";
    }
    /**
     * 数据上传
     */
    public static class DataInput{
        public static final String api_data_input = "dataInput";//数据上传
        public static final String api_user_bind = "userBind";//设备注册绑定
        public static final String message_success = "上传成功";
        public static final String message_fail = "上传失败";
        public static final String message_fail_jsonData_is_null = "jsonData is null";
    }
}

svr/svr-base/src/main/resources/bootstrap.yml

@@ -3,11 +3,11 @@ spring:
  #从发现服务里面取配置服务的信息
  cloud:
    config:
      failFast: true ##启动快速失败 即链接不到配置服务就启动失败
      failFast: true #启动快速失败 即链接不到配置服务就启动失败
      username: jw
      password: jkzl
      discovery:
        enabled: true ##使用发现服务
        enabled: true #使用发现服务
        service-id: svr-configurations ##配置服务的名字
@@ -20,7 +20,8 @@ eureka:
  client:
    serviceUrl:
      #http://账号:密码@127.0.0.1:8761/eureka/
      defaultZone: http://jw:jkzl@localhost:8761/eureka/
      defaultZone: http://jw:jkzl@172.19.103.33:8761//eureka/
#      defaultZone: http://jw:jkzl@127.0.0.1:8761//eureka/
---
spring:

svr/svr-iot/pom.xml

@@ -102,5 +102,38 @@
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-config</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-test-autoconfigure</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
        <dependency>
            <groupId>com.yihu.base</groupId>
            <artifactId>common-data-es</artifactId>
            <version>1.0.0</version>
        </dependency>
        <dependency>
            <groupId>com.yihu.base</groupId>
            <artifactId>common-data-hbase</artifactId>
            <version>1.0.0</version>
        </dependency>
    </dependencies>
</project>

svr/svr-iot/src/main/java/com/yihu/iot/dao/device/IotDeviceDao.java

@@ -13,4 +13,10 @@ public interface IotDeviceDao extends PagingAndSortingRepository<IotDeviceDO, Lo
    @Query("from IotDeviceDO w where w.id =?1")
    IotDeviceDO findById(String id);
    @Query("count(*) from IotDeviceDO w where w.deviceSn =?1")
    int countByDeviceSn(String deviceSn);
    @Query("from IotDeviceDO w where w.deviceSn =?1")
    IotDeviceDO findByDeviceSn(String deviceSn);
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/DataInputEnrty.java

@@ -1,4 +0,0 @@
package com.yihu.iot.data_input;
public class DataInputEnrty {
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/DataStandardConvert.java

@@ -1,9 +0,0 @@
package com.yihu.iot.data_input;
/**
 * 数据标准转换
 */
public class DataStandardConvert {
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/DataStandardService.java

@@ -1,10 +0,0 @@
package com.yihu.iot.data_input;
import com.yihu.base.mysql.query.BaseJpaService;
import com.yihu.jw.iot.data_input.DataStandardDO;
public class DataStandardService extends BaseJpaService<DataStandardDO,DataStandardDao> {
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/controller/DataInputController.java

@@ -0,0 +1,44 @@
package com.yihu.iot.data_input.controller;
import com.yihu.iot.data_input.service.DataInputService;
import com.yihu.jw.exception.ApiException;
import com.yihu.jw.restmodel.common.Envelop;
import com.yihu.jw.rm.iot.DataRequestMapping;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping(DataRequestMapping.api_iot_common)
@Api(value = "数据上传操作", description = "数据上传操作")
public class DataInputController {
    @Autowired
    private DataInputService dataInputService;
    @PostMapping(value = DataRequestMapping.DataInput.api_user_bind, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    @ApiOperation(value = "设备注册绑定", notes = "设备注册并绑定用户")
    public Envelop bindUser(@ApiParam(name = "json_data", value = "", defaultValue = "") @RequestBody String jsonData){
        try{
            return Envelop.getSuccess(DataRequestMapping.DataInput.message_success,dataInputService.bindUser(jsonData));
        } catch (ApiException e){
            return Envelop.getError(e.getMessage(), e.getErrorCode());
        }
    }
    @PostMapping(value = DataRequestMapping.DataInput.api_data_input, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    @ApiOperation(value = "数据上传", notes = "数据上传入库")
    public Envelop uoloadData(@ApiParam(name = "json_data", value = "", defaultValue = "") @RequestBody String jsonData){
        try {
            return Envelop.getSuccess(DataRequestMapping.DataInput.message_success, dataInputService.uploadData(jsonData));
        } catch (ApiException e) {
            return Envelop.getError(e.getMessage(), e.getErrorCode());
        }
    }
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/controller/DataStandardConvertController.java

@@ -0,0 +1,38 @@
package com.yihu.iot.data_input.controller;
import com.yihu.iot.data_input.service.DataStandardConvertService;
import com.yihu.jw.base.user.BaseMenuDO;
import com.yihu.jw.exception.ApiException;
import com.yihu.jw.restmodel.common.Envelop;
import com.yihu.jw.rm.base.BaseUserRequestMapping;
import com.yihu.jw.rm.iot.DataRequestMapping;
import com.yihu.jw.rm.iot.IotRequestMapping;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping(DataRequestMapping.api_iot_common)
@Api(value = "数据标准转换操作", description = "数据标准转换接口")
public class DataStandardConvertController {
    @Autowired
    private DataStandardConvertService dataStandardConvertService;
    @PostMapping(value = DataRequestMapping.DataStandardConvert.api_convert, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    @ApiOperation(value = "数据标准转换", notes = "数据标准转换")
    public Envelop convert(@ApiParam(name = "json_data", value = "", defaultValue = "") @RequestBody String jsonData){
        try{
            return Envelop.getSuccess(DataRequestMapping.DataStandardConvert.message_success_convert,dataStandardConvertService.iconvert(jsonData));
        } catch (ApiException e){
            return Envelop.getError(e.getMessage(), e.getErrorCode());
        }
    }
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/dao/DataProcessLogDao.java

@@ -0,0 +1,9 @@
package com.yihu.iot.data_input.dao;
import com.yihu.jw.iot.data_input.DataProcessLogDO;
import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
import org.springframework.data.repository.PagingAndSortingRepository;
public interface DataProcessLogDao extends PagingAndSortingRepository<DataProcessLogDO, Long>, JpaSpecificationExecutor<DataProcessLogDO> {
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/DataStandardDao.java

@@ -1,4 +1,4 @@
package com.yihu.iot.data_input;
package com.yihu.iot.data_input.dao;
import com.yihu.jw.iot.data_input.DataStandardDO;
import org.springframework.data.jpa.repository.JpaSpecificationExecutor;

svr/svr-iot/src/main/java/com/yihu/iot/data_input/enums/DataDeviceTypeEnum.java

@@ -0,0 +1,58 @@
package com.yihu.iot.data_input.enums;
import org.apache.commons.lang.StringUtils;
/**
 * 设备类型枚举类，有些设备测量血压，有些设备测量血糖等
 */
public enum DataDeviceTypeEnum {
    SYSTOLIC("1","血压"),
    BLOOD_SUGAR("2","血糖");
    private String type;
    private String name;
    private DataDeviceTypeEnum(String type,String name){
        this.type = type;
        this.name = name;
    }
    public String getType() {
        return type;
    }
    public void setType(String type) {
        this.type = type;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    /**
     * 根据传入的设备类型返回对应的测量名称
     * @param type
     * @return
     */
    public static String getNameByType(String type){
        String name = "";
        DataDeviceTypeEnum[] arrays = DataDeviceTypeEnum.values();
        for(int i = 0;i < arrays.length;i++){
            if(StringUtils.equals(type,arrays[i].getType())){
                name = arrays[i].getName();
                return name;
            }
        }
        return name;
    }
    @Override
    public String toString() {
        return this.type.toString();
    }
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/service/DataInputService.java

@@ -0,0 +1,195 @@
package com.yihu.iot.data_input.service;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yihu.base.es.config.ElastricSearchHelper;
import com.yihu.base.es.config.model.SaveModel;
import com.yihu.base.hbase.HBaseHelper;
import com.yihu.iot.data_input.util.RowKeyUtils;
import com.yihu.iot.service.device.IotDeviceService;
import com.yihu.jw.iot.device.IotDeviceDO;
import com.yihu.jw.util.date.DateUtil;
import org.apache.http.client.utils.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.UnsupportedEncodingException;
import java.security.NoSuchAlgorithmException;
import java.text.ParseException;
import java.util.*;
/**
 * 1、设备注册及绑定
 * 2、不含居民身份的数据上传协议
 * 3、含居民身份的数据上传协议
 * 4、数据查询返回的协议格式
 */
@Component
public class DataInputService {
    private Logger logger = LoggerFactory.getLogger(DataInputService.class);
    @Autowired
    private IotDeviceService iotDeviceService;
    @Autowired
    private DataProcessLogService dataProcessLogService;
    @Autowired
    private ElastricSearchHelper elastricSearchHelper;
    @Autowired
    private HBaseHelper hBaseHelper;
    private String esIndex = "body_health_data";
    private String esType = "signs_data";
    private String tableName = "body_health_data";
    private String familyA = "column_signs_header";
    private String familyB = "column_signs_data";
    /**
     * 居民设备注册及绑定
     */
    public String bindUser(String json){
        List<IotDeviceDO> deviceDOList = new ArrayList<>();
        JSONObject jsonObject = JSONObject.parseObject(json);
        String data_source = jsonObject.getString("data_source");
        JSONArray jsonArray = jsonObject.getJSONArray("data");
        try {
            if(null != jsonArray){
                for(Object array:jsonArray){
                    JSONObject dataJson = (JSONObject)JSONObject.toJSON(array);
                    IotDeviceDO iotDeviceDO = new IotDeviceDO();
                    String sn = dataJson.getString("sn");
                    int count = iotDeviceService.countByDeviceSn(sn);
                    if(count > 0){
                        continue; //表示设备已经绑定过
                    }
                    iotDeviceDO.setDeviceSn(sn);
                    iotDeviceDO.setCode(dataJson.getString("ext_code"));
                    iotDeviceDO.setName(dataJson.getString("device_name"));
                    iotDeviceDO.setDeviceModel(dataJson.getString("device_model"));
                    iotDeviceDO.setDeviceSource("2"); //设备来源为居民绑定
                    iotDeviceDO.setCreateUser(dataJson.getString("idcard")); //居民绑定的，暂定创建人和修改人均为居民
                    iotDeviceDO.setCreateUserName(dataJson.getString("username"));
                    iotDeviceDO.setUpdateUser(dataJson.getString("idcard"));
                    iotDeviceDO.setUpdateUserName(dataJson.getString("username"));
                    iotDeviceDO.setManufacturerCode(dataJson.getString("manufacture_code"));
                    iotDeviceDO.setManufacturerName(dataJson.getString("manufacture_name"));
                    iotDeviceDO.setManufactureTel(dataJson.getString("manufacture_tel"));
                    iotDeviceDO.setSupplierCode(dataJson.getString("owner_org_code"));
                    iotDeviceDO.setSupplierName(dataJson.getString("owner_org_name"));
                    iotDeviceDO.setName(dataJson.getString("sale_org_code"));
                    iotDeviceDO.setName(dataJson.getString("sale_org_name"));
                    deviceDOList.add(iotDeviceDO);
                }
                iotDeviceService.bindUser(deviceDOList);
                //保存日志
                dataProcessLogService.saveLog("","",data_source,"", DateUtils.formatDate(new Date(), DateUtil.yyyy_MM_dd_HH_mm_ss),"1","4","com.yihu.iot.data_input.service.DataInputService.bindUser",0);
            }
        }catch (Exception e){
            logger.error("注册绑定失败");
            //保存日志
            dataProcessLogService.saveLog("","",data_source,"", DateUtils.formatDate(new Date(), DateUtil.yyyy_MM_dd_HH_mm_ss),"1","3","com.yihu.iot.data_input.service.DataInputService.bindUser",1);
            return "fail";
        }
        return "success";
    }
    /**
     * 修改设备绑定居民信息
     */
    public void updateBindUser(String data_source,String deviveSn,String idcard,String username){
        IotDeviceDO iotDeviceDO = iotDeviceService.findByDeviceSn(deviveSn);
        if(null != iotDeviceDO){
            iotDeviceDO.setUpdateUser(idcard);
            iotDeviceDO.setUpdateUserName(username);
        }
        iotDeviceService.save(iotDeviceDO);
        //保存日志
        dataProcessLogService.saveLog("","",data_source,"", DateUtils.formatDate(new Date(), DateUtil.yyyy_MM_dd_HH_mm_ss),"1","3","com.yihu.iot.data_input.service.DataInputService.bindUser",1);
    }
    /**
     * 上传数据
     * @param json
     * @return
     */
    public String uploadData(String json){
        String fileName = "";
        String fileAbsPath = "";
        String rowkey = "";
        //提取json各项值
        JSONObject jsonObject = JSONObject.parseObject(json);
        String accessToken= jsonObject.getString("access_token");
        String dataSource = jsonObject.getString("data_source");
        String deviceSn = jsonObject.getString("sn");
        String extCode = jsonObject.getString("ext_code");
        String measuretime = jsonObject.getString("measure_time");
        //包含居民身份的数据，对设备数据进行校验绑定，此处包含的信息只有身份证号和用户名以及设备序列号，如果设备库中存在该序号的设备，则对绑定居民进行修改，改为当前居民，如果没有则跳过
        if(jsonObject.containsKey("idcard") && jsonObject.containsKey("username")){
            String idcard = jsonObject.getString("idcard");
            String username = jsonObject.getString("username");
            updateBindUser(dataSource,deviceSn,idcard,username);
        }
        try {
            rowkey = RowKeyUtils.makeRowKey(accessToken, deviceSn, extCode, DateUtil.dateTimeParse(measuretime).getTime());
        } catch (Exception e) {
            e.printStackTrace();
        }
        //将数据存入es
        jsonObject.put("_id", new SaveModel().getId());//es的id继承至jestId
        jsonObject.put("id", rowkey);//hbase的rowkey
        elastricSearchHelper.save(esIndex, esType, jsonObject.toJSONString());
        Map<String, Map<String, String>> family = new HashMap<>();
        Map<String, String> columnsA = new HashMap<>();
        Map<String, String> columnsB = new HashMap<>();
        //组装A列
        columnsA.put("access_token",accessToken);
        columnsA.put("data_source",dataSource);
        columnsA.put("sn",deviceSn);
        columnsA.put("ext_code",extCode);
        columnsA.put("device_name",jsonObject.getString("device_name"));
        columnsA.put("device_model",jsonObject.getString("device_model"));
        family.put(familyA,columnsA);
        JSONArray jsonArray = jsonObject.getJSONArray("data");
        if(null == jsonArray || jsonArray.size() == 0){
            return "json no data";
        }
        //组装B列
        for(Object obj:jsonArray){
            JSONObject data = (JSONObject)obj;
           for(String key:data.keySet()){
               columnsB.put(key,data.getString(key));
           }
           if(data.containsKey("ecg")){
               fileName = data.getString("fileName");
               fileAbsPath = data.getString("filepath");
           }
        }
        family.put(familyB,columnsB);
        try {
            hBaseHelper.add(tableName,rowkey,family);
        } catch (Exception e) {
            e.printStackTrace();
            //保存日志
            dataProcessLogService.saveLog(fileName,fileAbsPath,dataSource,"", DateUtils.formatDate(new Date(), DateUtil.yyyy_MM_dd_HH_mm_ss),"1","3","com.yihu.iot.data_input.service.DataInputService.uploadData",1);
        }
        //保存日志
        dataProcessLogService.saveLog(fileName,fileAbsPath,dataSource,"", DateUtils.formatDate(new Date(), DateUtil.yyyy_MM_dd_HH_mm_ss),"1","4","com.yihu.iot.data_input.service.DataInputService.uploadData",0);
        return "success";
    }
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/service/DataProcessLogService.java

@@ -0,0 +1,38 @@
package com.yihu.iot.data_input.service;
import com.yihu.base.mysql.query.BaseJpaService;
import com.yihu.iot.data_input.dao.DataProcessLogDao;
import com.yihu.jw.iot.data_input.DataProcessLogDO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
@Service
public class DataProcessLogService extends BaseJpaService<DataProcessLogDO,DataProcessLogDao> {
    @Autowired
    private DataProcessLogDao dataProcessLogDao;
    /**
     * 保存数据上传日志
     */
    public void saveDataLog(List<DataProcessLogDO> list){
        this.dataProcessLogDao.save(list);
    }
    public void saveLog(String fileName,String fileAbsPath,String dataSource,String receiveTime,String uploadTime,String processType,String status,String desc,int failCount){
        DataProcessLogDO dataProcessLog = new DataProcessLogDO();
        dataProcessLog.setFileName(fileName);
        dataProcessLog.setFileName(fileAbsPath);
        dataProcessLog.setFileName(dataSource);
        dataProcessLog.setFileName(receiveTime);
        dataProcessLog.setFileName(uploadTime);
        dataProcessLog.setProcessType(processType);
        dataProcessLog.setProcessStatus(status);
        dataProcessLog.setProcessDes(desc);
        dataProcessLog.setFileCount(failCount);
        dataProcessLogDao.save(dataProcessLog);
    }
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/service/DataStandardConvertService.java

@@ -0,0 +1,109 @@
package com.yihu.iot.data_input.service;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yihu.iot.data_input.enums.DataDeviceTypeEnum;
import com.yihu.jw.iot.data_input.DataStandardDO;
import com.yihu.jw.util.date.DateUtil;
import org.apache.http.client.utils.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.*;
/**
 * 数据标准转换
 */
@Component
public class DataStandardConvertService {
    private Logger logger = LoggerFactory.getLogger(DataStandardConvertService.class);
    public static Map<String,List<Object>> dataMap = new HashMap<>();
    @Autowired
    private DataStandardService dataStandardService;
    @Autowired
    private DataProcessLogService dataProcessLogService;
    /**
     * 从数据库读取标准,并放到map，key为协议大类名称，目前只有体征数据协议
     */
    public void init(){
        List<DataStandardDO>  resultList = dataStandardService.getList();
        if(null == resultList && resultList.size() < 1){
            return;
        }
        List list = new ArrayList();
        for(DataStandardDO dataStandardDO:resultList){
            String baseName = dataStandardDO.getBaseName();
            if(dataMap.containsKey(baseName)){
                list.add(dataStandardDO);
            } else {
                List tempList = new ArrayList();
                list.add(dataStandardDO);
                tempList.addAll(list);
                list.clear();
                dataMap.put(baseName, tempList);
            }
        }
    }
    /**
     * 数据标准转换，i健康数据-->物联网数据标准
     * i健康模板数据
     * 物联网模板数据
     * @param oldJson 要转换的json
     */
    public String iconvert(String oldJson){
        JSONObject jsonObject = JSONObject.parseObject(oldJson);
        //如果没有授权或者数据来源，则表示数据异常
        if(jsonObject.containsKey("access_token") || jsonObject.containsKey("data_source")){
            logger.warn("传过来的数据无有效access_token或data_source",oldJson);
            return "";
        }
        //拿到i健康json数据里的各项值
        Object id = jsonObject.get("id");
        if(null == id){
            id = UUID.randomUUID();
        }
        String access_token = (String)jsonObject.get("access_token");
        String data_source = (String)jsonObject.get("data_source");
        String sn = (String)jsonObject.get("deviceSn");
        String deviceType = (String)jsonObject.get("deviceType");
        String ext_code = (String)jsonObject.get("userType");
        String data = (String)jsonObject.get("userType") + (String)jsonObject.get("unit");
        String device_name = (String)jsonObject.get("device_name");
        String device_model = (String)jsonObject.get("device_model");
        String measure_time = (String)jsonObject.get("sendTime");
        JSONObject newJsonObject = new JSONObject();
        JSONArray jsonArray = new JSONArray();
        JSONObject dataJsonObject = new JSONObject();
        newJsonObject.put("id",id);
        newJsonObject.put("access_token",access_token);
        newJsonObject.put("data_source",data_source);
        newJsonObject.put("sn",sn);
        newJsonObject.put("ext_code",ext_code);
        newJsonObject.put("device_name",device_name);
        newJsonObject.put("device_model",device_model);
        newJsonObject.put("data",jsonArray);
        dataJsonObject.put("measure_time",measure_time);
        String deviceDataName = DataDeviceTypeEnum.getNameByType(deviceType);
        dataJsonObject.put(deviceDataName,data);
        jsonArray.add(dataJsonObject);
        //保存日志
        dataProcessLogService.saveLog("","",data_source,measure_time, DateUtils.formatDate(new Date(),DateUtil.yyyy_MM_dd_HH_mm_ss),"1","4","com.yihu.iot.data_input.service.DataStandardConvertService.iconvert",0);
        //转换后的标准json数据
        return newJsonObject.toJSONString();
    }
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/service/DataStandardService.java

@@ -0,0 +1,23 @@
package com.yihu.iot.data_input.service;
import com.yihu.base.mysql.query.BaseJpaService;
import com.yihu.iot.data_input.dao.DataStandardDao;
import com.yihu.jw.iot.data_input.DataStandardDO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;
@Service
public class DataStandardService extends BaseJpaService<DataStandardDO,DataStandardDao> {
    @Autowired
    private DataStandardDao dataStandardDao;
    List<DataStandardDO> getList(){
        List<DataStandardDO> list = new ArrayList<>();
        list = dataStandardDao.getList();
        return list;
    }
}

svr/svr-iot/src/main/java/com/yihu/iot/data_input/util/RowKeyUtils.java

@@ -0,0 +1,17 @@
package com.yihu.iot.data_input.util;
import org.springframework.util.DigestUtils;
import java.io.UnsupportedEncodingException;
import java.security.NoSuchAlgorithmException;
public class RowKeyUtils {
    public static String makeRowKey(String accessToken,String deviceSn,String extCode,long measureTime) throws NoSuchAlgorithmException, UnsupportedEncodingException {
        StringBuilder sb = new StringBuilder();
        sb.append(accessToken+"_");
        sb.append(deviceSn+"_");
        sb.append(extCode+"_");
        sb.append(measureTime);
        return DigestUtils.md5DigestAsHex(sb.toString().getBytes("utf-8"));
    }
}

svr/svr-iot/src/main/java/com/yihu/iot/service/device/IotDeviceService.java

@@ -6,6 +6,8 @@ import com.yihu.jw.iot.device.IotDeviceDO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
 * @author yeshijie on 2017/12/8.
 */
@@ -18,4 +20,30 @@ public class IotDeviceService extends BaseJpaService<IotDeviceDO,IotDeviceDao> {
    public IotDeviceDO findById(String id) {
        return iotDeviceDao.findById(id);
    }
    /**
     * 设备注册及绑定
     * @param iotDeviceDO
     */
    public void bindUser(List<IotDeviceDO> iotDeviceDO){
        this.batchInsert(iotDeviceDO);
    }
    /**
     * 根据设备序列号判断设备是否存在
     * @param deviceSn
     * @return
     */
    public int countByDeviceSn(String deviceSn) {
        return iotDeviceDao.countByDeviceSn(deviceSn);
    }
    /**
     * 根据设备序列号查找设备
     * @param deviceSn
     * @return
     */
    public IotDeviceDO findByDeviceSn(String deviceSn) {
        return iotDeviceDao.findByDeviceSn(deviceSn);
    }
}

svr/svr-iot/src/main/resources/application.yml

@@ -4,7 +4,7 @@ server:
spring:
  application:
    name:  svr-iot  ##注册到发现服务的id 如果id一样 eurika会自动做负载
    name:  svr-iot  #注册到发现服务的id 如果id一样 eurika会自动做负载
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
@@ -47,33 +47,38 @@ fast-dfs:
#es配置
es:
  host:  59.61.92.90
  port: 9067 #默认是9200
  tPort: 9068 #http端口 默认是9300
  host:  127.0.0.1
  port: 9200 #默认是9200
  tPort: 9300 #http端口 默认是9300
  clusterName: jkzl
spring:
  data:
    elasticsearch: #ElasticsearchProperties
      cluster-name: jkzl #默认即为elasticsearch  集群名
      cluster-nodes: 172.17.110.217:9300,172.17.110.128:9300 #配置es节点信息，逗号分隔，如果没有指定，则启动ClientNode
      local: false ##是否本地连接
      properties: # Additional properties used to configure the client.
  # JEST (Elasticsearch HTTP client) (JestProperties)
  elasticsearch:
    jest:
      uris: http://localhost:9200
      connection-timeout: # Connection timeout in milliseconds.
      multi-threaded: true # Enable connection requests from multiple execution threads.
      username: # Login user.
      password: # Login password.
      proxy.port:  # Proxy port the HTTP client should use.
      proxy.host:  # Proxy host the HTTP client should use.
data:
  elasticsearch: #ElasticsearchProperties
    cluster-name: jkzl #默认即为elasticsearch  集群名
    cluster-nodes: 172.17.110.217:9300,172.17.110.128:9300 #配置es节点信息，逗号分隔，如果没有指定，则启动ClientNode
    local: false #是否本地连接
    properties: # Additional properties used to configure the client.
  # JEST (Elasticsearch HTTP client) (JestProperties)
elasticsearch:
  jest:
    uris: http://localhost:9200
    connection-timeout: # Connection timeout in milliseconds.
    multi-threaded: true # Enable connection requests from multiple execution threads.
    username: # Login user.
    password: # Login password.
    proxy.port:  # Proxy port the HTTP client should use.
    proxy.host:  # Proxy host the HTTP client should use.
hadoop:
  hbase-properties:
    hbase.zookeeper.property.clientPort: 2181
    zookeeper.znode.parent: /hbase-unsecure
    hbase.zookeeper.quorum:  node1.hde.h3c.com,node2.hde.h3c.com,node3.hde.h3c.com
  user:
    name: root
@@ -87,6 +92,7 @@ spring:
    password: 123456
fastDFS:
  fastdfs_file_url: http://172.19.103.54:80/
---
spring:
  profiles: jwtest

svr/svr-iot/src/main/resources/bootstrap.yml

@@ -3,12 +3,12 @@ spring:
  #从发现服务里面取配置服务的信息
  cloud:
    config:
      failFast: true ##启动快速失败 即链接不到配置服务就启动失败
      failFast: true #启动快速失败 即链接不到配置服务就启动失败
      username: jw
      password: jkzl
      discovery:
        enabled: true ##使用发现服务
        service-id: svr-configurations ##配置服务的名字
        enabled: true #使用发现服务
        service-id: svr-configurations #配置服务的名字
---
@@ -20,7 +20,7 @@ eureka:
  client:
    serviceUrl:
      #http://账号:密码@127.0.0.1:8761/eureka/
      defaultZone: http://jw:jkzl@127.0.0.1:8761/eureka/
      defaultZone: http://jw:jkzl@172.19.103.33:8761/eureka/
---
spring:

web-gateway/src/main/java/com/yihu/jw/commnon/iot/DataConstants.java

@@ -0,0 +1,18 @@
package com.yihu.jw.commnon.iot;
public class DataConstants {
    public static final String iot = "iot";
    //数据标准转换
    public static class DataConvert{
        public static final String api_common = iot + "/dataConvert";
        public static final String api_convert = "convert";
    }
    //数据上传常量
    public static class DataInput {
        public static final String api_common = iot + "/dataInput";
        public static final String api_bind_user = "userBind";
        public static final String api_upload_data = "input";
    }
}

web-gateway/src/main/java/com/yihu/jw/controller/iot/data_input/DataInputController.java

@@ -0,0 +1,47 @@
package com.yihu.jw.controller.iot.data_input;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixProperty;
import com.yihu.jw.commnon.iot.DataConstants;
import com.yihu.jw.feign.iot.data_input.DataInputFeign;
import com.yihu.jw.restmodel.common.Envelop;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping(DataConstants.DataInput.api_common)
@Api(value = "数据上传入库", description = "数据上传入库")
public class DataInputController {
    @Autowired
    private DataInputFeign dataInputFeign;
    @PostMapping(value = DataConstants.DataInput.api_bind_user, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    @ApiOperation(value = "设备注册绑定", notes = "设备注册并绑定居民信息")
    @HystrixCommand(commandProperties = {
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "-1"),//超时时间
            @HystrixProperty(name = "execution.timeout.enabled", value = "false") })
    public Envelop bindUser(
            @ApiParam(name = "json_data", value = "", defaultValue = "")
            @RequestBody String jsonData) {
        return dataInputFeign.bindUser(jsonData);
    }
    @PostMapping(value = DataConstants.DataInput.api_upload_data, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    @ApiOperation(value = "数据上传", notes = "数据上传")
    @HystrixCommand(commandProperties = {
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "-1"),//超时时间
            @HystrixProperty(name = "execution.timeout.enabled", value = "false") })
    public Envelop uploadData(
            @ApiParam(name = "json_data", value = "", defaultValue = "")
            @RequestBody String jsonData) {
        return dataInputFeign.uploadData(jsonData);
    }
}

web-gateway/src/main/java/com/yihu/jw/controller/iot/data_input/DataStandardConvertController.java

@@ -0,0 +1,36 @@
package com.yihu.jw.controller.iot.data_input;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixProperty;
import com.yihu.jw.commnon.iot.DataConstants;
import com.yihu.jw.feign.iot.data_input.DataStandardConvertFeign;
import com.yihu.jw.restmodel.common.Envelop;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping(DataConstants.DataConvert.api_common)
@Api(value = "数据标准转换", description = "数据标准转换")
public class DataStandardConvertController {
    @Autowired
    private DataStandardConvertFeign dataStandardConvertFeign;
    @PostMapping(value = DataConstants.DataConvert.api_convert, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    @ApiOperation(value = "数据标准转换", notes = "数据标准转换")
    @HystrixCommand(commandProperties = {
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "-1"),//超时时间
            @HystrixProperty(name = "execution.timeout.enabled", value = "false") })
    public Envelop convert(
            @ApiParam(name = "json_data", value = "", defaultValue = "")
            @RequestBody String jsonData) {
        return dataStandardConvertFeign.convert(jsonData);
    }
}

web-gateway/src/main/java/com/yihu/jw/feign/fallbackfactory/iot/data_input/DataInputFeignFallbackFactory.java

@@ -0,0 +1,34 @@
package com.yihu.jw.feign.fallbackfactory.iot.data_input;
import com.yihu.jw.feign.iot.data_input.DataInputFeign;
import com.yihu.jw.restmodel.common.Envelop;
import feign.hystrix.FallbackFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.web.bind.annotation.RequestBody;
public class DataInputFeignFallbackFactory implements FallbackFactory<DataInputFeign> {
    @Autowired
    private Tracer tracer;
    @Override
    public DataInputFeign create(Throwable e) {
        return new DataInputFeign() {
            @Override
            public Envelop bindUser(@RequestBody String jsonData) {
                tracer.getCurrentSpan().logEvent("设备注册绑定失败:原因:" + e.getMessage());
                tracer.getCurrentSpan().logEvent("jsonData:" + jsonData);
                return null;
            }
            @Override
            public Envelop uploadData(@RequestBody String jsonData) {
                tracer.getCurrentSpan().logEvent("数据上传失败:原因:" + e.getMessage());
                tracer.getCurrentSpan().logEvent("jsonData:" + jsonData);
                return null;
            }
        };
    }
}

web-gateway/src/main/java/com/yihu/jw/feign/fallbackfactory/iot/data_input/DataStandardConvertFeignFallbackFactory.java

@@ -0,0 +1,31 @@
package com.yihu.jw.feign.fallbackfactory.iot.data_input;
import com.yihu.jw.exception.business.JiWeiException;
import com.yihu.jw.feign.iot.data_input.DataInputFeign;
import com.yihu.jw.feign.iot.data_input.DataStandardConvertFeign;
import com.yihu.jw.feign.iot.supplier.IotDeviceSupplierFeign;
import com.yihu.jw.restmodel.common.Envelop;
import feign.hystrix.FallbackFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
public class DataStandardConvertFeignFallbackFactory implements FallbackFactory<DataStandardConvertFeign> {
    @Autowired
    private Tracer tracer;
    @Override
    public DataStandardConvertFeign create(Throwable throwable) {
        return new DataStandardConvertFeign() {
            @Override
            public Envelop convert(String jsonData) {
                tracer.getCurrentSpan().logEvent("标准转换失败:原因:" + throwable.getMessage());
                tracer.getCurrentSpan().logEvent("jsonData:" + jsonData);
                return null;
            }
        };
    }
}

web-gateway/src/main/java/com/yihu/jw/feign/iot/data_input/DataInputFeign.java

@@ -0,0 +1,28 @@
package com.yihu.jw.feign.iot.data_input;
import com.yihu.jw.feign.fallbackfactory.iot.data_input.DataInputFeignFallbackFactory;
import com.yihu.jw.feign.fallbackfactory.iot.supplier.IotDeviceSupplierFeignFallbackFactory;
import com.yihu.jw.restmodel.CommonContants;
import com.yihu.jw.restmodel.common.Envelop;
import com.yihu.jw.rm.iot.DataRequestMapping;
import com.yihu.jw.rm.iot.IotRequestMapping;
import org.springframework.cloud.netflix.feign.FeignClient;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
@FeignClient(
        name = CommonContants.svr_iot // name值是eurika的实例名字
        ,fallbackFactory  = DataInputFeignFallbackFactory.class
)
@RequestMapping(DataRequestMapping.api_iot_common)
public interface DataInputFeign {
    @PostMapping(value = DataRequestMapping.DataInput.api_user_bind, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    Envelop bindUser(@RequestBody String jsonData);
    @PostMapping(value = DataRequestMapping.DataInput.api_data_input, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    Envelop uploadData(@RequestBody String jsonData);
}

web-gateway/src/main/java/com/yihu/jw/feign/iot/data_input/DataStandardConvertFeign.java

@@ -0,0 +1,23 @@
package com.yihu.jw.feign.iot.data_input;
import com.yihu.jw.feign.fallbackfactory.iot.data_input.DataInputFeignFallbackFactory;
import com.yihu.jw.feign.fallbackfactory.iot.data_input.DataStandardConvertFeignFallbackFactory;
import com.yihu.jw.restmodel.CommonContants;
import com.yihu.jw.restmodel.common.Envelop;
import com.yihu.jw.rm.iot.DataRequestMapping;
import org.springframework.cloud.netflix.feign.FeignClient;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
@FeignClient(
        name = CommonContants.svr_iot // name值是eurika的实例名字
        ,fallbackFactory  = DataStandardConvertFeignFallbackFactory.class
)
@RequestMapping(DataRequestMapping.api_iot_common)
public interface DataStandardConvertFeign {
    @PostMapping(value = DataRequestMapping.DataStandardConvert.api_convert, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE)
    Envelop convert(@RequestBody String jsonData);
}