From 40d91369d9322f2f41a5c2de3957b418d8073c94 Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Wed, 18 Aug 2021 18:49:37 +0800 Subject: [PATCH] =?UTF-8?q?perf[orm]:=20=E5=8F=96=E6=B6=88FileChannelHeap?= =?UTF-8?q?=E7=9A=84insert=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../lpmap/ConcurrentFileChannelHeapMap.java | 95 +++++++++++++++++++ .../com/zfoo/orm/lpmap/ConcurrentHeapMap.java | 22 +++-- .../zfoo/orm/lpmap/FileChannelHeapMap.java | 12 +-- .../com/zfoo/orm/lpmap/FileChannelMap.java | 20 ++-- .../java/com/zfoo/orm/lpmap/FileHeapMap.java | 14 ++- .../main/java/com/zfoo/orm/lpmap/HeapMap.java | 42 +++----- .../main/java/com/zfoo/orm/lpmap/LpMap.java | 9 +- .../zfoo/orm/lpmap/FileChannelMapTest.java | 6 +- .../com/zfoo/orm/lpmap/FileHeapMapTest.java | 19 +--- .../java/com/zfoo/orm/lpmap/HeapMapTest.java | 11 +-- 10 files changed, 155 insertions(+), 95 deletions(-) create mode 100644 orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java new file mode 100644 index 00000000..b36d1017 --- /dev/null +++ b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java @@ -0,0 +1,95 @@ +/* + * 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.protocol.IPacket; + +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.BiConsumer; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class ConcurrentFileChannelHeapMap implements LpMap { + + private ReentrantLock fileChannelLock = new ReentrantLock(); + + private FileChannelMap fileChannelMap; + + private ConcurrentHeapMap concurrentHeapMap; + + public ConcurrentFileChannelHeapMap(String dbPath, int initialCapacity, Class clazz) { + fileChannelMap = new FileChannelMap<>(dbPath, clazz); + concurrentHeapMap = new ConcurrentHeapMap<>(); + + load(); + } + + @Override + public V put(long key, V value) { + fileChannelLock.lock(); + try { + fileChannelMap.put(key, value); + } finally { + fileChannelLock.unlock(); + } + return concurrentHeapMap.put(key, value); + } + + @Override + public V delete(long key) { + fileChannelLock.lock(); + try { + fileChannelMap.delete(key); + } finally { + fileChannelLock.unlock(); + } + return concurrentHeapMap.delete(key); + } + + @Override + public V get(long key) { + return concurrentHeapMap.get(key); + } + + @Override + public long getMaxIndex() { + return concurrentHeapMap.getMaxIndex(); + } + + @Override + public long getIncrementIndex() { + return concurrentHeapMap.getIncrementIndex(); + } + + 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 forEach(BiConsumer biConsumer) { + concurrentHeapMap.forEach(biConsumer); + } +} diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java index 3a3d9bd2..825edee4 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java @@ -28,16 +28,21 @@ public class ConcurrentHeapMap implements LpMap { private AtomicLong maxIndexAtomic = new AtomicLong(0); - @Override - public long insert(V value) { - var key = maxIndexAtomic.incrementAndGet(); - map.put(key, value); - return key; - } @Override public V put(long key, V value) { checkKey(key); + + while (true) { + var maxIndex = maxIndexAtomic.get(); + + if (key <= maxIndex) { + break; + } + + maxIndexAtomic.compareAndExchange(maxIndex, key); + } + return map.put(key, value); } @@ -57,4 +62,9 @@ public class ConcurrentHeapMap implements LpMap { public long getMaxIndex() { return maxIndexAtomic.get(); } + + @Override + public long getIncrementIndex() { + return maxIndexAtomic.incrementAndGet(); + } } diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java index e542d860..27b1066e 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java @@ -33,13 +33,6 @@ public class FileChannelHeapMap implements LpMap { load(); } - @Override - public long insert(V value) { - var key = fileChannelMap.insert(value); - heapMap.put(key, value); - return key; - } - @Override public V put(long key, V value) { fileChannelMap.put(key, value); @@ -62,6 +55,11 @@ public class FileChannelHeapMap implements LpMap { return heapMap.getMaxIndex(); } + @Override + public long getIncrementIndex() { + return heapMap.getIncrementIndex(); + } + private void load() { var maxIndex = fileChannelMap.getMaxIndex(); if (maxIndex <= 0) { diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java index 8349a22d..199334c1 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java @@ -75,17 +75,6 @@ public class FileChannelMap implements LpMap, Closeable { } - @Override - public long insert(V value) { - var maxIndex = getMaxIndex() + 1; - - // index索引文件的头16个字节是当前index的大小 - setMaxIndex(maxIndex); - - setKeyValue(maxIndex, value); - return maxIndex; - } - @Override public V put(long key, V packet) { checkKey(key); @@ -158,6 +147,15 @@ public class FileChannelMap implements LpMap, Closeable { } } + @Override + public long getIncrementIndex() { + var maxIndex = getMaxIndex() + 1; + + // index索引文件的头16个字节是当前index的大小 + setMaxIndex(maxIndex); + return maxIndex; + } + protected void setKeyValue(long key, V value) { try { clearByteBuf(); diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java index 2fdf9764..321c0894 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java @@ -57,11 +57,6 @@ public class FileHeapMap implements LpMap { } - @Override - public long insert(V value) { - return heapMap.insert(value); - } - @Override public V put(long key, V value) { return heapMap.put(key, value); @@ -82,6 +77,11 @@ public class FileHeapMap implements LpMap { return heapMap.getMaxIndex(); } + @Override + public long getIncrementIndex() { + return heapMap.getIncrementIndex(); + } + private void load() { FileInputStream fileInputStream = null; FileChannel fileChannel = null; @@ -90,6 +90,10 @@ public class FileHeapMap implements LpMap { fileInputStream = FileUtils.openInputStream(dbFile); fileChannel = fileInputStream.getChannel(); + if (fileChannel.size() <= 0) { + return; + } + buffer = ByteBufAllocator.DEFAULT.ioBuffer(1000); buffer.writeBytes(fileChannel, 0L, (int) dbFile.length()); diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java index 918f692d..6589d73b 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java @@ -15,8 +15,6 @@ package com.zfoo.orm.lpmap; import com.zfoo.protocol.IPacket; import io.netty.util.collection.LongObjectHashMap; -import java.util.LinkedList; -import java.util.Queue; import java.util.function.BiConsumer; /** @@ -27,42 +25,22 @@ public class HeapMap implements LpMap { protected LongObjectHashMap map; - /** - * 没有被使用的key - */ - protected Queue freeKeyQueue = new LinkedList<>(); - protected long maxIndex = 0; public HeapMap(int initialCapacity) { map = new LongObjectHashMap<>(initialCapacity); } - @Override - public long insert(V value) { - if (freeKeyQueue.isEmpty()) { - map.put(++maxIndex, value); - return maxIndex; - } else { - var freeKey = freeKeyQueue.poll(); - map.put(freeKey, value); - return freeKey; - } - } @Override public V put(long key, V value) { checkKey(key); - if (key <= maxIndex) { - return map.put(key, value); - } else { - for (var i = maxIndex + 1; i < key; i++) { - freeKeyQueue.add(i); - } + + if (key > maxIndex) { maxIndex = key; - map.put(key, value); - return null; } + + return map.put(key, value); } @Override @@ -70,11 +48,9 @@ public class HeapMap implements LpMap { checkKey(key); if (key > maxIndex) { return null; - } else { - var previousValue = map.remove(key); - freeKeyQueue.add(key); - return previousValue; } + + return map.remove(key); } @Override @@ -88,6 +64,12 @@ public class HeapMap implements LpMap { return maxIndex; } + @Override + public long getIncrementIndex() { + maxIndex++; + return maxIndex; + } + @Override public void forEach(BiConsumer biConsumer) { map.forEach(biConsumer); diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java index a3db838b..52bc1890 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java @@ -26,11 +26,6 @@ import java.util.function.BiConsumer; */ public interface LpMap { - /** - * 插入一条数据，返回一个自增的key，效率比较高 - */ - long insert(V packet); - /** * @param packet the previous value associated with key, or null if there was no mapping for key. */ @@ -45,9 +40,11 @@ public interface LpMap { long getMaxIndex(); + long getIncrementIndex(); + default void checkKey(long key) { if (key <= 0) { - throw new RunException("key[{}]不能为负数", key); + throw new RunException("key[{}]不能为负数或0", key); } } diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java b/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java index d72965db..0770e623 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java @@ -34,9 +34,9 @@ public class FileChannelMapTest { var myPacket = new MyPacket(); myPacket.setA(9999); - map.insert(myPacket); - map.insert(myPacket); - map.insert(myPacket); + map.put(1, myPacket); + map.put(2, myPacket); + map.put(3, myPacket); } @Test diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java b/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java index f9d8890f..1e7a9e9f 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java @@ -35,24 +35,15 @@ public class FileHeapMapTest { var myPacket = new MyPacket(); myPacket.setA(9999); - var key = map.insert(myPacket); - Assert.assertEquals(key, 1L); - var packet = map.put(1, myPacket); - Assert.assertEquals(packet, myPacket); - - packet = map.put(3, myPacket); Assert.assertNull(packet); - key = map.insert(myPacket); - Assert.assertEquals(key, 2L); + packet = map.put(2, myPacket); + Assert.assertNull(packet); packet = map.delete(4); Assert.assertNull(packet); - packet = map.delete(3); - Assert.assertEquals(packet, myPacket); - packet = map.delete(2); Assert.assertEquals(packet, myPacket); @@ -65,10 +56,4 @@ public class FileHeapMapTest { } - @Test - public void readTest() { - ProtocolManager.initProtocol(Set.of(MyPacket.class)); - var map = new FileHeapMap("tc", 10, MyPacket.class); - Assert.assertNotNull(map.get(5)); - } } diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java b/orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java index 46a9c058..d5d75da0 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java @@ -29,26 +29,17 @@ public class HeapMapTest { var map = new HeapMap(10); var myPacket = new MyPacket(); - var key = map.insert(myPacket); - Assert.assertEquals(key, 1L); - var packet = map.put(1, myPacket); - Assert.assertEquals(packet, myPacket); + Assert.assertNull(packet); packet = map.put(3, myPacket); Assert.assertNull(packet); - key = map.insert(myPacket); - Assert.assertEquals(key, 2L); - packet = map.delete(4); Assert.assertNull(packet); packet = map.delete(3); Assert.assertEquals(packet, myPacket); - - packet = map.delete(2); - Assert.assertEquals(packet, myPacket); } }