redisUtil方法修改

themis-business/src/main/java/com/qmth/themis/business/util/RedisUtil.java

@@ -23,7 +23,7 @@ import java.util.concurrent.TimeUnit;
 public class RedisUtil {
 
     @Resource
-    RedisTemplate redisTemplate;
+    RedisTemplate<String,Object> redisTemplate;
 
     /**
      * 获取机构信息
@@ -190,19 +190,6 @@ public class RedisUtil {
         return redisTemplate.opsForValue().multiGet(keys);
     }
 
-    /**
-     * 获取key like
-     *
-     * @param key
-     * @return
-     */
-    public Set getKeyPatterns(String key) {
-        if (null != key) {
-            return redisTemplate.keys(key + "*");
-        } else {
-            return null;
-        }
-    }
 
     /**
      * 设置hash
@@ -211,7 +198,7 @@ public class RedisUtil {
      * @param hashKey
      * @param hashValue
      */
-    public void setHashList(String key, String hashKey, Object hashValue) {
+    public void set(String key, String hashKey, Object hashValue) {
         redisTemplate.opsForHash().put(key, hashKey, hashValue);
     }
 
@@ -222,7 +209,7 @@ public class RedisUtil {
      * @param hashKey
      * @return
      */
-    public Object getHashList(String key, String hashKey) {
+    public Object get(String key, String hashKey) {
         return redisTemplate.opsForHash().get(key, hashKey);
     }
 
@@ -232,7 +219,7 @@ public class RedisUtil {
      * @param key
      * @param hashKey
      */
-    public void deleteHashList(String key, String hashKey) {
+    public void delete(String key, String hashKey) {
         redisTemplate.opsForHash().delete(key, hashKey);
     }
 
@@ -295,13 +282,12 @@ public class RedisUtil {
      * 删除锁
      *
      * @param key
-     * @param hashKey
      * @return
      */
-    public boolean releaseLock(String key, String hashKey) {
-        Object result = redisTemplate.opsForValue().get(key + hashKey);
+    public boolean releaseLock(String key) {
+        Object result = redisTemplate.opsForValue().get(key);
         if (Objects.nonNull(result)) {
-            redisTemplate.delete(key + hashKey);
+            delete(key);
             return true;
         } else {
             return false;
@@ -312,45 +298,41 @@ public class RedisUtil {
      * 设置缓存
      *
      * @param key
-     * @param hashKey
      * @param o
      */
-    public void setCacheValue(String key, String hashKey, Object o) {
-        redisTemplate.opsForValue().set(key + hashKey, o);
+    public void set(String key, Object o) {
+        redisTemplate.opsForValue().set(key, o);
     }
 
     /**
      * 设置缓存
      *
      * @param key
-     * @param hashKey
      * @param o
      * @param time
      * @param timeUnit
      */
-    public void setCacheValue(String key, String hashKey, Object o, long time, TimeUnit timeUnit) {
-        redisTemplate.opsForValue().set(key + hashKey, o, time, timeUnit);
+    public void set(String key, Object o, long time, TimeUnit timeUnit) {
+        redisTemplate.opsForValue().set(key, o, time, timeUnit);
     }
 
     /**
      * 获取缓存
      *
      * @param key
-     * @param hashKey
      * @return
      */
-    public Object getCacheValue(String key, String hashKey) {
-        return redisTemplate.opsForValue().get(key + hashKey);
+    public Object get(String key) {
+        return redisTemplate.opsForValue().get(key);
     }
 
     /**
      * 删除缓存
      *
      * @param key
-     * @param hashKey
      * @return
      */
-    public Object deleteCacheValue(String key, String hashKey) {
-        return redisTemplate.delete(key + hashKey);
+    public void delete(String key) {
+    	redisTemplate.expire(key,0,TimeUnit.SECONDS);
     }
 }

themis-exam/src/main/java/com/qmth/themis/exam/websocket/WebSocketServer.java

@@ -205,7 +205,7 @@ public class WebSocketServer
      * @return
      */
     public synchronized int getOnlineCount() {
-        Object o = redisUtil.getCacheValue(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, "");
+        Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
         return Objects.isNull(o) ? 0 : (int) o;
     }
 
@@ -214,13 +214,13 @@ public class WebSocketServer
      */
     public synchronized void addOnlineCount() {
         if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX, this.sessionId, SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT, TimeUnit.SECONDS)) {
-            Object o = redisUtil.getCacheValue(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, "");
+            Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
             int count = 0;
             if (Objects.nonNull(o)) {
                 count = (int) o;
             }
             count++;
-            redisUtil.setCacheValue(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, "", count);
+            redisUtil.set(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, count);
         }
     }
 
@@ -229,13 +229,13 @@ public class WebSocketServer
      */
     public synchronized void subOnlineCount() {
         if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX, this.sessionId, SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT, TimeUnit.SECONDS)) {
-            Object o = redisUtil.getCacheValue(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, "");
+            Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
             int count = 0;
             if (Objects.nonNull(o)) {
                 count = (int) o;
             }
             count--;
-            redisUtil.setCacheValue(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, "", count < 0 ? 0 : count);
+            redisUtil.set(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, count < 0 ? 0 : count);
         }
     }
 

themis-mq/src/main/java/com/qmth/themis/mq/listener/RocketSessionConsumer.java

@@ -1,15 +1,11 @@
 package com.qmth.themis.mq.listener;
 
-import com.google.gson.Gson;
-import com.qmth.themis.business.constant.SystemConstant;
-import com.qmth.themis.business.entity.TBSession;
-import com.qmth.themis.business.entity.TMRocketMessage;
-import com.qmth.themis.business.service.TBSessionService;
-import com.qmth.themis.business.service.TMRocketMessageService;
-import com.qmth.themis.business.util.JacksonUtil;
-import com.qmth.themis.business.util.RedisUtil;
-import com.qmth.themis.common.contanst.Constants;
-import com.qmth.themis.mq.dto.MqDto;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+import javax.annotation.Resource;
+
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
@@ -24,12 +20,17 @@ import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
 
-import javax.annotation.Resource;
-import java.util.List;
-import java.util.Objects;
-import java.util.concurrent.TimeUnit;
+import com.google.gson.Gson;
+import com.qmth.themis.business.constant.SystemConstant;
+import com.qmth.themis.business.entity.TBSession;
+import com.qmth.themis.business.entity.TMRocketMessage;
+import com.qmth.themis.business.service.TBSessionService;
+import com.qmth.themis.business.service.TMRocketMessageService;
+import com.qmth.themis.business.util.JacksonUtil;
+import com.qmth.themis.business.util.RedisUtil;
+import com.qmth.themis.common.contanst.Constants;
+import com.qmth.themis.mq.dto.MqDto;
 
 /**
  * @Description: 普通消息监听 session_topic
@@ -74,14 +75,14 @@ public class RocketSessionConsumer implements
                 mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
                 log.info(":{}-:{} sessionConsumer 接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
 //                log.info(":{}-:{} sessionConsumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
-                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.getHashList(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
+                if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
                     log.info(":{}-:{} 更新db", threadId, threadName);
                     tbSessionService.saveSessionInfo(JacksonUtil.readJson(JacksonUtil.parseJson(mqDto.getBody()), TBSession.class), mqDto.getTimestamp());
                     mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
                     TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
                     tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
                     tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-                    redisUtil.deleteHashList(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId());
+                    redisUtil.delete(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId());
                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                 } else {
                     log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
@@ -93,7 +94,7 @@ public class RocketSessionConsumer implements
             return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
         } finally {
             if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId());
+                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX+mqDto.getId());
             }
         }
         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功

themis-mq/src/main/java/com/qmth/themis/mq/listener/RocketTaskConsumer.java

@@ -75,7 +75,7 @@ public class RocketTaskConsumer {
                     mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
                     log.info(":{}-:{} task import Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
 //                    log.info(":{}-:{} task import Consumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.getHashList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
+                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
                         Map<String, Object> map = (Map<String, Object>) mqDto.getBody();
                         String tag = mqDto.getTag();
                         myThreadPool.arbitratePoolTaskExecutor.execute(() -> {
@@ -97,7 +97,7 @@ public class RocketTaskConsumer {
                         mqDto.setBody(JacksonUtil.parseJson(mqDto.getBody()));
                         TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
                         tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-                        redisUtil.deleteHashList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+                        redisUtil.delete(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST+ mqDto.getId());
                         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                     } else {
                         log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
@@ -109,7 +109,7 @@ public class RocketTaskConsumer {
                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
             } finally {
                 if (Objects.nonNull(mqDto)) {
-                    redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId());
+                    redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX+mqDto.getId());
                 }
             }
             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功
@@ -175,7 +175,7 @@ public class RocketTaskConsumer {
                     mqDto = JacksonUtil.readJson(new String(messageExt.getBody(), Constants.CHARSET), MqDto.class);
                     log.info(":{}-:{} task export Consumer接收到的消息:{}", threadId, threadName, JacksonUtil.parseJson(mqDto));
 //                    log.info(":{}-:{} task ExamStudentImport Consumer mqDto sequence:{},tag:{}", threadId, threadName, mqDto.getSequence(), mqDto.getTag());
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.getHashList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
+                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
                         Map<String, Object> map = (Map<String, Object>) mqDto.getBody();
                         String tag = mqDto.getTag();
                         myThreadPool.arbitratePoolTaskExecutor.execute(() -> {
@@ -193,7 +193,7 @@ public class RocketTaskConsumer {
                         mqDto.setBody(JacksonUtil.parseJson(mqDto.getBody()));
                         TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
                         tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-                        redisUtil.deleteHashList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+                        redisUtil.delete(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST+mqDto.getId());
                         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                     } else {
                         log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
@@ -205,7 +205,7 @@ public class RocketTaskConsumer {
                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
             } finally {
                 if (Objects.nonNull(mqDto)) {
-                    redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId());
+                    redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX+ mqDto.getId());
                 }
             }
             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功

themis-mq/src/main/java/com/qmth/themis/mq/listener/RocketUserLogConsumer.java

@@ -73,10 +73,10 @@ public class RocketUserLogConsumer implements MessageListenerConcurrently {
                 Object o = null;
                 String mqTopic = null;
                 if (tag.contains("user")) {
-                    o = redisUtil.getHashList(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+                    o = redisUtil.get(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId());
                     mqTopic = SystemConstant.USERLOG_TOPIC_BUFFER_LIST;
                 } else if (tag.contains("student")) {
-                    o = redisUtil.getHashList(SystemConstant.STUDENTLOG_TOPIC_BUFFER_LIST, mqDto.getId());
+                    o = redisUtil.get(SystemConstant.STUDENTLOG_TOPIC_BUFFER_LIST, mqDto.getId());
                     mqTopic = SystemConstant.STUDENTLOG_TOPIC_BUFFER_LIST;
                 }
                 if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(o) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT, TimeUnit.SECONDS)) {
@@ -89,7 +89,7 @@ public class RocketUserLogConsumer implements MessageListenerConcurrently {
                     mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
                     TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
                     tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-                    redisUtil.deleteHashList(mqTopic, mqDto.getId());
+                    redisUtil.delete(mqTopic, mqDto.getId());
                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                 } else {
                     log.info(":{}-:{} 消息ack未确认，重发", threadId, threadName);
@@ -101,7 +101,7 @@ public class RocketUserLogConsumer implements MessageListenerConcurrently {
             return ConsumeConcurrentlyStatus.RECONSUME_LATER;//重试
         } finally {
             if (Objects.nonNull(mqDto)) {
-                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX, mqDto.getId());
+                redisUtil.releaseLock(SystemConstant.REDIS_LOCK_MQ_PREFIX+mqDto.getId());
             }
         }
         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//成功

themis-mq/src/main/java/com/qmth/themis/mq/service/impl/MqDtoServiceImpl.java

@@ -93,16 +93,16 @@ public class MqDtoServiceImpl implements MqDtoService {
         if (Objects.nonNull(mqDto)) {
             switch (mqDto.getType().ordinal()) {
                 case 0:
-                    redisUtil.setHashList(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
+                    redisUtil.set(SystemConstant.SESSION_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
                     break;
                 case 1:
-                    redisUtil.setHashList(SystemConstant.STUDENTLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
+                    redisUtil.set(SystemConstant.STUDENTLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
                     break;
                 case 5:
-                    redisUtil.setHashList(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
+                    redisUtil.set(SystemConstant.USERLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
                     break;
                 default:
-                    redisUtil.setHashList(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
+                    redisUtil.set(SystemConstant.TASKLOG_TOPIC_BUFFER_LIST, mqDto.getId(), mqDto);
                     break;
             }
         }