IotServerReceiveMessageThread.java 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475
  1. package com.persagy.iottransfer.server;
  2. import cn.hutool.core.thread.ThreadUtil;
  3. import com.persagy.iottransfer.communication.entity.PacketEntity;
  4. import com.persagy.iottransfer.config.ProjectClent;
  5. import com.persagy.iottransfer.kafka.KafkaProducer;
  6. import com.persagy.iottransfer.server.IotServer;
  7. import lombok.extern.slf4j.Slf4j;
  8. /**
  9. * @description: AlarmMessage消费线程
  10. * @author:LuoGuangyi
  11. * @company:PersagyTechnologyCo.,Ltd
  12. * @since:2021/01/29 17:08
  13. * @version:V1.0
  14. **/
  15. @Slf4j
  16. public class IotServerReceiveMessageThread implements Runnable {
  17. KafkaProducer kafkaProducer;
  18. ProjectClent projectClent;
  19. PacketEntity tcpControl = null;
  20. PacketEntity tcpCollect = null;
  21. public IotServerReceiveMessageThread(KafkaProducer kafkaProducer,ProjectClent projectClent) {
  22. this.kafkaProducer = kafkaProducer;
  23. this.projectClent = projectClent;
  24. }
  25. @Override
  26. public void run() {
  27. ThreadUtil.execAsync(()->{
  28. while (true){
  29. try {
  30. tcpControl = IotServer.tcpControlServerManager.takeRece();
  31. if (tcpControl != null) {
  32. if (tcpControl.address == null) {
  33. break;
  34. }
  35. addProject(tcpControl,"tcpControl");
  36. log.info("收到iot-project 控制消息:{}", tcpControl);
  37. kafkaProducer.sendFromedge2control(tcpControl);
  38. }
  39. } catch (Exception e) {
  40. log.error(e.getMessage(),e);
  41. }
  42. }
  43. },true);
  44. ThreadUtil.execAsync(()->{
  45. while (true){
  46. try {
  47. tcpCollect = IotServer.tcpCollectServerManager.takeRece();
  48. if (tcpCollect != null) {
  49. if (tcpCollect.address == null) {
  50. break;
  51. }
  52. addProject(tcpCollect,"tcpCollect");
  53. log.info("收到iot-project 采集消息:{}", tcpCollect);
  54. kafkaProducer.sendFromedge2collect(tcpCollect);
  55. }
  56. } catch (Exception e) {
  57. log.error(e.getMessage(),e);
  58. }
  59. }
  60. },true);
  61. log.info("----------contextInitialize end--------------");
  62. }
  63. //添加项目id
  64. private void addProject(PacketEntity packetEntity,String type){
  65. String address = packetEntity.address;
  66. packetEntity.setProjectId(projectClent.getProject(address,packetEntity.getContent().packetString));
  67. packetEntity.setType(type);
  68. }
  69. }