耳标平台

huimv-eartag2-eartag/src/main/java/com/huimv/eartag2/eartag/listener/EartagListener.java

@@ -3,7 +3,6 @@ package com.huimv.eartag2.eartag.listener;
 import cn.hutool.core.util.ObjectUtil;
 import com.alibaba.fastjson.JSONObject;
 import com.huimv.eartag2.common.mq.Const;
-import com.huimv.eartag2.common.utils.DateUtil;
 import com.huimv.eartag2.common.utils.MathUtil;
 import com.huimv.eartag2.eartag.service.IDeviceService;
 import com.huimv.eartag2.eartag.service.IEartagService;
@@ -157,72 +156,75 @@ public class EartagListener {
 //    @RabbitListener(queues = Const.QUEUE_ASK_EARTAG )
     @RabbitListener(queues = "#{'direct.askEartag.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
-    public void processRawdata(Map RawMap) throws ParseException, IOException {
-         log.info("开始处理原始耳标数据------>"+RawMap);
-        String askText = RawMap.get("askText").toString();
-
-        //{处理请求报文}
-        Map askMap = handleAskText(askText);
-        if (askMap == null){
-            return;
-        }
-        String type = askMap.get("type").toString();
-        JSONObject dataJo = (JSONObject) askMap.get("data");
-        if (type.trim().equalsIgnoreCase("eartag")) {
-            String deviceCode = dataJo.getString("device");
-            String farmId = deviceService.getFarmIdByDeviceCode(deviceCode);
-            System.out.println("####### farmId=" + farmId);
-            if (ObjectUtil.isEmpty(farmId)){
-                log.error("出错:该请求的FarmId为空.请检查该设备编码[" + deviceCode + "]是否存在FarmId.");
-                return;
-            }
-            dataJo.put("farmId",farmId);
-
-            //计算小时运动量
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_COUNT_ACT_HOUR,Const.ROUTING_KEY_COUNT_ACT_HOUR,dataJo);
-
-            //保存耳标流水数据
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_FLOW,Const.ROUTING_KEY_EARTAG_FLOW,dataJo);
-
-            //更新设备注册信息
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_DEVICE_REGIST,Const.ROUTING_KEY_DEVICE_REGIST,dataJo);
-
-            //更新耳标注册信息
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_REGIST,Const.ROUTING_KEY_EARTAG_REGIST,dataJo);
-
-            //更新设备在线统计
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_DEVICE_COUNT,Const.ROUTING_KEY_DEVICE_COUNT,dataJo);
-
-            //更新耳标在线统计
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_COUNT,Const.ROUTING_KEY_EARTAG_COUNT,dataJo);
-
-            //更新耳标在线状态
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_LIVE,Const.ROUTING_KEY_EARTAG_LIVE,dataJo);
-
-            //更新耳标设备在线关联统计
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_DEVICE_ONLINE,Const.ROUTING_KEY_EARTAG_DEVICE_ONLINE,dataJo);
-
-            //异常统计和异常报警
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_ABNORMAL,Const.ROUTING_KEY_ABNORMAL,dataJo);
-
-            //判断该设备编号是否存在牧场缓存在线集合当中
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_AllSTATUS,Const.ROUTING_KEY_AllSTATUS,dataJo);
+    public void processRawdata(JSONObject dataJo) throws ParseException, IOException {
+         log.info("开始处理原始耳标数据------>"+dataJo);
+//        String askText = RawMap.get("askText").toString();
+//
+//        //{处理请求报文}
+//        Map askMap = handleAskText(askText);
+//        if (askMap == null){
+//            return;
+//        }
+//        String type = askMap.get("type").toString();
+//        JSONObject dataJo = (JSONObject) askMap.get("data");
+////        if (type.trim().equalsIgnoreCase("eartag")) {
+//            String deviceCode = dataJo.getString("device");
+//            String farmId = deviceService.getFarmIdByDeviceCode(deviceCode);
+//            System.out.println("####### farmId=" + farmId);
+//            if (ObjectUtil.isEmpty(farmId)){
+//                log.error("出错:该请求的FarmId为空.请检查该设备编码[" + deviceCode + "]是否存在FarmId.");
+//                return;
+//            }
+//            dataJo.put("farmId",farmId);
+
+//            //计算小时运动量
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_COUNT_ACT_HOUR,Const.ROUTING_KEY_COUNT_ACT_HOUR,dataJo);
+//
+//            //保存耳标流水数据
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_FLOW,Const.ROUTING_KEY_EARTAG_FLOW,dataJo);
+//
+//            //更新设备注册信息
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_DEVICE_REGIST,Const.ROUTING_KEY_DEVICE_REGIST,dataJo);
+//
+//            //更新耳标注册信息
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_REGIST,Const.ROUTING_KEY_EARTAG_REGIST,dataJo);
+//
+//            //更新设备在线统计
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_DEVICE_COUNT,Const.ROUTING_KEY_DEVICE_COUNT,dataJo);
+//
+//            //更新耳标在线统计
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_COUNT,Const.ROUTING_KEY_EARTAG_COUNT,dataJo);
+//
+//            //更新耳标在线状态
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_LIVE,Const.ROUTING_KEY_EARTAG_LIVE,dataJo);
+//
+//            //更新耳标设备在线关联统计
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_EARTAG_DEVICE_ONLINE,Const.ROUTING_KEY_EARTAG_DEVICE_ONLINE,dataJo);
+//
+//            //异常统计和异常报警
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_ABNORMAL,Const.ROUTING_KEY_ABNORMAL,dataJo);
+//
+//            //判断该设备编号是否存在牧场缓存在线集合当中
+//            rabbitTemplate.convertAndSend(Const.EXCHANGE_AllSTATUS,Const.ROUTING_KEY_AllSTATUS,dataJo);
 
 //            eartagService.handleEartag(dataJo);
-        }
+//        }
+        eartagService.handleEartag(dataJo);
 
     }
 
 
 
-    @RabbitListener(queues = Const.QUEUE_COUNT_ACT_HOUR)
+    @RabbitListener(queues = "#{'direct.countActHour.queue' + ${rabbitmq.queues}}")
+//    @RabbitListener(queues = Const.QUEUE_COUNT_ACT_HOUR)
     @RabbitHandler
     public void countActHour(JSONObject dataJo) {
         log.info("开始计算小时运动量------>"+dataJo);
         eartagService.countHourAct(dataJo);
     }
 
-    @RabbitListener(queues = Const.QUEUE_EARTAG_FLOW)
+//    @RabbitListener(queues = Const.QUEUE_EARTAG_FLOW)
+    @RabbitListener(queues = "#{'direct.eartagFlow.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void saveEartagFlow(JSONObject dataJo) {
         log.info("开始保存耳标流水数据------>"+dataJo);
@@ -230,14 +232,16 @@ public class EartagListener {
 
     }
 
-    @RabbitListener(queues = Const.QUEUE_DEVICE_REGIST)
+//    @RabbitListener(queues = Const.QUEUE_DEVICE_REGIST)
+    @RabbitListener(queues = "#{'direct.deviceRegist.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void updateDeviceRegister(JSONObject dataJo) {
         log.info("开始更新设备注册信息------>"+dataJo);
         eartagService.updateDeviceRegister(dataJo);
     }
 
-    @RabbitListener(queues = Const.QUEUE_EARTAG_REGIST)
+//    @RabbitListener(queues = Const.QUEUE_EARTAG_REGIST)
+    @RabbitListener(queues = "#{'direct.eartagRegist.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void updateEartagRegister(JSONObject dataJo) {
         log.info("开始更新耳标注册信息------>"+dataJo);
@@ -245,7 +249,8 @@ public class EartagListener {
 
     }
 
-    @RabbitListener(queues = Const.QUEUE_DEVICE_COUNT)
+//    @RabbitListener(queues = Const.QUEUE_DEVICE_COUNT)
+    @RabbitListener(queues = "#{'direct.deviceCount.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void updateDeviceOnline(JSONObject dataJo) throws ParseException {
         log.info("开始更新设备在线统计------>"+dataJo);
@@ -253,7 +258,8 @@ public class EartagListener {
 
     }
 
-    @RabbitListener(queues = Const.QUEUE_EARTAG_COUNT)
+//    @RabbitListener(queues = Const.QUEUE_EARTAG_COUNT)
+    @RabbitListener(queues = "#{'direct.eartagCount.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void updateEartagOnline(JSONObject dataJo){
         log.info("开始更新耳标在线统计------>"+dataJo);
@@ -261,7 +267,8 @@ public class EartagListener {
 
     }
 
-    @RabbitListener(queues = Const.QUEUE_EARTAG_DEVICE_ONLINE)
+//    @RabbitListener(queues = Const.QUEUE_EARTAG_DEVICE_ONLINE)
+    @RabbitListener(queues = "#{'direct.updateEartagDeviceOnlineCount.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void updateEartagDeviceOnlineCount(JSONObject dataJo)   {
         log.info("更新耳标设备在线关联统计------>"+dataJo);
@@ -270,7 +277,8 @@ public class EartagListener {
     }
 
 
-    @RabbitListener(queues = Const.QUEUE_EARTAG_LIVE)
+//    @RabbitListener(queues = Const.QUEUE_EARTAG_LIVE)
+    @RabbitListener(queues = "#{'direct.eartagLiveStatus.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void updateEartagLiveStatus(JSONObject dataJo)   {
         log.info("更新耳标连线状态------>"+dataJo);
@@ -280,7 +288,8 @@ public class EartagListener {
 
 
 
-    @RabbitListener(queues = Const.QUEUE_ABNORMAL)
+//    @RabbitListener(queues = Const.QUEUE_ABNORMAL)
+    @RabbitListener(queues = "#{'direct.abnormalCountAndAlarm.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void abnormalCountAndAlarm(JSONObject dataJo) throws ParseException {
         log.info("异常统计和异常报警------>"+dataJo);
@@ -288,7 +297,8 @@ public class EartagListener {
 
     }
 
-    @RabbitListener(queues = Const.QUEUE_AllSTATUS)
+//    @RabbitListener(queues = Const.QUEUE_AllSTATUS)
+    @RabbitListener(queues = "#{'direct.updateEartagPropertyOfAllStatus.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void updateEartagPropertyOfAllStatus(JSONObject dataJo) throws ParseException {
         log.info("判断该设备编号是否存在牧场缓存在线集合当中------>"+dataJo);

huimv-eartag2-input/src/main/java/com/huimv/eartag2/config/RabbitMQInitializer.java

@@ -27,7 +27,7 @@ public class RabbitMQInitializer implements ApplicationRunner {
     private Integer queues;
 
     @Override  
-    public void run(ApplicationArguments args) throws Exception {
+    public void run(ApplicationArguments args)   {
         // 声明交换机
         // 创建交换机
         rabbitMQService.createExchange(Const.EXCHANGE_ASK_EARTAG);
@@ -37,8 +37,9 @@ public class RabbitMQInitializer implements ApplicationRunner {
             rabbitMQService.createQueue(Const.QUEUE_ASK_EARTAG+integer);
             // 创建绑定
             rabbitMQService.createBinding(Const.QUEUE_ASK_EARTAG+integer, Const.EXCHANGE_ASK_EARTAG,Const.ROUTING_KEY_ASK_EARTAG+integer  );
-
         }
+    }
+
+
 
-    }  
 }

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/AbnormalCountAndAlarmRabbitMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class AbnormalCountAndAlarmRabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_ABNORMAL);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_ABNORMAL+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_ABNORMAL+integer, Const.EXCHANGE_ABNORMAL,Const.ROUTING_KEY_ABNORMAL+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/CountActHourRabbitMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class CountActHourRabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_COUNT_ACT_HOUR);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_COUNT_ACT_HOUR+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_COUNT_ACT_HOUR+integer, Const.EXCHANGE_COUNT_ACT_HOUR,Const.ROUTING_KEY_COUNT_ACT_HOUR+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/DeviceCountRabbitMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class DeviceCountRabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_ASK_EARTAG);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_ASK_EARTAG+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_ASK_EARTAG+integer, Const.EXCHANGE_ASK_EARTAG,Const.ROUTING_KEY_ASK_EARTAG+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/DeviceRegistMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class DeviceRegistMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_EARTAG_COUNT);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_EARTAG_COUNT+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_EARTAG_COUNT+integer, Const.EXCHANGE_EARTAG_COUNT,Const.ROUTING_KEY_EARTAG_COUNT+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/EartagCountRabbitMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class EartagCountRabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_ASK_EARTAG);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_ASK_EARTAG+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_ASK_EARTAG+integer, Const.EXCHANGE_ASK_EARTAG,Const.ROUTING_KEY_ASK_EARTAG+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/EartagLiveStatusRabbitMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class EartagLiveStatusRabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_EARTAG_LIVE);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_EARTAG_LIVE+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_EARTAG_LIVE+integer, Const.EXCHANGE_EARTAG_LIVE,Const.ROUTING_KEY_EARTAG_LIVE+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/EartagRegistRabbitMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class EartagRegistRabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_SYNC_EARTAG_REGISTER);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_SYNC_EARTAG_REGISTER+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_SYNC_EARTAG_REGISTER+integer, Const.EXCHANGE_SYNC_EARTAG_REGISTER,Const.ROUTING_KEY_SYNC_EARTAG_REGISTER+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/UpdateEartagDeviceOnlineCountRabbitMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class UpdateEartagDeviceOnlineCountRabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_EARTAG_DEVICE_ONLINE);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_EARTAG_DEVICE_ONLINE+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_EARTAG_DEVICE_ONLINE+integer, Const.EXCHANGE_EARTAG_DEVICE_ONLINE,Const.ROUTING_KEY_EARTAG_DEVICE_ONLINE+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/controller/UpdateEartagPropertyOfAllStatusRabbitMQInitializer.java

@@ -0,0 +1,36 @@
+package com.huimv.eartag2.controller;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+@Component  
+public class UpdateEartagPropertyOfAllStatusRabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args)   {
+        // 声明交换机
+        // 创建交换机
+        rabbitMQService.createExchange(Const.EXCHANGE_AllSTATUS);
+        //
+        for (Integer integer = 0; integer < queues; integer++) {
+            // 声明队列
+            rabbitMQService.createQueue(Const.QUEUE_AllSTATUS+integer);
+            // 创建绑定
+            rabbitMQService.createBinding(Const.QUEUE_AllSTATUS+integer, Const.EXCHANGE_AllSTATUS,Const.ROUTING_KEY_AllSTATUS+integer  );
+        }
+    }
+
+
+
+}

huimv-eartag2-input/src/main/java/com/huimv/eartag2/service/ICacheService.java

@@ -1,5 +1,7 @@
 package com.huimv.eartag2.service;
 
+import com.huimv.eartag2.common.dao.entity.EartagDeviceRegisterEntity;
+
 public interface ICacheService {
     //{}
     Object getDeviceCodeByChipId(String chipId);
@@ -10,4 +12,8 @@ public interface ICacheService {
     boolean checkEffectiveDeviceSet(String idCode);
 
     long putEffectiveDeviceSet(String idCode);
+
+    Object getFarmIdByDeviceCode(String deviceCode);
+
+    void putDeviceRegister(String deviceCode, EartagDeviceRegisterEntity deviceRegisterEntity);
 }

huimv-eartag2-input/src/main/java/com/huimv/eartag2/service/IDeviceService.java

@@ -19,4 +19,6 @@ public interface IDeviceService {
 
     //{根据芯片id读取设备编码}
     String getDeviceCodeByChipId(String chipId);
+
+    String getFarmIdByDeviceCode(String deviceCode);
 }

huimv-eartag2-input/src/main/java/com/huimv/eartag2/service/impl/CacheServiceImpl.java

@@ -1,11 +1,14 @@
 package com.huimv.eartag2.service.impl;
 
+import com.huimv.eartag2.common.dao.entity.EartagDeviceRegisterEntity;
 import com.huimv.eartag2.common.utils.BizConst;
 import com.huimv.eartag2.service.ICacheService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -44,4 +47,39 @@ public class CacheServiceImpl implements ICacheService {
         String setKey = BizConst.SET_ALL_DEVICE_PREFIX;
         return redisTemplate.opsForSet().add(setKey,idCode);
     }
+
+    @Override
+    public Object getFarmIdByDeviceCode(String deviceCode) {
+        Object farmIdObj = redisTemplate.opsForHash().get(BizConst.DEVICE_REGISTER_PREFIX + deviceCode,"farmId");
+        return farmIdObj;
+    }
+    @Override
+    public void putDeviceRegister(String deviceCode, EartagDeviceRegisterEntity deviceRegisterEntity) {
+        //--更新redis中注册表数据
+        String deviceRegHashKey = BizConst.DEVICE_REGISTER_PREFIX + deviceCode;
+        //将Entity转为Map
+        Map registerMap = RegisterEntityToMap(deviceRegisterEntity);
+        redisTemplate.opsForHash().putAll(deviceRegHashKey, registerMap);
+    }
+    private Map RegisterEntityToMap(EartagDeviceRegisterEntity deviceRegisterEntity) {
+        Map map = new HashMap();
+        map.put("Id", deviceRegisterEntity.getId());
+        map.put("deviceCode", deviceRegisterEntity.getDeviceCode());
+        map.put("farmId", deviceRegisterEntity.getFarmId());
+        map.put("farmName", deviceRegisterEntity.getFarmName());
+        map.put("typeF", deviceRegisterEntity.getTypeF());
+        map.put("stage", deviceRegisterEntity.getStage());
+        map.put("pigpenId", deviceRegisterEntity.getPigpenId());
+        map.put("unitId", deviceRegisterEntity.getUnitId());
+        map.put("registerTime", deviceRegisterEntity.getRegisterTime());
+        map.put("lastTime", deviceRegisterEntity.getLastTime());
+        map.put("activeStatus", deviceRegisterEntity.getActiveStatus());
+        map.put("activeTime", deviceRegisterEntity.getActiveTime());
+        map.put("deviceStatus", deviceRegisterEntity.getDeviceStatus());
+        map.put("deviceAlias", deviceRegisterEntity.getDeviceAlias());
+        map.put("location", deviceRegisterEntity.getLocation());
+        map.put("remark", deviceRegisterEntity.getRemark());
+        return map;
+    }
+
 }

huimv-eartag2-input/src/main/java/com/huimv/eartag2/service/impl/DataServiceImpl.java

@@ -1,5 +1,6 @@
 package com.huimv.eartag2.service.impl;
 
+import cn.hutool.core.util.ObjectUtil;
 import com.alibaba.fastjson.JSONObject;
 
 import com.huimv.eartag2.common.dao.entity.*;
@@ -7,6 +8,7 @@ import com.huimv.eartag2.common.dao.repo.*;
 import com.huimv.eartag2.common.mq.Const;
 import com.huimv.eartag2.common.utils.BizConst;
 import com.huimv.eartag2.common.utils.FileUtil;
+import com.huimv.eartag2.common.utils.MathUtil;
 import com.huimv.eartag2.producer.Producer;
 //import com.huimv.eartag2.server.MsgSendConfirmCallBack;
 import com.huimv.eartag2.service.ICacheService;
@@ -66,12 +68,6 @@ public class DataServiceImpl implements IDataService {
     private IEartagService eartagService;
     @Autowired
     private Producer producer;
-    @Autowired
-    private RegexUtil regexUtil;
-    @Autowired
-    private ICacheService cacheService;
-    @Autowired
-    private EartagDeviceRegisterRepo eartagDeviceRegisterRepo;
 
 
     //处理接收数据
@@ -297,22 +293,132 @@ public class DataServiceImpl implements IDataService {
 
     @Override
     public void sendEartagMQ(Map map,String eartag) throws IOException, ParseException {
-//        try {
-//            rabbitTemplate.setConfirmCallback(new MsgSendConfirmCallBack());
-//
-//            CorrelationData correlationData = new CorrelationData();
-//            log.info("开始发送消息");
-//            correlationData.setId("100");
-            //发送到耳标处理消息队列
-            //zj+330102002000003+13320241020004+87+0+10+154894+80+20240126174333+1.2#3.0#65535#10794+0+end
         int i = Math.abs(eartag.hashCode()) % queues;
-        log.info("routinKey-->"+Const.ROUTING_KEY_ASK_EARTAG+i);
-        rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_ASK_EARTAG+i, map);
-//        } catch (AmqpTimeoutException e) {
-//            log.error("met timeout exception: ", e);
-//        }
+
+
+        String askText = map.get("askText").toString();
+
+        //{处理请求报文}
+        Map askMap = handleAskText(askText);
+        if (askMap == null){
+            return;
+        }
+        String type = askMap.get("type").toString();
+        JSONObject dataJo = (JSONObject) askMap.get("data");
+        if (type.trim().equalsIgnoreCase("eartag")) {
+            String deviceCode = dataJo.getString("device");
+            String farmId = deviceService.getFarmIdByDeviceCode(deviceCode);
+            System.out.println("####### farmId=" + farmId);
+            if (ObjectUtil.isEmpty(farmId)) {
+                log.error("出错:该请求的FarmId为空.请检查该设备编码[" + deviceCode + "]是否存在FarmId.");
+                return;
+            }
+            dataJo.put("farmId", farmId);
+
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_ASK_EARTAG+i, dataJo);
+
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_COUNT_ACT_HOUR+i, dataJo);
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_DEVICE_REGIST+i, dataJo);
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_EARTAG_REGIST+i, dataJo);
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_DEVICE_COUNT+i, dataJo);
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_EARTAG_COUNT+i, dataJo);
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_EARTAG_LIVE+i, dataJo);
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_EARTAG_DEVICE_ONLINE+i, dataJo);
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_ABNORMAL+i, dataJo);
+            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_AllSTATUS+i, dataJo);
+        }
+        log.error("耳标数据有误------>"+dataJo);
+
     }
 
+        private Map handleAskText(String askText) {
+            Map outMap = new HashMap();
+            JSONObject dataJo = new JSONObject();
+            String type = "";
+            String[] dataArray = askText.split("\\+");
+            String cmdHeader = dataArray[0];
+            if (cmdHeader.trim().equalsIgnoreCase("hm")) {
+                //采集器原始数据
+                String device = dataArray[1];
+                String cmd = dataArray[2];
+                if (cmd.trim().equalsIgnoreCase("0")) {
+                    //不需要处理
+                } else if (cmd.trim().equalsIgnoreCase("1")) {
+                    //获取设备编码
+                } else if (cmd.trim().equalsIgnoreCase("2")) {
+                    //获取服务器时间命令
+                } else if (cmd.trim().equalsIgnoreCase("3")) {
+                    //心跳包命令
+                    dataJo.put("device", device);
+                    type = "device-heartbeat";
+                } else if (cmd.trim().equalsIgnoreCase("4")) {
+                    //环境温度-设备、数值、上传时间
+                    dataJo.put("device", dataArray[1]);
+                    dataJo.put("temp", dataArray[4]);
+                    dataJo.put("askTime", dataArray[5]);
+                    type = "device-temp";
+                } else {
+                    log.error("##当前数据为非法数据>>" + askText);
+                }
+            } else if (cmdHeader.trim().equalsIgnoreCase("zj")) {
+//            String device = dataArray[1];
+                //-- 耳标数据 --//
+                //计算耳标温度
+                Double earTemp = 0D;
+                Double earTemp2 = 0D;
+                String other = dataArray[9];
+                if (other != null && other.trim().length() > 0) {
+                    String[] otherArray = other.split("#");
+                    String resi = otherArray[2];
+                    if (otherArray.length>3){
+                        String resi2 = otherArray[3];
+                        if (null != resi2 && !"0".equals(resi2)){
+                            //耳面温度
+                            Double temp2 = MathUtil.ln( Integer.parseInt(resi2));
+                            earTemp2 = new BigDecimal(temp2).setScale(1, BigDecimal.ROUND_HALF_UP).doubleValue();
+                        }
+                    }
+
+                    int resiInt = Integer.parseInt(resi);
+                    if (resiInt != 0) {
+                        Double temp = MathUtil.ln(resiInt);
+                        earTemp = new BigDecimal(temp).setScale(1, BigDecimal.ROUND_HALF_UP).doubleValue();
+                    }
+
+                }
+                //计算运动量//
+                Integer act1Int = eartagService.countAct(dataArray[2], dataArray[6]);
+                //命令头
+                dataJo.put("cmdHeader", dataArray[0]);
+                //采集器id
+                dataJo.put("device", dataArray[1]);
+                //耳标号
+                dataJo.put("earmark", dataArray[2]);
+                //电池电量
+                dataJo.put("bat", dataArray[3]);
+                //耳标温度
+                dataJo.put("earTemp", dataArray[4]);
+                dataJo.put("earTemp1", earTemp);
+                dataJo.put("earTemp2", earTemp2);
+                //环境温度
+                dataJo.put("envTemp", dataArray[5]);
+                dataJo.put("envTemp1", MathUtil.countEnvtemp(dataArray[5], 2));
+                //运动量
+                dataJo.put("act", dataArray[6]);
+                dataJo.put("act1", act1Int);
+                //信号强度
+                dataJo.put("signal", dataArray[7]);
+                //采集时间
+                dataJo.put("askTime", dataArray[8]);
+                //预留字段
+                dataJo.put("other", dataArray[9]);
+                type = "eartag";
+            }
+            outMap.put("type", type);
+            outMap.put("data", dataJo);
+            return outMap;
+        }
+
     public static void main(String[] args) {
         String eartag = "13320241020005";
         int i = eartag.hashCode();

huimv-eartag2-input/src/main/java/com/huimv/eartag2/service/impl/DeviceServiceImpl.java

@@ -84,6 +84,21 @@ public class DeviceServiceImpl implements IDeviceService {
         }
     }
 
+    @Override
+    public String getFarmIdByDeviceCode(String deviceCode) {
+        //
+        String farmId = (String)cacheService.getFarmIdByDeviceCode(deviceCode);
+        System.out.println("####### farmId=" + farmId);
+        if(farmId != null && farmId.trim().length()>0){
+            return farmId;
+        }else{
+            EartagDeviceRegisterEntity deviceRegisterEntity = deviceRegisterRepo.getDeviceRegister(deviceCode);
+            System.out.println("####### deviceRegisterEntity=" + deviceRegisterEntity);
+            //{将设备注册信息保存到缓存中}
+            cacheService.putDeviceRegister(deviceCode,deviceRegisterEntity);
+            return deviceRegisterEntity.getFarmId();
+        }
+    }
     //--处理状态数据
     private void handleEartagAllState() {