tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-18 08:25:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf[orm]: 取消FileChannelHeap的insert接口

Browse Source
This commit is contained in:
jaysunxiao
2021-08-18 18:49:37 +08:00
parent 64a67b414b
commit 40d91369d9
10 changed files with 155 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java
+95
View File
@@ -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<V extends IPacket> implements LpMap<V> {
private ReentrantLock fileChannelLock = new ReentrantLock();
private FileChannelMap<V> fileChannelMap;
private ConcurrentHeapMap<V> concurrentHeapMap;
public ConcurrentFileChannelHeapMap(String dbPath, int initialCapacity, Class<V> 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<Long, V> biConsumer) {
concurrentHeapMap.forEach(biConsumer);
}
}
orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java
+16 -6
View File
@@ -28,16 +28,21 @@ public class ConcurrentHeapMap<V extends IPacket> implements LpMap<V> {
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<V extends IPacket> implements LpMap<V> {
public long getMaxIndex() {
return maxIndexAtomic.get();
}
@Override
public long getIncrementIndex() {
return maxIndexAtomic.incrementAndGet();
}
}
orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java
+5 -7
View File
@@ -33,13 +33,6 @@ public class FileChannelHeapMap<V extends IPacket> implements LpMap<V> {
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<V extends IPacket> implements LpMap<V> {
return heapMap.getMaxIndex();
}
@Override
public long getIncrementIndex() {
return heapMap.getIncrementIndex();
}
private void load() {
var maxIndex = fileChannelMap.getMaxIndex();
if (maxIndex <= 0) {
orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java
+9 -11
View File
@@ -75,17 +75,6 @@ public class FileChannelMap<V extends IPacket> implements LpMap<V>, 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<V extends IPacket> implements LpMap<V>, 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();
orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java
+9 -5
View File
@@ -57,11 +57,6 @@ public class FileHeapMap<V extends IPacket> implements LpMap<V> {
}
@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<V extends IPacket> implements LpMap<V> {
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<V extends IPacket> implements LpMap<V> {
fileInputStream = FileUtils.openInputStream(dbFile);
fileChannel = fileInputStream.getChannel();
if (fileChannel.size() <= 0) {
return;
}
buffer = ByteBufAllocator.DEFAULT.ioBuffer(1000);
buffer.writeBytes(fileChannel, 0L, (int) dbFile.length());
orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java
+12 -30
View File
@@ -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<V extends IPacket> implements LpMap<V> {
protected LongObjectHashMap<V> map;
/**
* 没有被使用的key
*/
protected Queue<Long> 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<V extends IPacket> implements LpMap<V> {
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<V extends IPacket> implements LpMap<V> {
return maxIndex;
}
@Override
public long getIncrementIndex() {
maxIndex++;
return maxIndex;
}
@Override
public void forEach(BiConsumer<Long, V> biConsumer) {
map.forEach(biConsumer);
orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java
+3 -6
View File
@@ -26,11 +26,6 @@ import java.util.function.BiConsumer;
*/
public interface LpMap<V extends IPacket> {
/**
* 插入一条数据，返回一个自增的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<V extends IPacket> {
long getMaxIndex();
long getIncrementIndex();
default void checkKey(long key) {
if (key <= 0) {
throw new RunException("key[{}]不能为负数", key);
throw new RunException("key[{}]不能为负数或0", key);
}
}
orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java
+3 -3
View File
@@ -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
orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java
+2 -17
View File
@@ -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<MyPacket>("tc", 10, MyPacket.class);
Assert.assertNotNull(map.get(5));
}
}
orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java
+1 -10
View File
@@ -29,26 +29,17 @@ public class HeapMapTest {
var map = new HeapMap<MyPacket>(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);
}
}