package com.persagy.iottransfer.server; import cn.hutool.core.thread.ThreadUtil; import com.persagy.iottransfer.communication.entity.PacketEntity; import com.persagy.iottransfer.config.ProjectClent; import com.persagy.iottransfer.kafka.KafkaProducer; import com.persagy.iottransfer.server.IotServer; import lombok.extern.slf4j.Slf4j; /** * @description: AlarmMessage消费线程 * @author:LuoGuangyi * @company:PersagyTechnologyCo.,Ltd * @since:2021/01/29 17:08 * @version:V1.0 **/ @Slf4j public class IotServerReceiveMessageThread implements Runnable { KafkaProducer kafkaProducer; ProjectClent projectClent; PacketEntity tcpControl = null; PacketEntity tcpCollect = null; public IotServerReceiveMessageThread(KafkaProducer kafkaProducer,ProjectClent projectClent) { this.kafkaProducer = kafkaProducer; this.projectClent = projectClent; } @Override public void run() { ThreadUtil.execAsync(()->{ while (true){ try { tcpControl = IotServer.tcpControlServerManager.takeRece(); if (tcpControl != null) { if (tcpControl.address == null) { break; } addProject(tcpControl,"tcpControl"); log.info("收到iot-project 控制消息:{}", tcpControl); kafkaProducer.sendFromedge2control(tcpControl); } } catch (Exception e) { log.error(e.getMessage(),e); } } },true); ThreadUtil.execAsync(()->{ while (true){ try { tcpCollect = IotServer.tcpCollectServerManager.takeRece(); if (tcpCollect != null) { if (tcpCollect.address == null) { break; } addProject(tcpCollect,"tcpCollect"); log.info("收到iot-project 采集消息:{}", tcpCollect); kafkaProducer.sendFromedge2collect(tcpCollect); } } catch (Exception e) { log.error(e.getMessage(),e); } } },true); log.info("----------contextInitialize end--------------"); } //添加项目id private void addProject(PacketEntity packetEntity,String type){ String address = packetEntity.address; packetEntity.setProjectId(projectClent.getProject(address,packetEntity.getContent().packetString)); packetEntity.setType(type); } }