回复消息 bug修改

pom.xml

@@ -204,7 +204,14 @@
 			<artifactId>xmlbeans</artifactId>
 			<version>5.1.1</version>
 		</dependency>
-    </dependencies>
+		<!-- sharding-jdbc依赖 -->
+		<dependency>
+			<groupId>org.apache.shardingsphere</groupId>
+			<artifactId>sharding-jdbc-spring-boot-starter</artifactId>
+			<version>4.0.0-RC1</version>
+		</dependency>
+
+	</dependencies>
 
 
 	<dependencyManagement>

src/main/java/com/cn/esermis/config/RedisQueueService.java

@@ -2,13 +2,18 @@ package com.cn.esermis.config;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.ListOperations;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
+import java.util.concurrent.TimeUnit;
+
 @Service
 public class RedisQueueService {
 
     @Autowired
     private ListOperations<String, Object> listOps;
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
 
     public void sendMessage(String queueKey, Object message) {
         listOps.rightPush(queueKey, message);
@@ -18,4 +23,20 @@ public class RedisQueueService {
         return listOps.leftPop(queueKey);
     }
 
+    public void addElementWithExpire(String key, String element, long expireTime, TimeUnit timeUnit) {
+        // 将元素添加到 Hash 中
+        redisTemplate.opsForHash().put("dataKey:"+key, element, element);
+        // 设置过期时间
+        redisTemplate.expire("dataKey:"+key, expireTime, timeUnit);
+    }
+
+    public boolean isElementExist(String key, String element) {
+        // 检查元素是否存在
+        return redisTemplate.opsForHash().hasKey("dataKey:"+key, element);
+    }
+
+    public void removeElement(String key, String element) {
+        // 删除元素
+        redisTemplate.opsForHash().delete("dataKey:"+key, element);
+    }
 }

src/main/java/com/cn/esermis/netty/BootNettyChannelInboundHandlerAdapter.java

@@ -23,6 +23,7 @@ import java.net.InetSocketAddress;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 @Log4j2
 @Component
@@ -91,7 +92,13 @@ public class BootNettyChannelInboundHandlerAdapter extends ChannelInboundHandler
                             ||(jsonObject!=null && jsonObject.getString(Constant.CN).equals(Constant.HOUR))){
 //                        adapter.publisherService.pubMsg(jsonObject);
                         if (ObjectUtils.notEqual(jsonObject.get("CN"),Constant.MINUTE_STR)){
-                            adapter.redisQueueService.sendMessage(Constant.SERVER, jsonObject);
+                            String CN = jsonObject.getString("CN");
+                            String MN = jsonObject.getString("MN");
+                            String dataTime = jsonObject.getString("DataTime");
+                            if (!adapter.redisQueueService.isElementExist(CN + MN + dataTime, dataTime)){
+                                adapter.redisQueueService.addElementWithExpire(CN+MN+dataTime, dataTime, 1, TimeUnit.HOURS);
+                                adapter.redisQueueService.sendMessage(Constant.SERVER, jsonObject);
+                            }
                         }
                     }
                 } else {
@@ -102,7 +109,13 @@ public class BootNettyChannelInboundHandlerAdapter extends ChannelInboundHandler
                         if ((jsonObject!=null &&  jsonObject.getString(Constant.CN).equals(Constant.MINUTE))
                                 || (jsonObject!=null && jsonObject.getString(Constant.CN).equals(Constant.HOUR))){
                             if (ObjectUtils.notEqual(jsonObject.get("CN"),Constant.MINUTE_STR)) {
-                                adapter.redisQueueService.sendMessage(Constant.SERVER, jsonObject);
+                                String CN = jsonObject.getString("CN");
+                                String MN = jsonObject.getString("MN");
+                                String dataTime = jsonObject.getString("DataTime");
+                                if (!adapter.redisQueueService.isElementExist(CN + MN + dataTime, dataTime)){
+                                    adapter.redisQueueService.addElementWithExpire(CN+MN+dataTime, dataTime, 1, TimeUnit.HOURS);
+                                    adapter.redisQueueService.sendMessage(Constant.SERVER, jsonObject);
+                                }
                             }
 //                            adapter.publisherService.pubMsg(jsonObject);
                         }

src/main/java/com/cn/esermis/server/Consumer.java

@@ -8,7 +8,9 @@ import com.cn.esermis.dpld.business.MonitorBusiness;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
+import redis.clients.jedis.Jedis;
 
+import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -23,7 +25,10 @@ public class Consumer {
     private MonitorBusiness monitorBusiness;
     @Autowired
     private RedisQueueService redisQueueService;
-
+//    @Autowired
+//    private Jedis jedis;
+//    private String lockKey = "serverlockey";
+//    private int lockTimeout = 10;
 
     @Async
     public void receive() throws InterruptedException {
@@ -32,7 +37,9 @@ public class Consumer {
         int numberOfThreads = 8;
         int tolot = 0;
         ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
-
+//        String identifier = UUID.randomUUID().toString();
+//        String lockValue = identifier;
+//        long end = System.currentTimeMillis() + lockTimeout;
         while (true){
             //取出消息
             Object message = redisQueueService.receiveMessage(Constant.SERVER);

src/main/resources/application.yml

@@ -14,19 +14,62 @@ spring:
 #    url: ${DB_URL:jdbc:oracle:thin:@120.77.183.31:1521:xe}
 #    username: ${DB_USERNAME:CNOOC_JNJP}
 #    password: ${DB_PASSWORD:cnooc}
-  datasource:
-    dynamic:
-      primary: master
-      strict: false
-      datasource:
-        master:
-          driver-class-name: com.mysql.cj.jdbc.Driver
-          url: jdbc:mysql://39.105.121.97:13306/scy?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
-#          url: jdbc:mysql://192.168.3.19:13306/scy?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
-          username: root
+#  datasource:
+#    dynamic:
+#      primary: master
+#      strict: false
+#      datasource:
+#        master:
+#          driver-class-name: com.mysql.cj.jdbc.Driver
+#          url: jdbc:mysql://39.105.121.97:13306/scy?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
+##          url: jdbc:mysql://192.168.3.19:13306/scy?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
+#          username: root
+##          password: Eitc@gnhz702
 #          password: Eitc@gnhz702
-          password: Eitc@gnhz702
-
+  shardingsphere:
+    props:
+      #d打印Sql语句
+      sql-show: true
+    datasource:
+      #创建我们的ds0数据源
+      scy:
+        #下边这些都是老套路了
+        driver-class-name: com.mysql.cj.jdbc.Driver
+        jdbc-url: jdbc:mysql://192.168.3.19:3366/scy?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
+#        jdbc-url: jdbc:mysql://39.105.121.97:13306/scy?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
+        password: Eitc@gnhz702
+        type: com.zaxxer.hikari.HikariDataSource
+        username: root
+      #创建我们的ds1数据源
+#      ds1:
+#        #一样的老套路
+#        driver-class-name: com.mysql.cj.jdbc.Driver
+#        jdbc-url: jdbc:mysql://39.105.121.97:13306/scy?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
+#        password: Eitc@gnhz702
+#        type: com.zaxxer.hikari.HikariDataSource
+#        username: root
+      names: scy
+#指定 course 表分布情况，配置表在哪个数据库里面，表名称都是什么 m1.course_1 ,m1.course_2
+    sharding:
+      tables:
+        #scs_product可以任意命名，sql中一致，为了方便理解，这里一般写分表共有的
+        # 比如我的scs_product_1  和 scs_product_2  那就写scs_product好区分
+        # 当执行sql中出现scs_product，sharding-jdbc会将其操作到对应的表
+        air_monitor_minute_record:
+          actual-data-nodes: scy.air_monitor_minute_record_$->{1..2}
+          # 指定主键生成策略为雪花id,全局主键
+          key-generator:
+            column: id
+            type: SNOWFLAKE
+          #指定scs_product表的分片策略，  分片键和分片算法 用于计算真正的表名
+          table-strategy:
+            inline:
+              # 偶数进到scs_product_1  奇数进到scs_product_2
+              # 对于根据分片字段为条件的，会先判断是否涉及两张表，
+              # 如何是两张表则会两个表都查，如果只涉及单表，则只查询一张表。
+              algorithm-expression: air_monitor_minute_record_$->{id % 2 + 1}
+              # 指定分片键为PRODUCT_ID
+              sharding-column: id
   # Redis 配置
   redis:
     timeout: 30000 # 连接超时时间（毫秒）
@@ -47,74 +90,4 @@ spring:
 mybatis-plus:
   mapper-locations: classpath:static/**/*.xml
   configuration:
-    log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl # 指定 MyBatis 使用 SLF4J 日志实现
-#logging:
-#  level:
-#    cnooc.esermis: DEBUG
-#    com.baomidou.mybatisplus.core.mapper: debug # 设置 Mapper 接口的日志级别为 DEBUG
-  ## 工程设置
-#project:
-#  debug: false
-#  # 上传文件目录
-#  uploadDir: ${UPLOAD_DIR:./uploads}
-#  userspaceDir: ${USERSPACE_DIR:./userspace}
-#
-## 全局常量
-#constants:
-#  defaultPassword: P@ssw0rd
-#mybatis
-#mybatis-plus:
-#  mapper-locations: classpath:/static/mybatis/*.xml,classpath:/static/mybatis/gf/*.xml,classpath:/static/mybatis/leg/*.xml
-#  #实体扫描，多个package用逗号或者分号分隔
-#  typeAliasesPackage: com.cnooc.esermis.entity,com.cnooc.esermis.gf.entity,com.cnooc.esermis.gf.sewage.entity,com.cnooc.esermis.leg.entity
-#  typeEnumsPackage: com.cnooc.esermis.entity.enums,com.cnooc.esermis.gf.state
-#  typeHandlersPackage: com.cnooc.esermis.gf.handler
-#  global-config:
-#    # 数据库相关配置
-#    db-config:
-#      #主键类型  AUTO:"数据库ID自增", INPUT:"用户输入ID",ID_WORKER:"全局唯一ID (数字类型唯一ID)", ASSIGN_UUID:"全局唯一ID ASSIGN_UUID";
-#      id-type: INPUT # 李原：将ID_WORKER修改为INPUT 以配合每个表自己配置ID生成策略
-#      #字段策略 IGNORED:"忽略判断",NOT_NULL:"非 NULL 判断"),NOT_EMPTY:"非空判断"
-#      field-strategy: NOT_NULL
-#      #驼峰下划线转换
-#      column-underline: false
-#      #数据库大写下划线转换
-#      capital-mode: false
-#      #逻辑删除配置
-#      logic-delete-value: 0
-#      logic-not-delete-value: 1
-#      db-type: oracle
-#    #刷新mapper 调试神器
-#    refresh: true
-#  # 原生配置
-#  configuration:
-#    map-underscore-to-camel-case: true
-#    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
-#    cache-enabled: false
-
-  ##日志配置
-
-
-#线程池
-##线程池维护线程的最少数量 -
-#consumer:
-#  corePoolSize: 5
-#  #线程池维护线程的最大数量
-#  maxPoolSize: 150
-#  #线程池维护线程所允许的空闲时间
-#  keepAliveSeconds: 500
-#  #线程池所使用的缓冲队列
-#  queueCapacity: 80
-## 邮件通知
-#emailsettings:
-#  # 设置为debug模式, 可以查看详细的发送 log
-#  debug: true
-#  #发送人账户服务器信息
-#  host: cmail.cnooc.com.cn #10.68.81.34
-#  port: 25
-#  senderAccount: sy_huanbao
-#  senderPassword: swu5!DHW11
-#  from: 环保系统自动邮件<sy_huanbao@cnooc.com.cn>
-
-
-
+    log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl # 指定 MyBatis 使用 SLF4J 日志实现