update: 更新nutzboot/fastjson版本 & 优化掉redis keys -> scan 提升性能

Wizzercn · Oct 21, 2020 · 7138ed0 · 7138ed0
1 parent 3186a76
commit 7138ed0
Show file tree

Hide file tree

Showing 9 changed files with 57 additions and 49 deletions.
diff --git a/mqtt-auth/pom.xml b/mqtt-auth/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>mqtt-wk</artifactId>
         <groupId>cn.wizzer</groupId>
-        <version>1.1.2-netty</version>
+        <version>1.2.0-netty</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>

diff --git a/mqtt-broker/pom.xml b/mqtt-broker/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>mqtt-wk</artifactId>
         <groupId>cn.wizzer</groupId>
-        <version>1.1.2-netty</version>
+        <version>1.2.0-netty</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>

diff --git a/mqtt-common/pom.xml b/mqtt-common/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>mqtt-wk</artifactId>
         <groupId>cn.wizzer</groupId>
-        <version>1.1.2-netty</version>
+        <version>1.2.0-netty</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

diff --git a/mqtt-store/pom.xml b/mqtt-store/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>mqtt-wk</artifactId>
         <groupId>cn.wizzer</groupId>
-        <version>1.1.2-netty</version>
+        <version>1.2.0-netty</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>

diff --git a/mqtt-store/src/main/java/cn/wizzer/iot/mqtt/server/store/cache/RetainMessageCache.java b/mqtt-store/src/main/java/cn/wizzer/iot/mqtt/server/store/cache/RetainMessageCache.java
@@ -7,6 +7,8 @@
 import org.nutz.ioc.impl.PropertiesProxy;
 import org.nutz.ioc.loader.annotation.Inject;
 import org.nutz.ioc.loader.annotation.IocBean;
+import redis.clients.jedis.ScanParams;
+import redis.clients.jedis.ScanResult;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -43,14 +45,15 @@ public void remove(String topic) {
 
     public Map<String, RetainMessageStore> all() {
         Map<String, RetainMessageStore> map = new HashMap<>();
-        Set<String> set=redisService.keys(CACHE_PRE + "*");
-        if(set!=null&&!set.isEmpty()) {
-            set.forEach(
-                    entry -> {
-                        map.put(entry.substring(CACHE_PRE.length()), JSONObject.parseObject(redisService.get(entry), RetainMessageStore.class));
-                    }
-            );
-        }
+        ScanParams match = new ScanParams().match(CACHE_PRE + "*");
+        ScanResult<String> scan = null;
+        do {
+            scan = redisService.scan(scan == null ? ScanParams.SCAN_POINTER_START : scan.getStringCursor(), match);
+            for (String key : scan.getResult()) {
+                map.put(key.substring(CACHE_PRE.length()), JSONObject.parseObject(redisService.get(key), RetainMessageStore.class));
+
+            }
+        } while (!scan.isCompleteIteration());
         return map;
     }
 }
diff --git a/...-store/src/main/java/cn/wizzer/iot/mqtt/server/store/cache/SubscribeNotWildcardCache.java b/...-store/src/main/java/cn/wizzer/iot/mqtt/server/store/cache/SubscribeNotWildcardCache.java
@@ -6,6 +6,8 @@
 import org.nutz.integration.jedis.RedisService;
 import org.nutz.ioc.loader.annotation.Inject;
 import org.nutz.ioc.loader.annotation.IocBean;
+import redis.clients.jedis.ScanParams;
+import redis.clients.jedis.ScanResult;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -50,21 +52,21 @@ public void removeForClient(String clientId) {
 
     public Map<String, ConcurrentHashMap<String, SubscribeStore>> all() {
         Map<String, ConcurrentHashMap<String, SubscribeStore>> map = new HashMap<>();
-        Set<String> set = redisService.keys(CACHE_PRE + "*");
-        if (set != null && !set.isEmpty()) {
-            set.forEach(
-                    entry -> {
-                        ConcurrentHashMap<String, SubscribeStore> map1 = new ConcurrentHashMap<>();
-                        Map<String, String> map2 = redisService.hgetAll(entry);
-                        if (map2 != null && !map2.isEmpty()) {
-                            map2.forEach((k, v) -> {
-                                map1.put(k, JSONObject.parseObject(v, SubscribeStore.class));
-                            });
-                            map.put(entry.substring(CACHE_PRE.length()), map1);
-                        }
-                    }
-            );
-        }
+        ScanParams match = new ScanParams().match(CACHE_PRE + "*");
+        ScanResult<String> scan = null;
+        do {
+            scan = redisService.scan(scan == null ? ScanParams.SCAN_POINTER_START : scan.getStringCursor(), match);
+            for (String key : scan.getResult()) {
+                ConcurrentHashMap<String, SubscribeStore> map1 = new ConcurrentHashMap<>();
+                Map<String, String> map2 = redisService.hgetAll(key);
+                if (map2 != null && !map2.isEmpty()) {
+                    map2.forEach((k, v) -> {
+                        map1.put(k, JSONObject.parseObject(v, SubscribeStore.class));
+                    });
+                    map.put(key.substring(CACHE_PRE.length()), map1);
+                }
+            }
+        } while (!scan.isCompleteIteration());
         return map;
     }
 

diff --git a/mqtt-store/src/main/java/cn/wizzer/iot/mqtt/server/store/cache/SubscribeWildcardCache.java b/mqtt-store/src/main/java/cn/wizzer/iot/mqtt/server/store/cache/SubscribeWildcardCache.java
@@ -1,11 +1,14 @@
 package cn.wizzer.iot.mqtt.server.store.cache;
 
+import cn.wizzer.iot.mqtt.server.common.message.RetainMessageStore;
 import cn.wizzer.iot.mqtt.server.common.subscribe.SubscribeStore;
 import com.alibaba.fastjson.JSONObject;
 import org.nutz.aop.interceptor.async.Async;
 import org.nutz.integration.jedis.RedisService;
 import org.nutz.ioc.loader.annotation.Inject;
 import org.nutz.ioc.loader.annotation.IocBean;
+import redis.clients.jedis.ScanParams;
+import redis.clients.jedis.ScanResult;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -50,21 +53,21 @@ public void removeForClient(String clientId) {
 
     public Map<String, ConcurrentHashMap<String, SubscribeStore>> all() {
         Map<String, ConcurrentHashMap<String, SubscribeStore>> map = new HashMap<>();
-        Set<String> set = redisService.keys(CACHE_PRE + "*");
-        if (set != null && !set.isEmpty()) {
-            set.forEach(
-                    entry -> {
-                        ConcurrentHashMap<String, SubscribeStore> map1 = new ConcurrentHashMap<>();
-                        Map<String, String> map2 = redisService.hgetAll(entry);
-                        if (map2 != null && !map2.isEmpty()) {
-                            map2.forEach((k, v) -> {
-                                map1.put(k, JSONObject.parseObject(v, SubscribeStore.class));
-                            });
-                            map.put(entry.substring(CACHE_PRE.length()), map1);
-                        }
-                    }
-            );
-        }
+        ScanParams match = new ScanParams().match(CACHE_PRE + "*");
+        ScanResult<String> scan = null;
+        do {
+            scan = redisService.scan(scan == null ? ScanParams.SCAN_POINTER_START : scan.getStringCursor(), match);
+            for (String key : scan.getResult()) {
+                ConcurrentHashMap<String, SubscribeStore> map1 = new ConcurrentHashMap<>();
+                Map<String, String> map2 = redisService.hgetAll(key);
+                if (map2 != null && !map2.isEmpty()) {
+                    map2.forEach((k, v) -> {
+                        map1.put(k, JSONObject.parseObject(v, SubscribeStore.class));
+                    });
+                    map.put(key.substring(CACHE_PRE.length()), map1);
+                }
+            }
+        } while (!scan.isCompleteIteration());
         return map;
     }
 

diff --git a/mqtt-zoo/mqtt-test-kafka/pom.xml b/mqtt-zoo/mqtt-test-kafka/pom.xml
@@ -4,12 +4,12 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <groupId>cn.wizzer</groupId>
     <artifactId>mqtt-test-kafka</artifactId>
-    <version>1.1.2-netty</version>
+    <version>1.2.0-netty</version>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>
     <properties>
-        <mqttwk.version>1.1.2-netty</mqttwk.version>
-        <nutzboot.version>2.3.5.v20190516</nutzboot.version>
+        <mqttwk.version>1.2.0-netty</mqttwk.version>
+        <nutzboot.version>2.4.1.v20201014</nutzboot.version>
         <netty.version>4.1.28.Final</netty.version>
         <fastjson.version>1.2.55</fastjson.version>
         <hutool.version>4.1.2</hutool.version>

diff --git a/pom.xml b/pom.xml
@@ -6,7 +6,7 @@
     <groupId>cn.wizzer</groupId>
     <artifactId>mqtt-wk</artifactId>
     <packaging>pom</packaging>
-    <version>1.1.2-netty</version>
+    <version>1.2.0-netty</version>
     <name>MqttWk</name>
     <modules>
         <module>mqtt-common</module>
@@ -15,10 +15,10 @@
         <module>mqtt-store</module>
     </modules>
     <properties>
-        <mqttwk.version>1.1.2-netty</mqttwk.version>
-        <nutzboot.version>2.3.5.v20190516</nutzboot.version>
+        <mqttwk.version>1.2.0-netty</mqttwk.version>
+        <nutzboot.version>2.4.1.v20201014</nutzboot.version>
         <netty.version>4.1.28.Final</netty.version>
-        <fastjson.version>1.2.60</fastjson.version>
+        <fastjson.version>1.2.74</fastjson.version>
         <hutool.version>4.1.2</hutool.version>
         <kafka_2.12.version>2.0.0</kafka_2.12.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>