耳标平台

huimv-eartag2-common/src/main/java/com/huimv/eartag2/common/config/InterceptorConfig.java

@@ -32,6 +32,7 @@ public class InterceptorConfig implements WebMvcConfigurer {
                         "/manage2/eartagData/getEnvByTime",
                         "/**/eartagData/getEnvByTime",
                         "/**/eartag-hour-act/getAct",
-                        "/manage2/eartag-hour-act/getAct");
+                        "/manage2/eartag-hour-act/getAct",
+                        "/sendDirectMessage2");
     }
 }

huimv-eartag2-common/src/main/java/com/huimv/eartag2/common/mq/Const.java

@@ -1,5 +1,6 @@
 package com.huimv.eartag2.common.mq;
 
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
 /**
@@ -11,6 +12,9 @@ import org.springframework.stereotype.Component;
  * @Create : 2020-12-25
  **/
 public class Const {
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
     public final static String ROUTING_KEY_ASK = "topic.ask.key";
     public final static String QUEUE_ASK = "topic.ask.queue";
     public final static String EXCHANGE_ASK = "topic.ask.exchange";
@@ -33,9 +37,9 @@ public class Const {
     public final static String ROUTING_KEY_ASK_DEVICE = "topic.askDevice.routeKey";
     public final static String QUEUE_ASK_DEVICE = "topic.askDevice.queue";
 
-    public final static String EXCHANGE_ASK_EARTAG = "topic.askEartag.exchange";
-    public final static String ROUTING_KEY_ASK_EARTAG = "topic.askEartag.routeKey";
-    public final static String QUEUE_ASK_EARTAG = "topic.askEartag.queue";
+    public final static String EXCHANGE_ASK_EARTAG = "direct.askEartag.exchange";
+    public final static String ROUTING_KEY_ASK_EARTAG = "direct.askEartag.routeKey";
+    public final static String QUEUE_ASK_EARTAG = "direct.askEartag.queue";
 
     public final static String QUEUE_SYNC_DEVICE_REGISTER = "topic.syncDeviceRegister.queue";
     public final static String EXCHANGE_SYNC_DEVICE_REGISTER = "topic.syncDeviceRegister.exchange";

huimv-eartag2-eartag/pom.xml

@@ -13,14 +13,12 @@
 
     <dependencies>
 
-        <!-- kafka-->
-<!--        <dependency>-->
-<!--            <groupId>org.springframework.kafka</groupId>-->
-<!--            <artifactId>spring-kafka</artifactId>-->
-<!--            <version>2.6.9</version>-->
-<!--        </dependency>-->
 
         <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+        </dependency>
+        <dependency>
             <!--自动配置-->
             <groupId>com.alibaba</groupId>
             <artifactId>druid-spring-boot-starter</artifactId>
@@ -50,11 +48,7 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-undertow</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-test</artifactId>
-            <scope>test</scope>
-        </dependency>
+
         <!--rabbitmq-->
         <dependency>
             <groupId>org.springframework.boot</groupId>

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

@@ -12,6 +12,7 @@ import org.springframework.amqp.rabbit.annotation.RabbitHandler;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.beans.factory.annotation.Autowired;
 //import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
 import java.io.BufferedWriter;
@@ -41,9 +42,9 @@ public class EartagListener {
     @Autowired
     private IEartagService eartagService;
 
-
     private static final String TOPIC = "test-topic";
 
+
 //    @KafkaListener(topics = TOPIC, groupId = "my-group")
 //    public void listen(ConsumerRecord<String, String> record) {
 //        System.out.println("Received message: " + record.value());
@@ -57,10 +58,12 @@ public class EartagListener {
      * @Date : 2022/3/12
      * @Time : 17:53
      */
-    @RabbitListener(queues = Const.QUEUE_ASK_EARTAG)
+//    @RabbitListener(queues = Const.QUEUE_ASK_EARTAG )
+    @RabbitListener(queues = "#{'direct.askEartag.queue' + ${rabbitmq.queues}}")
     @RabbitHandler
     public void processRawdata(Map RawMap) throws ParseException, IOException {
         System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<< 处理耳标 <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<");
+        System.out.println(RawMap);
         String askText = RawMap.get("askText").toString();
 
         //本模块主要执行保存以下4类数据：（其他注册数据，在线数据，状态数据等都是由其他模块工程执行处理-process2模块工程）
@@ -70,7 +73,7 @@ public class EartagListener {
         //--保存耳标流水数据（耳标流水表）
 
         //测试-临时写入耳标数据到文件
-        DateUtil dateUtil = new DateUtil();
+//        DateUtil dateUtil = new DateUtil();
 //        if(askText.trim().indexOf("330110004000010") != -1){
 //            writeTxt(dateUtil.getTodayMissionText()+" "+askText,"eartag");
 //        }

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

@@ -38,8 +38,6 @@ import java.util.Map;
 @Slf4j
 public class DeviceServiceImpl implements IDeviceService {
     @Autowired
-    private MathUtil mathUtil;
-    @Autowired
     private DateUtil dateUtil;
     @Autowired
     private EartagRawRepo rawRepo;

huimv-eartag2-eartag/src/main/resources/bootstrap.yml

@@ -7,8 +7,8 @@ spring:
     nacos:
       config:
         # 配置中心的地址
-       # server-addr: 47.98.40.154:8848
-        server-addr: 127.0.0.1:8848
+        server-addr: 47.98.40.154:8848
+#        server-addr: 127.0.0.1:8848
         # 配置文件prefix
         prefix: ${spring.application.name}
         # 配置文件的格式

huimv-eartag2-input/pom.xml

@@ -28,12 +28,12 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
             <!-- 移除掉默认支持的 Tomcat -->
-            <exclusions>
-                <exclusion>
-                    <groupId>org.springframework.boot</groupId>
-                    <artifactId>spring-boot-starter-tomcat</artifactId>
-                </exclusion>
-            </exclusions>
+<!--            <exclusions>-->
+<!--                <exclusion>-->
+<!--                    <groupId>org.springframework.boot</groupId>-->
+<!--                    <artifactId>spring-boot-starter-tomcat</artifactId>-->
+<!--                </exclusion>-->
+<!--            </exclusions>-->
         </dependency>
         <!-- 添加 Undertow 容器 -->
         <dependency>

huimv-eartag2-input/src/main/java/com/huimv/eartag2/HuimvEartag2InputApplication.java

@@ -2,9 +2,13 @@ package com.huimv.eartag2;
 
 import com.huimv.eartag2.server.EartagServer;
 import com.huimv.eartag2.server.EartagServer2;
+import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
 
 /**
  * @Project : huimv.shiwan
@@ -15,6 +19,7 @@ import org.springframework.context.ApplicationContext;
  * @Create : 2020-12-25
  **/
 @SpringBootApplication
+@EnableRabbit
 public class HuimvEartag2InputApplication {
     public static void main(String[] args) throws InterruptedException {
 //        SpringApplication.run(HuimvEartagApplication.class, args);

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

@@ -0,0 +1,44 @@
+package com.huimv.eartag2.config;
+
+import com.huimv.eartag2.common.mq.Const;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.core.Binding;  
+import org.springframework.amqp.core.BindingBuilder;  
+import org.springframework.amqp.core.DirectExchange;  
+import org.springframework.amqp.core.Queue;
+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;
+import sun.rmi.runtime.Log;
+
+import java.util.List;  
+import java.util.Map;  
+  
+@Component  
+public class RabbitMQInitializer implements ApplicationRunner {
+
+    @Autowired
+    private RabbitMQService rabbitMQService;
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
+    @Override  
+    public void run(ApplicationArguments args) throws Exception {
+        // 声明交换机
+        // 创建交换机
+        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/config/RabbitMqConfig.java

@@ -0,0 +1,43 @@
+package com.huimv.eartag2.config;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitMqConfig {
+    
+    @Bean
+    public Queue dynamicQueue(final RabbitAdmin rabbitAdmin) {
+        return new Queue(queueName(), true);
+    }
+    
+    @Bean
+    public DirectExchange dynamicExchange(final RabbitAdmin rabbitAdmin) {
+        return new DirectExchange(exchangeName(), true, false);
+    }
+    
+    @Bean
+    public Binding binding(Queue dynamicQueue, DirectExchange dynamicExchange) {
+        return BindingBuilder.bind(dynamicQueue).to(dynamicExchange).with(routingKey());
+    }
+    
+    private String queueName() {
+        // 动态生成的队列名称
+        return "dynamicQueue";
+    }
+    
+    private String exchangeName() {
+        // 动态生成的交换机名称
+        return "dynamicExchange";
+    }
+    
+    private String routingKey() {
+        // 路由键
+        return "routingKey";
+    }
+}

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

@@ -23,31 +23,31 @@ public class TopicRabbitMQConfig {
     public Queue askDeviceQueue() {
         return new Queue(Const.QUEUE_ASK_DEVICE);
     }
-    @Bean
-    public Queue askEartagQueue() {
-        return new Queue(Const.QUEUE_ASK_EARTAG);
-    }
+//    @Bean
+//    public Queue askEartagQueue() {
+//        return new Queue(Const.QUEUE_ASK_EARTAG);
+//    }
 
     //原始交换机
-    @Bean
-    TopicExchange askDeviceExchange() {
-        return new TopicExchange(Const.EXCHANGE_ASK_DEVICE);
-    }
-    @Bean
-    TopicExchange askEartagExchange() {
-        return new TopicExchange(Const.EXCHANGE_ASK_EARTAG);
-    }
+//    @Bean
+//    TopicExchange askDeviceExchange() {
+//        return new TopicExchange(Const.EXCHANGE_ASK_DEVICE);
+//    }
+//    @Bean
+//    TopicExchange askEartagExchange() {
+//        return new TopicExchange(Const.EXCHANGE_ASK_EARTAG);
+//    }
 
     //绑定关系
-    @Bean
-    Binding bindingDeviceExchangeMessage() {
-        return BindingBuilder.bind(askDeviceQueue()).to(askDeviceExchange()).with(Const.ROUTING_KEY_ASK_DEVICE);
-    }
-    @Bean
-    Binding bindingEartagExchangeMessage() {
-
-        return BindingBuilder.bind(askEartagQueue()).to(askEartagExchange()).with(Const.ROUTING_KEY_ASK_EARTAG);
-    }
+//    @Bean
+//    Binding bindingDeviceExchangeMessage() {
+//        return BindingBuilder.bind(askDeviceQueue()).to(askDeviceExchange()).with(Const.ROUTING_KEY_ASK_DEVICE);
+//    }
+//    @Bean
+//    Binding bindingEartagExchangeMessage() {
+//
+//        return BindingBuilder.bind(askEartagQueue()).to(askEartagExchange()).with(Const.ROUTING_KEY_ASK_EARTAG);
+//    }
 
     // 同步配置消息 原始数据队列
     @Bean

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

@@ -2,9 +2,16 @@ package com.huimv.eartag2.controller;
 
 import com.alibaba.fastjson.JSONObject;
 import com.huimv.eartag2.service.IDeviceService;
+import com.huimv.eartag2.service.RabbitMQService;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 import java.time.LocalDateTime;
@@ -22,12 +29,39 @@ import java.util.UUID;
  * @Create : 2020-12-25
  **/
 @RestController
+@RequestMapping
 public class SendMessageController {
     @Autowired
     RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法
     @Autowired
     private IDeviceService deviceService;
 
+    //动态创建队列
+//    @Autowired
+//    private RabbitMQService rabbitMQService;
+
+//    @GetMapping("/sendDirectMessage2")
+//    public void sendMessage(){
+//
+//        rabbitMQService.createQueue("myDynamicQueue");
+//
+//        // 创建交换机
+//        rabbitMQService.createExchange("myDynamicExchange");
+//
+//        // 创建绑定
+//        rabbitMQService.createBinding("myDynamicQueue", "myDynamicExchange", "myRoutingKey");
+//
+//
+//        rabbitTemplate.convertAndSend("myDynamicExchange", "myRoutingKey", "message");
+//    }
+//
+//
+//    @RabbitListener(queues = "myDynamicQueue")
+//    public void receiveMessage(String message) {
+//        System.out.println("Received message: " + message);
+//    }
+
+
     @GetMapping("/sendDirectMessage")
     public String sendDirectMessage() {
         String messageId = String.valueOf(java.util.UUID.randomUUID());

huimv-eartag2-input/src/main/java/com/huimv/eartag2/server/EartagServerHandler2.java

@@ -245,7 +245,7 @@ public class EartagServerHandler2 extends ChannelInboundHandlerAdapter {
 //                    writeTxt(dateUtil.getTodayMissionText()+" "+askText,"##earmark_mq_330110004000010");
 //                }
                 //{发送到耳标处理消息队列}
-                dataService.sendEartagMQ(map);
+                dataService.sendEartagMQ(map,cmd);
                 log.info(">>发送耳标数据消息到MQ." + map);
             } else {
                 System.out.println("无效设备编码 idCode=" + idCode);

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

@@ -21,7 +21,7 @@ public interface IDataService {
     void sendBizMQ(String cmdHeader, String cmd, String askText);
 
     //发送设备消息
-    void sendEartagMQ(Map map) throws IOException, ParseException;
+    void sendEartagMQ(Map map,String  eartag) throws IOException, ParseException;
 
     //
     String getDeviceCodeByChipId(String idCode);

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

@@ -0,0 +1,39 @@
+package com.huimv.eartag2.service;
+
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+
+@Service  
+public class RabbitMQService {  
+  
+    private final AmqpAdmin amqpAdmin;  
+  
+    @Autowired  
+    public RabbitMQService(AmqpAdmin amqpAdmin) {  
+        this.amqpAdmin = amqpAdmin;  
+    }
+
+
+    public void createQueue(String queueName) {
+        Queue queue = new Queue(queueName);
+        amqpAdmin.declareQueue(queue);
+    }
+
+    public void createExchange(String exchangeName) {
+        Exchange exchange = new DirectExchange(exchangeName);
+        amqpAdmin.declareExchange(exchange);
+    }
+    public void createBinding(String queueName, String exchangeName, String routingKey) {
+
+//        amqpAdmin.declareBinding(BindingBuilder.bind(new Queue(queueName, true)).to(new DirectExchange(exchangeName, true, false)).with(routingKey));
+        Queue queue = new Queue(queueName);
+        Binding binding = BindingBuilder.bind(queue).to(new DirectExchange(exchangeName)).with(routingKey);
+        amqpAdmin.declareBinding(binding);
+    }
+}

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

@@ -21,6 +21,7 @@ import org.springframework.amqp.AmqpTimeoutException;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
@@ -290,8 +291,12 @@ public class DataServiceImpl implements IDataService {
         System.out.println("-->3.5");
     }
 
+
+    @Value("${rabbitmq.queues}")
+    private Integer queues;
+
     @Override
-    public void sendEartagMQ(Map map) throws IOException, ParseException {
+    public void sendEartagMQ(Map map,String eartag) throws IOException, ParseException {
 //        try {
 //            rabbitTemplate.setConfirmCallback(new MsgSendConfirmCallBack());
 //
@@ -299,7 +304,11 @@ public class DataServiceImpl implements IDataService {
 //            log.info("开始发送消息");
 //            correlationData.setId("100");
             //发送到耳标处理消息队列
-            rabbitTemplate.convertAndSend(Const.EXCHANGE_ASK_EARTAG, Const.ROUTING_KEY_ASK_EARTAG, map);
+            //zj+330102002000003+13320241020004+87+0+10+154894+80+20240126174333+1.2#3.0#65535#10794+0+end
+        int i = eartag.hashCode() %queues;
+        log.info("eartag.hashCode()"+eartag.hashCode());
+        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);
 //        }

huimv-eartag2-input/src/main/resources/bootstrap.yml

@@ -7,8 +7,8 @@ spring:
     nacos:
       config:
         # 配置中心的地址
-#        server-addr: 47.98.40.154:8848
-        server-addr: 127.0.0.1:8848
+        server-addr: 47.98.40.154:8848
+#        server-addr: 127.0.0.1:8848
         # 配置文件prefix
         prefix: ${spring.application.name}
         # 配置文件的格式

huimv-eartag2-manage/src/main/java/com/huimv/eartag2/manage/config/TopicRabbitMQConfig.java

@@ -73,22 +73,22 @@ public class TopicRabbitMQConfig {
     }
 
     //原始交换机
-    @Bean
-    TopicExchange askDeviceExchange() {
-        return new TopicExchange(Const.EXCHANGE_ASK_DEVICE);
-    }
-    @Bean
-    TopicExchange askEartagExchange() {
-        return new TopicExchange(Const.EXCHANGE_ASK_EARTAG);
-    }
+//    @Bean
+//    TopicExchange askDeviceExchange() {
+//        return new TopicExchange(Const.EXCHANGE_ASK_DEVICE);
+//    }
+//    @Bean
+//    TopicExchange askEartagExchange() {
+//        return new TopicExchange(Const.EXCHANGE_ASK_EARTAG);
+//    }
 
     //绑定关系
-    @Bean
-    Binding bindingDeviceExchangeMessage() {
-        return BindingBuilder.bind(askDeviceQueue()).to(askDeviceExchange()).with(Const.ROUTING_KEY_ASK_DEVICE);
-    }
-    @Bean
-    Binding bindingEartagExchangeMessage() {
-        return BindingBuilder.bind(askEartagQueue()).to(askEartagExchange()).with(Const.ROUTING_KEY_ASK_EARTAG);
-    }
+//    @Bean
+//    Binding bindingDeviceExchangeMessage() {
+//        return BindingBuilder.bind(askDeviceQueue()).to(askDeviceExchange()).with(Const.ROUTING_KEY_ASK_DEVICE);
+//    }
+//    @Bean
+//    Binding bindingEartagExchangeMessage() {
+//        return BindingBuilder.bind(askEartagQueue()).to(askEartagExchange()).with(Const.ROUTING_KEY_ASK_EARTAG);
+//    }
 }