perf[orm]: 简化lpmap的代码结构，并增加测试用例

orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java

@@ -14,6 +14,8 @@ package com.zfoo.orm.lpmap;
 
 import com.zfoo.protocol.IPacket;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
 
@@ -21,7 +23,7 @@ import java.util.function.BiConsumer;
  * @author jaysunxiao
  * @version 3.0
  */
-public class ConcurrentFileChannelHeapMap<V extends IPacket> implements LpMap<V> {
+public class ConcurrentFileChannelHeapMap<V extends IPacket> implements LpMap<V>, Closeable {
 
     private ReentrantLock fileChannelLock = new ReentrantLock();
 
@@ -33,7 +35,7 @@ public class ConcurrentFileChannelHeapMap<V extends IPacket> implements LpMap<V>
         fileChannelMap = new FileChannelMap<>(dbPath, clazz);
         concurrentHeapMap = new ConcurrentHeapMap<>();
 
-        load();
+        fileChannelMap.forEach((key, v) -> concurrentHeapMap.put(key, v));
     }
 
     @Override
@@ -93,18 +95,14 @@ public class ConcurrentFileChannelHeapMap<V extends IPacket> implements LpMap<V>
         }
     }
 
-    private void load() {
-        var maxIndex = fileChannelMap.getMaxIndex();
-        if (maxIndex <= 0) {
-            return;
-        }
-
-        for (var key = 1; key <= maxIndex; key++) {
-            var value = fileChannelMap.get(key);
-            if (value == null) {
-                continue;
-            }
-            concurrentHeapMap.put(key, value);
+    @Override
+    public void close() throws IOException {
+        fileChannelLock.lock();
+        try {
+            fileChannelMap.close();
+            concurrentHeapMap.clear();
+        } finally {
+            fileChannelLock.unlock();
         }
     }
 
@@ -112,4 +110,5 @@ public class ConcurrentFileChannelHeapMap<V extends IPacket> implements LpMap<V>
     public void forEach(BiConsumer<Long, V> biConsumer) {
         concurrentHeapMap.forEach(biConsumer);
     }
+
 }

orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMap.java

@@ -14,6 +14,8 @@ package com.zfoo.orm.lpmap;
 
 import com.zfoo.protocol.IPacket;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.List;
 import java.util.function.BiConsumer;
 
@@ -21,7 +23,7 @@ import java.util.function.BiConsumer;
  * @author jaysunxiao
  * @version 3.0
  */
-public class ConcurrentFileChannelMap<V extends IPacket> implements LpMap<V> {
+public class ConcurrentFileChannelMap<V extends IPacket> implements LpMap<V>, Closeable {
 
 
     private FileChannelMap<V> fileChannelMap;
@@ -69,8 +71,14 @@ public class ConcurrentFileChannelMap<V extends IPacket> implements LpMap<V> {
         fileChannelMap.clear();
     }
 
+    @Override
+    public synchronized void close() throws IOException {
+        fileChannelMap.close();
+    }
+
     @Override
     public synchronized void forEach(BiConsumer<Long, V> biConsumer) {
         fileChannelMap.forEach(biConsumer);
     }
+
 }

orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java

@@ -14,13 +14,15 @@ package com.zfoo.orm.lpmap;
 
 import com.zfoo.protocol.IPacket;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.function.BiConsumer;
 
 /**
  * @author jaysunxiao
  * @version 3.0
  */
-public class FileChannelHeapMap<V extends IPacket> implements LpMap<V> {
+public class FileChannelHeapMap<V extends IPacket> implements LpMap<V>, Closeable {
 
     private FileChannelMap<V> fileChannelMap;
 
@@ -30,7 +32,7 @@ public class FileChannelHeapMap<V extends IPacket> implements LpMap<V> {
         fileChannelMap = new FileChannelMap<>(dbPath, clazz);
         heapMap = new HeapMap<>(initialCapacity);
 
-        load();
+        fileChannelMap.forEach((key, v) -> heapMap.put(key, v));
     }
 
     @Override
@@ -66,23 +68,15 @@ public class FileChannelHeapMap<V extends IPacket> implements LpMap<V> {
         heapMap.clear();
     }
 
-    private void load() {
-        var maxIndex = fileChannelMap.getMaxIndex();
-        if (maxIndex <= 0) {
-            return;
-        }
-
-        for (var key = 1; key <= maxIndex; key++) {
-            var value = fileChannelMap.get(key);
-            if (value == null) {
-                continue;
-            }
-            heapMap.put(key, value);
-        }
+    @Override
+    public void close() throws IOException {
+        fileChannelMap.close();
+        heapMap.clear();
     }
 
     @Override
     public void forEach(BiConsumer<Long, V> biConsumer) {
         heapMap.forEach(biConsumer);
     }
+
 }

orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java

@@ -193,6 +193,12 @@ public class FileChannelMap<V extends IPacket> implements LpMap<V>, Closeable {
 
     }
 
+    @Override
+    public void close() throws IOException {
+        IOUtils.closeIO(indexFileRandomAccess, indexFileChannel, dbFileRandomAccess, dbFileChannel);
+        ReferenceCountUtil.release(indexBuffer);
+        ReferenceCountUtil.release(dbBuffer);
+    }
 
     protected void setKeyValue(long key, V value) {
         try {
@@ -236,11 +242,4 @@ public class FileChannelMap<V extends IPacket> implements LpMap<V>, Closeable {
         dbBuffer.clear();
     }
 
-    @Override
-    public void close() throws IOException {
-        IOUtils.closeIO(indexFileRandomAccess, indexFileChannel, dbFileRandomAccess, dbFileChannel);
-        ReferenceCountUtil.release(indexBuffer);
-        ReferenceCountUtil.release(dbBuffer);
-    }
-
 }

orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java

@@ -43,13 +43,13 @@ public class FileHeapMap<V extends IPacket> implements LpMap<V> {
 
     private HeapMap<V> heapMap;
 
-    public FileHeapMap(String dbPath, int initialCapacity, Class<V> clazz) {
+    public FileHeapMap(String dbPath, Class<V> clazz) {
         try {
             this.dbFile = FileUtils.getOrCreateFile(dbPath, StringUtils.format("{}.db", clazz.getSimpleName()));
 
             var protocolId = ProtocolAnalysis.getProtocolIdByClass(clazz);
             protocolRegistration = ProtocolManager.getProtocol(protocolId);
-            heapMap = new HeapMap<>(initialCapacity);
+            heapMap = new HeapMap<>();
 
             load();
         } catch (IOException e) {

orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java

@@ -27,6 +27,10 @@ public class HeapMap<V extends IPacket> implements LpMap<V> {
 
     protected long maxIndex = 0;
 
+    public HeapMap() {
+        this(128);
+    }
+
     public HeapMap(int initialCapacity) {
         map = new LongObjectHashMap<>(initialCapacity);
     }

orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMapTest.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the License.
+ */
+
+package com.zfoo.orm.lpmap;
+
+import com.zfoo.event.manager.EventBus;
+import com.zfoo.orm.lpmap.model.MyPacket;
+import com.zfoo.protocol.ProtocolManager;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author jaysunxiao
+ * @version 3.0
+ */
+public class ConcurrentFileChannelMapTest {
+
+    @Test
+    public void benchmarkTest() throws IOException, InterruptedException {
+        ProtocolManager.initProtocol(Set.of(MyPacket.class));
+
+        var map = new ConcurrentFileChannelMap<MyPacket>("db", MyPacket.class);
+        var atomicInt = new AtomicInteger(0);
+        var count = 1000_0000;
+
+        var countdown = new CountDownLatch(EventBus.EXECUTORS_SIZE);
+        for (int i = 0; i < EventBus.EXECUTORS_SIZE; i++) {
+            EventBus.asyncExecute().execute(new Runnable() {
+                @Override
+                public void run() {
+                    var key = atomicInt.getAndIncrement();
+                    while (key < count) {
+                        var myPacket = MyPacket.valueOf(key, String.valueOf(key));
+                        map.put(key, myPacket);
+                        key = atomicInt.getAndIncrement();
+                    }
+                    countdown.countDown();
+                }
+            });
+        }
+        countdown.await();
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            var packet = map.get(i);
+            Assert.assertEquals(myPacket, packet);
+        }
+
+        map.close();
+        var newMap = new ConcurrentFileChannelMap<MyPacket>("db", MyPacket.class);
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            var packet = newMap.get(i);
+            Assert.assertEquals(myPacket, packet);
+        }
+    }
+}

orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentHeapMapTest.java

@@ -12,12 +12,16 @@
 
 package com.zfoo.orm.lpmap;
 
+import com.zfoo.event.manager.EventBus;
 import com.zfoo.orm.lpmap.model.MyPacket;
 import com.zfoo.protocol.ProtocolManager;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.io.IOException;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author jaysunxiao
@@ -42,4 +46,34 @@ public class ConcurrentHeapMapTest {
         Assert.assertEquals(previous3, myPacket);
     }
 
+    @Test
+    public void benchmarkTest() throws IOException, InterruptedException {
+        ProtocolManager.initProtocol(Set.of(MyPacket.class));
+
+        var map = new ConcurrentHeapMap<MyPacket>();
+        var atomicInt = new AtomicInteger(0);
+        var count = 1000_0000;
+
+        var countdown = new CountDownLatch(EventBus.EXECUTORS_SIZE);
+        for (int i = 0; i < EventBus.EXECUTORS_SIZE; i++) {
+            EventBus.asyncExecute().execute(new Runnable() {
+                @Override
+                public void run() {
+                    var key = atomicInt.getAndIncrement();
+                    while (key < count) {
+                        var myPacket = MyPacket.valueOf(key, String.valueOf(key));
+                        map.put(key, myPacket);
+                        key = atomicInt.getAndIncrement();
+                    }
+                    countdown.countDown();
+                }
+            });
+        }
+        countdown.await();
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            var packet = map.get(i);
+            Assert.assertEquals(myPacket, packet);
+        }
+    }
 }

orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java

@@ -14,9 +14,11 @@ package com.zfoo.orm.lpmap;
 
 import com.zfoo.orm.lpmap.model.MyPacket;
 import com.zfoo.protocol.ProtocolManager;
+import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
 
+import java.io.IOException;
 import java.util.Set;
 
 /**
@@ -61,4 +63,29 @@ public class FileChannelMapTest {
         System.out.println(map.get(2));
         System.out.println(map.get(3));
     }
+
+    @Test
+    public void benchmarkTest() throws IOException {
+        ProtocolManager.initProtocol(Set.of(MyPacket.class));
+
+        var map = new FileChannelMap<MyPacket>("db", MyPacket.class);
+        var count = 1000_0000;
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            map.put(i, myPacket);
+        }
+
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            var packet = map.get(i);
+            Assert.assertEquals(myPacket, packet);
+        }
+        map.close();
+        map = new FileChannelMap<MyPacket>("db", MyPacket.class);
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            var packet = map.get(i);
+            Assert.assertEquals(myPacket, packet);
+        }
+    }
 }

orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java

@@ -31,7 +31,7 @@ public class FileHeapMapTest {
     public void test() {
         ProtocolManager.initProtocol(Set.of(MyPacket.class));
 
-        var map = new FileHeapMap<MyPacket>("db", 10, MyPacket.class);
+        var map = new FileHeapMap<MyPacket>("db", MyPacket.class);
         var myPacket = new MyPacket();
         myPacket.setA(9999);
 
@@ -55,5 +55,36 @@ public class FileHeapMapTest {
         map.save();
     }
 
+    @Test
+    public void benchmarkTest() {
+        ProtocolManager.initProtocol(Set.of(MyPacket.class));
 
+        var map = new FileHeapMap<MyPacket>("db", MyPacket.class);
+        var count = 1000_0000;
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            map.put(i, myPacket);
+        }
+
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            var packet = map.get(i);
+            Assert.assertEquals(myPacket, packet);
+        }
+        map.save();
+    }
+
+    @Test
+    public void loadTest() {
+        ProtocolManager.initProtocol(Set.of(MyPacket.class));
+
+        var map = new FileHeapMap<MyPacket>("db", MyPacket.class);
+        var count = 1000_0000;
+
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            var packet = map.get(i);
+            Assert.assertEquals(myPacket, packet);
+        }
+    }
 }

orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java

@@ -26,7 +26,7 @@ public class HeapMapTest {
 
     @Test
     public void test() {
-        var map = new HeapMap<MyPacket>(10);
+        var map = new HeapMap<MyPacket>();
         var myPacket = new MyPacket();
 
         var packet = map.put(0, myPacket);
@@ -42,4 +42,21 @@ public class HeapMapTest {
         Assert.assertEquals(packet, myPacket);
     }
 
+    @Test
+    public void benchmarkTest() {
+        var map = new HeapMap<MyPacket>();
+
+        var count = 1000_0000;
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            map.put(i, myPacket);
+        }
+
+        for (var i = 0; i < count; i++) {
+            var myPacket = MyPacket.valueOf(i, String.valueOf(i));
+            var packet = map.get(i);
+            Assert.assertEquals(myPacket, packet);
+        }
+    }
+
 }

orm/src/test/java/com/zfoo/orm/lpmap/model/MyPacket.java

@@ -14,6 +14,8 @@ package com.zfoo.orm.lpmap.model;
 
 import com.zfoo.protocol.IPacket;
 
+import java.util.Objects;
+
 /**
  * @author jaysunxiao
  * @version 3.0
@@ -24,11 +26,42 @@ public class MyPacket implements IPacket {
 
     private int a;
 
+    private String b;
+
+    public static MyPacket valueOf(int a, String b) {
+        var packet = new MyPacket();
+        packet.a = a;
+        packet.b = b;
+        return packet;
+    }
+
     @Override
     public short protocolId() {
         return PROTOCOL_ID;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        MyPacket myPacket = (MyPacket) o;
+        return a == myPacket.a && Objects.equals(b, myPacket.b);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(a, b);
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("MyPacket{");
+        sb.append("a=").append(a);
+        sb.append(", b='").append(b).append('\'');
+        sb.append('}');
+        return sb.toString();
+    }
+
     public int getA() {
         return a;
     }
@@ -36,4 +69,12 @@ public class MyPacket implements IPacket {
     public void setA(int a) {
         this.a = a;
     }
+
+    public String getB() {
+        return b;
+    }
+
+    public void setB(String b) {
+        this.b = b;
+    }
 }