分布式项目（四）Mapping Server 数据映射

上回说道CoAp client和server的实现，数据也按照既定格式发送到了kafka中，接下来就是Mapping server的实现，物理设备数据映射到抽象设备上，并赋予数据业务含义。

iot-mapping

构建iot-mapping模块，引入kafka公共模块

SourceListener

SourceListener监听Coap server 发送的原始数据，并从redis中取出web manage缓存的产品物模型和设备数据，因为redis公用的比较多，因此这也构建一个iot-redis模块，用与redis操做。

逻辑说明

• 获取kafka中的数据转成KafkaSourceVO
• 根据imei获取redis中的RedisDeviceVO
• 根据productId获取RedisProductVO
• 根据RedisProductVO中的format，肯定是byte处理仍是json处理

@Autowired
    private BaseRedisUtil redisUtil;
    @KafkaListener(topics = SOURCE_TOPIC)
    public void iotListener(String msg){
        System.out.println("-----------"+msg);
        KafkaSourceVO sourceVO = JSONObject.parseObject(msg,
		KafkaSourceVO.class);
        //设备信息
        RedisDeviceVO deviceVO = redisUtil.get(sourceVO.getImei());
        //产品信息
        RedisProductVO productVO = redisUtil.get(deviceVO.getProductId());
        if (EDataFormat.BYTE.getFormat().equals(productVO.getFormat())){
            analysisByte(sourceVO,productVO,deviceVO);
        }else if (EDataFormat.JSON.getFormat().equals(
		productVO.getFormat())){
            analysisJson(sourceVO,productVO,deviceVO);
        }
    }

byte 解析

• 获取KafkaSourceVO中的原始数据data，并转成char[]
• 获取产品物模型中的属性模型RedisPropertyVO
• 获取属性模型，按照属性模型中定义的ofset，在原始数据char[]中获取对应的值
• 把解析出来的数据封装为KafkaDownVO

public void analysisByte(KafkaSourceVO sourceVO,
  RedisProductVO productVO,RedisDeviceVO deviceVO){
        char[] chars = sourceVO.getData().toCharArray();
        List<RedisPropertyVO> propertys = productVO.getPropertys();

        KafkaDownVO downVO = new KafkaDownVO();
        downVO.setDeviceId(deviceVO.getId());
        downVO.setCollTime(sourceVO.getCollTime());

        List<KafkaDownVO.PropertyData> propertyDatas = 
		new ArrayList<>(propertys.size());

        propertys.forEach(property->{
            String[] str = property.getOfset().split("-");
            int begin = Integer.valueOf(str[0]);
            int end = Integer.valueOf(str[1]);

            KafkaDownVO.PropertyData data = new KafkaDownVO.PropertyData();
            data.setPropertyId(property.getId());
            StringBuilder sb = new StringBuilder();
            for (int i = begin;i <= end; i++){
                sb.append(chars[i]);
            }
            data.setData(sb.toString());
            propertyDatas.add(data);
        });
        downVO.setDataList(propertyDatas);
        System.out.println("byte---"+downVO);
    }

json 解析

• json定义中，物理设备发送的key与属性模型中的identifier一一对应，而在封装RedisPropertyVO的时候，为了与byte保持统一把identifier赋值给了ofset，因此这里获取属性模型，并转成Map<ofset,id> 格式。
• KafkaSourceVO中的原始数据也序列化成map<key,value>
• 便利属性模型对应的map(propertyMap)，并从原始数据map(dataMap)取出对应的数据
• 为了数据的统一格式，这里也把数据封装成KafkaDownVO

public void analysisJson(KafkaSourceVO sourceVO,
RedisProductVO productVO,RedisDeviceVO deviceVO){
        Map<String,Long> propertyMap = productVO.getPropertys().
		stream().collect(Collectors.toMap(RedisPropertyVO ::
		getOfset,RedisPropertyVO::getId));
        Map<String,String> dataMap = JSONObject.parseObject(
		sourceVO.getData(), HashMap.class);

        KafkaDownVO downVO = new KafkaDownVO();
        downVO.setDeviceId(deviceVO.getId());
        downVO.setCollTime(sourceVO.getCollTime());

        List<KafkaDownVO.PropertyData> propertyDatas = 
		new ArrayList<>(dataMap.size());

        dataMap.forEach((key,val)->{
            KafkaDownVO.PropertyData data = new KafkaDownVO.PropertyData();
            data.setPropertyId(propertyMap.get(key));
            data.setData(val);
            propertyDatas.add(data);
        });
        downVO.setDataList(propertyDatas);
        System.out.println("json---"+downVO);
    }

启动项目，检验一下数据是否封装正确

按iot-pt架构设计，如今须要把映射好的数据，再次写入kakfa中，供订阅费服务使用

kafka 写入

在iot-kafka模块中添加对Mapping 数据的写入

[@Component](https://my.oschina.net/u/3907912)
public class KafkaMapping {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void send(KafkaDownVO downVO){
        String json = JSONObject.toJSONString(downVO);
        kafkaTemplate.send(DOWN_TOPIC,json);
    }

}

修改analysisByte()和analysisJson()

public void analysisByte(KafkaSourceVO sourceVO,
  RedisProductVO productVO,RedisDeviceVO deviceVO){
        char[] chars = sourceVO.getData().toCharArray();
        List<RedisPropertyVO> propertys = productVO.getPropertys();

        KafkaDownVO downVO = new KafkaDownVO();
        downVO.setDeviceId(deviceVO.getId());
        downVO.setCollTime(sourceVO.getCollTime());

        List<KafkaDownVO.PropertyData> propertyDatas = 
		new ArrayList<>(propertys.size());

        propertys.forEach(property->{
            String[] str = property.getOfset().split("-");
            int begin = Integer.valueOf(str[0]);
            int end = Integer.valueOf(str[1]);

            KafkaDownVO.PropertyData data = new KafkaDownVO.PropertyData();
            data.setPropertyId(property.getId());
            StringBuilder sb = new StringBuilder();
            for (int i = begin;i <= end; i++){
                sb.append(chars[i]);
            }
            data.setData(sb.toString());
            propertyDatas.add(data);
        });
        downVO.setDataList(propertyDatas);
       kafkaMapping.send(downVO);
    }

public void analysisJson(KafkaSourceVO sourceVO,
RedisProductVO productVO,RedisDeviceVO deviceVO){
        Map<String,Long> propertyMap = productVO.getPropertys().
		stream().collect(Collectors.toMap(RedisPropertyVO ::
		getOfset,RedisPropertyVO::getId));
        Map<String,String> dataMap = JSONObject.parseObject(
		sourceVO.getData(), HashMap.class);

        KafkaDownVO downVO = new KafkaDownVO();
        downVO.setDeviceId(deviceVO.getId());
        downVO.setCollTime(sourceVO.getCollTime());

        List<KafkaDownVO.PropertyData> propertyDatas = 
		new ArrayList<>(dataMap.size());

        dataMap.forEach((key,val)->{
            KafkaDownVO.PropertyData data = new KafkaDownVO.PropertyData();
            data.setPropertyId(propertyMap.get(key));
            data.setData(val);
            propertyDatas.add(data);
        });
        downVO.setDataList(propertyDatas);
        kafkaMapping.send(downVO);
    }

再次启动项目，教研Mapping数据是否成写入kakfa

结束语

接下来就是订阅服务的实现了，请听下回分解，具体的代码细节在git

https://gitee.com/distant/iot-pt.git