防止重复数据 入队列

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(MN +":"+ CN +":"+ dataTime, dataTime)){
+                                adapter.redisQueueService.addElementWithExpire(MN +":"+ CN +":"+ dataTime, dataTime, 2, 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(MN +":"+ CN +":"+ dataTime, dataTime)){
+                                    adapter.redisQueueService.addElementWithExpire(MN +":"+ CN +":"+ dataTime, dataTime, 2, TimeUnit.HOURS);
+                                    adapter.redisQueueService.sendMessage(Constant.SERVER, jsonObject);
+                                }
                             }
 //                            adapter.publisherService.pubMsg(jsonObject);
                         }