分布式项目（六）iot-device-data 设备数据监控

书接上回，当设备发送数据后，咱们须要在管理页面看到设备发送的数据是什么，因此如今咱们就来完成设备数据监控模块

iot-device-data

建立iot-device-data模块，由于此模块也是属于订阅板块的服务，因此它也是消费kakfa中的Mapping数据，引入对应的redis，kakfa模块。

逻辑图

web socket

数据是实时与页面交互，须要用到web socket

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
            <version>2.1.4.RELEASE</version>
        </dependency>

spring boot 对web socket的使用有两种方式，ServerEndpointExporter和TextWebSocketHandler方式，而ServerEndpointExporter，由于webSocket bean不是由spring 容器管理，因此会有注入没法使用的问题，虽然能够解决，但有点不爽；而TextWebSocketHandler，则彻底由spring实现，因此不会有注入的问题，所以笔者采用TextWebSocketHandler的方式。

web socket handler

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

    @Autowired
    private DeviceDateService deviceDateService;

    /**
     * 打开会话
     * [@param](https://my.oschina.net/u/2303379) session
     */

    [@Override](https://my.oschina.net/u/1162528)
    public void afterConnectionEstablished(WebSocketSession session) 
	throws Exception {
        deviceDateService.onOpen(session);
    }

    /**
     * 关闭会话
     * [@param](https://my.oschina.net/u/2303379) session
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, 
	CloseStatus status) throws Exception {
        deviceDateService.onClose(session);
    }

    /**
     * 异常处理
     * @param session
     * @param
     */
    @Override
    public void handleTransportError(WebSocketSession session, 
	Throwable exception) throws Exception {
        deviceDateService.onError(session,exception);
    }
}

web socket config

@Configuration
@EnableWebSocket //开启web socket
public class WebSocketConfig implements WebSocketConfigurer {

    @Autowired
    private DeviceDateHandler deviceDateHandler;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry 
	registry) {
        registry.addHandler(deviceDateHandler, "/ws/device").
		setAllowedOrigins("*"); //容许跨域
    }
}

DeviceDateService web socket创建登陆

web socket创建登陆时，须要把设备的IMEI和id号一块儿发送过来，咱们在onOpen()方法中对数据进行校验，只有经过校验的连接，才能完成登陆。

@Service
public class DeviceDateService {

    @Autowired
    private BaseRedisUtil baseRedisUtil;

    private static final String IEMI = "imei";
    private static final String DID = "did";

    /**
     * 打开会话
     * @param session
     */
    public void onOpen(WebSocketSession session) {
        Map<String,String> map = getMap(session);
        RedisDeviceVO vo = baseRedisUtil.get(map.get(IEMI));
        if (vo == null){
            throw new RuntimeException("设备未注册");
        }
        if (vo.getId().equals(map.get(DID))){
            throw new RuntimeException("设备号错误");
        }
        WebSocketUtil.put(map.get(DID),session);
    }

    /**8
     * 关闭会话
     * @param session
     */
    public void onClose(WebSocketSession session) throws IOException {
        Map<String,String> map = getMap(session);
        WebSocketUtil.remove(map.get(DID),session);
        session.close();
    }

    /**
     * 异常处理
     * @param session
     * @param error
     */
    public void onError(WebSocketSession session, Throwable error) 
	throws IOException {
        if (error instanceof RuntimeException){
            session.sendMessage(new TextMessage(error.getMessage()));
        }else {
            session.sendMessage(new TextMessage("系统异常"));
        }
        onClose(session);
    }

    private Map<String,String> getMap(WebSocketSession session){
        String query = session.getUri().getQuery();
        if (StringUtils.isEmpty(query)){
            throw new RuntimeException("参数错误");
        }
        String[] param = query.split("&");
        if (param.length != 2){
            throw new RuntimeException("参数错误");
        }
        Map<String,String> map = new HashMap<>();
        map.put(IEMI,param[0]);
        map.put(DID,param[1]);
        return map;
    }
}

web socket 连接缓存

对于已经登陆成功的连接，须要缓存session，而一个设备可能同时有好几个web socket同时监控，因此这里按照一对多的形式保存session。session是不一样的线程同时在建立和销毁，属于并发操做，因此这里使用并发集合进行处理。

public class WebSocketUtil {

    private static final Map<String, CopyOnWriteArrayList<WebSocketSession>>
	SESSION_MAP = new HashMap<>();


    public static void put(String did,WebSocketSession session){
        CopyOnWriteArrayList<WebSocketSession> list = SESSION_MAP.get(did);
        if (list == null){
            list = new CopyOnWriteArrayList();
            SESSION_MAP.put(did,list);
        }
        list.add(session);
    }

    public static void remove(String did,WebSocketSession session){
        CopyOnWriteArrayList<WebSocketSession> list = SESSION_MAP.get(did);
        if (CollectionUtils.isEmpty(list)){
            return;
        }

        int index = -1;
        Iterator<WebSocketSession> it = list.iterator();
        while (it.hasNext()){
            index++;
            WebSocketSession se = it.next();
            if (se == session){
                list.remove(index);
                return;
            }
        }
        if (CollectionUtils.isEmpty(list)){
            SESSION_MAP.remove(did);
        }
    }

    public static boolean isEmpty(){
        return SESSION_MAP.size() > 0 ? false : true;
    }

    public static CopyOnWriteArrayList<WebSocketSession>
	findSocketConnect(String did){
        return SESSION_MAP.get(did);
    }
}

kafka 监听

订阅kakfa的下行数据(mapping 数据)，判断是否有设备连接，若是有就把数据发送给对应的session。

@Component
public class DeviceListener {

    @KafkaListener(topics = DOWN_TOPIC)
    public void listener(String msg){
        System.out.println(msg);
        if (WebSocketUtil.isEmpty()){
            return;
        }
        KafkaDownVO vo = JSONObject.parseObject(msg,KafkaDownVO.class);

        List<WebSocketSession> socket = WebSocketUtil.findSocketConnect(
		String.valueOf(vo.getDeviceId()));
        if (CollectionUtils.isEmpty(socket)){
            return;
        }
        TextMessage textMessage = new TextMessage(msg);
        socket.forEach(session -> {
            try {
                session.sendMessage(textMessage);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }
}

application.properties

#访问端口
server.port=8082
#项目路径
server.servlet.context-path=/device-data
#项目名称
spring.application.name=device-data

spring.kafka.consumer.group-id=websocket-device-data

启动项目，建立web socket连接，验证数据是否推送到了页面。

笔者这里使用的是web socket在线测试工具，http://www.blue-zero.com/WebSocket/

结束语

iot-pt，咱们已经实现了整个数据的上行，从设备的注入，设备发送数据，Mapping设备数据，pgsql持久化数据，web socket推送实时数据，已经有一个完整的流程了，因此接下来就要使用spring cloud搭建分布式了。

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