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

feat[lpmap]: 增加FileChannelHeapMap

Browse Source
This commit is contained in:
jaysunxiao
2021-08-17 17:48:14 +08:00
parent 91cfd38052
commit bad0a3f88c
8 changed files with 415 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java
+6 -6
View File
@@ -26,19 +26,19 @@ public class ConcurrentHeapMap<V extends IPacket> implements LpMap<V> {
private ConcurrentNavigableMap<Long, V> map = new ConcurrentSkipListMap<>();
private AtomicLong index = new AtomicLong(0);
private AtomicLong maxIndexAtomic = new AtomicLong(0);
@Override
public long insert(V packet) {
var key = index.incrementAndGet();
map.put(key, packet);
public long insert(V value) {
var key = maxIndexAtomic.incrementAndGet();
map.put(key, value);
return key;
}
@Override
public V put(long key, V packet) {
public V put(long key, V value) {
checkKey(key);
return map.put(key, packet);
return map.put(key, value);
}
@Override
orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java
+74
View File
@@ -0,0 +1,74 @@
/*
* 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;
/**
* @author jaysunxiao
* @version 3.0
*/
public class FileChannelHeapMap<V extends IPacket> implements LpMap<V> {
private FileChannelMap<V> fileChannelMap;
private HeapMap<V> heapMap;
public FileChannelHeapMap(String dbPath, int initialCapacity, Class<V> clazz) {
fileChannelMap = new FileChannelMap<>(dbPath, clazz);
heapMap = new HeapMap<>(initialCapacity);
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);
return heapMap.put(key, value);
}
@Override
public V delete(long key) {
fileChannelMap.delete(key);
return heapMap.delete(key);
}
@Override
public V get(long key) {
return heapMap.get(key);
}
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);
}
}
}
orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java
+221
View File
@@ -0,0 +1,221 @@
/*
* 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 com.zfoo.protocol.ProtocolManager;
import com.zfoo.protocol.registration.IProtocolRegistration;
import com.zfoo.protocol.registration.ProtocolAnalysis;
import com.zfoo.protocol.util.FileUtils;
import com.zfoo.protocol.util.IOUtils;
import com.zfoo.protocol.util.StringUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.util.ReferenceCountUtil;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;
/**
* @author jaysunxiao
* @version 3.0
*/
public class FileChannelMap<V extends IPacket> implements LpMap<V>, Closeable {
private File dbFile;
protected RandomAccessFile dbFileRandomAccess;
protected FileChannel dbFileChannel;
private File indexFile;
protected RandomAccessFile indexFileRandomAccess;
protected FileChannel indexFileChannel;
protected IProtocolRegistration protocolRegistration;
protected ByteBuf indexBuffer;
protected ByteBuf dbBuffer;
public FileChannelMap(String dbPath, Class<V> clazz) {
try {
this.dbFile = FileUtils.getOrCreateFile(dbPath, StringUtils.format("{}.db", clazz.getSimpleName()));
this.dbFileRandomAccess = new RandomAccessFile(dbFile, "rw");
this.dbFileChannel = this.dbFileRandomAccess.getChannel();
this.indexFile = FileUtils.getOrCreateFile(dbPath, StringUtils.format("{}.index", clazz.getSimpleName()));
this.indexFileRandomAccess = new RandomAccessFile(indexFile, "rw");
this.indexFileChannel = this.indexFileRandomAccess.getChannel();
if (indexFileChannel.size() <= 0) {
indexFileRandomAccess.writeLong(0L);
indexFileRandomAccess.writeLong(0L);
}
var protocolId = ProtocolAnalysis.getProtocolIdByClass(clazz);
protocolRegistration = ProtocolManager.getProtocol(protocolId);
indexBuffer = ByteBufAllocator.DEFAULT.ioBuffer(16);
dbBuffer = ByteBufAllocator.DEFAULT.ioBuffer(100);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
@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);
var maxIndex = getMaxIndex();
V previousValue = null;
if (key <= maxIndex) {
previousValue = get(key);
} else {
for (var i = maxIndex + 1; i < key; i++) {
resetKey(i);
}
maxIndex = key;
setMaxIndex(maxIndex);
}
setKeyValue(key, packet);
return previousValue;
}
@Override
public V delete(long key) {
checkKey(key);
var maxIndex = getMaxIndex();
if (key <= maxIndex) {
var previousValue = get(key);
resetKey(key);
return previousValue;
} else {
return null;
}
}
@Override
public V get(long key) {
checkKey(key);
var maxIndex = getMaxIndex();
if (key > maxIndex) {
return null;
}
try {
indexBuffer.writeBytes(indexFileChannel, key * 16L, 16);
var packetPosition = indexBuffer.readLong();
var packetSize = indexBuffer.readLong();
if (packetSize <= 0) {
return null;
}
dbBuffer.writeBytes(dbFileChannel, packetPosition, (int) packetSize);
var packet = protocolRegistration.read(dbBuffer);
return (V) packet;
} catch (Exception e) {
return null;
}
}
protected void setKeyValue(long key, V value) {
try {
clearByteBuf();
protocolRegistration.write(dbBuffer, value);
// db文件
var packetPosition = dbFileChannel.size();
// db文件数据的起始位置
indexBuffer.writeLong(packetPosition);
// db文件的值的大小
indexBuffer.writeLong(dbBuffer.readableBytes());
indexFileChannel.write(indexBuffer.nioBuffer(), key * 16);
dbFileChannel.write(dbBuffer.nioBuffer(), packetPosition);
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
clearByteBuf();
}
}
protected void resetKey(long key) {
try {
clearByteBuf();
indexBuffer.writeLong(0L);
indexBuffer.writeLong(0L);
indexFileChannel.write(indexBuffer.nioBuffer(), key * 16);
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
clearByteBuf();
}
}
protected void setMaxIndex(long maxIndex) {
try {
clearByteBuf();
indexBuffer.writeLong(maxIndex);
indexFileChannel.write(indexBuffer.nioBuffer(), 0);
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
clearByteBuf();
}
}
protected long getMaxIndex() {
try {
clearByteBuf();
indexBuffer.writeBytes(indexFileChannel, 0, 8);
return indexBuffer.readLong();
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
clearByteBuf();
}
}
protected void clearByteBuf() {
indexBuffer.clear();
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
+22 -14
View File
@@ -19,6 +19,7 @@ import com.zfoo.protocol.registration.IProtocolRegistration;
import com.zfoo.protocol.registration.ProtocolAnalysis;
import com.zfoo.protocol.util.FileUtils;
import com.zfoo.protocol.util.IOUtils;
import com.zfoo.protocol.util.StringUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.util.ReferenceCountUtil;
@@ -35,30 +36,35 @@ import java.nio.channels.FileChannel;
*/
public class FileHeapMap<V extends IPacket> implements LpMap<V> {
private File file;
private File dbFile;
private IProtocolRegistration protocolRegistration;
private HeapMap<V> heapMap;
public FileHeapMap(File file, int initialCapacity, Class<V> clazz) {
this.file = file;
var protocolId = ProtocolAnalysis.getProtocolIdByClass(clazz);
protocolRegistration = ProtocolManager.getProtocol(protocolId);
heapMap = new HeapMap<>(initialCapacity);
public FileHeapMap(String dbPath, int initialCapacity, Class<V> clazz) {
try {
this.dbFile = FileUtils.getOrCreateFile(dbPath, StringUtils.format("{}.db", clazz.getSimpleName()));
load();
var protocolId = ProtocolAnalysis.getProtocolIdByClass(clazz);
protocolRegistration = ProtocolManager.getProtocol(protocolId);
heapMap = new HeapMap<>(initialCapacity);
load();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public long insert(V packet) {
return heapMap.insert(packet);
public long insert(V value) {
return heapMap.insert(value);
}
@Override
public V put(long key, V packet) {
return heapMap.put(key, packet);
public V put(long key, V value) {
return heapMap.put(key, value);
}
@Override
@@ -76,11 +82,11 @@ public class FileHeapMap<V extends IPacket> implements LpMap<V> {
FileChannel fileChannel = null;
ByteBuf buffer = null;
try {
fileInputStream = FileUtils.openInputStream(file);
fileInputStream = FileUtils.openInputStream(dbFile);
fileChannel = fileInputStream.getChannel();
buffer = ByteBufAllocator.DEFAULT.ioBuffer(1000);
buffer.writeBytes(fileChannel, 0L, (int) file.length());
buffer.writeBytes(fileChannel, 0L, (int) dbFile.length());
var size = ByteBufUtils.readLong(buffer);
for (var i = 0; i < size; i++) {
@@ -89,6 +95,8 @@ public class FileHeapMap<V extends IPacket> implements LpMap<V> {
put(key, value);
}
} catch (IOException e) {
throw new RuntimeException(e);
} finally {
IOUtils.closeIO(fileChannel, fileInputStream);
ReferenceCountUtil.release(buffer);
}
@@ -98,7 +106,7 @@ public class FileHeapMap<V extends IPacket> implements LpMap<V> {
FileOutputStream fileOutputStream = null;
ByteBuf buffer = null;
try {
fileOutputStream = FileUtils.openOutputStream(file, false);
fileOutputStream = FileUtils.openOutputStream(dbFile, false);
buffer = ByteBufAllocator.DEFAULT.heapBuffer(1000);
// 写入长度
orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java
+13 -13
View File
@@ -29,37 +29,37 @@ public class HeapMap<V extends IPacket> implements LpMap<V> {
/**
* 没有被使用的key
*/
private Queue<Long> freeKeyQueue = new LinkedList<>();
protected Queue<Long> freeKeyQueue = new LinkedList<>();
private long index = 0;
protected long maxIndex = 0;
public HeapMap(int initialCapacity) {
map = new LongObjectHashMap<>(initialCapacity);
}
@Override
public long insert(V packet) {
public long insert(V value) {
if (freeKeyQueue.isEmpty()) {
map.put(++index, packet);
return index;
map.put(++maxIndex, value);
return maxIndex;
} else {
var freeKey = freeKeyQueue.poll();
map.put(freeKey, packet);
map.put(freeKey, value);
return freeKey;
}
}
@Override
public V put(long key, V packet) {
public V put(long key, V value) {
checkKey(key);
if (key <= index) {
return map.put(key, packet);
if (key <= maxIndex) {
return map.put(key, value);
} else {
for (var i = index + 1; i < key; i++) {
for (var i = maxIndex + 1; i < key; i++) {
freeKeyQueue.add(i);
}
index = key;
map.put(key, packet);
maxIndex = key;
map.put(key, value);
return null;
}
}
@@ -67,7 +67,7 @@ public class HeapMap<V extends IPacket> implements LpMap<V> {
@Override
public V delete(long key) {
checkKey(key);
if (key > index) {
if (key > maxIndex) {
return null;
} else {
var previousValue = map.remove(key);
orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java
+63
View File
@@ -0,0 +1,63 @@
/*
* 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.orm.lpmap.model.MyPacket;
import com.zfoo.protocol.ProtocolManager;
import org.junit.Ignore;
import org.junit.Test;
import java.util.Set;
/**
* @author jaysunxiao
* @version 3.0
*/
@Ignore
public class FileChannelMapTest {
@Test
public void test() {
ProtocolManager.initProtocol(Set.of(MyPacket.class));
var map = new FileChannelMap<MyPacket>("db", MyPacket.class);
var myPacket = new MyPacket();
myPacket.setA(9999);
map.insert(myPacket);
map.insert(myPacket);
map.insert(myPacket);
}
@Test
public void readTest() {
ProtocolManager.initProtocol(Set.of(MyPacket.class));
var map = new FileChannelMap<MyPacket>("db", MyPacket.class);
System.out.println(map.get(1));
System.out.println(map.get(2));
System.out.println(map.get(3));
}
@Test
public void channelHeapTest() {
ProtocolManager.initProtocol(Set.of(MyPacket.class));
var map = new FileChannelHeapMap<MyPacket>("db", 1000, MyPacket.class);
System.out.println(map.get(1));
System.out.println(map.get(2));
System.out.println(map.get(3));
}
}
orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java
+2 -3
View File
@@ -18,7 +18,6 @@ import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import java.io.File;
import java.util.Set;
/**
@@ -32,7 +31,7 @@ public class FileHeapMapTest {
public void test() {
ProtocolManager.initProtocol(Set.of(MyPacket.class));
var map = new FileHeapMap<MyPacket>(new File("myPacket.db"), 10, MyPacket.class);
var map = new FileHeapMap<MyPacket>("db", 10, MyPacket.class);
var myPacket = new MyPacket();
myPacket.setA(9999);
@@ -69,7 +68,7 @@ public class FileHeapMapTest {
@Test
public void readTest() {
ProtocolManager.initProtocol(Set.of(MyPacket.class));
var map = new FileHeapMap<MyPacket>(new File("myPacket.db"), 10, MyPacket.class);
var map = new FileHeapMap<MyPacket>("tc", 10, MyPacket.class);
Assert.assertNotNull(map.get(5));
}
}
protocol/src/main/java/com/zfoo/protocol/util/FileUtils.java
+14
View File
@@ -169,6 +169,20 @@ public abstract class FileUtils {
return newFile;
}
public static File getOrCreateFile(String path, String fileName) throws IOException {
var file = createDirectory(path);
var newFile = new File(file.getAbsoluteFile() + File.separator + fileName);
if (newFile.exists()) {
return newFile;
}
if (!newFile.createNewFile()) {
throw new RuntimeException(StringUtils.format("创建文件[fileName:{}]失败", fileName));
}
return newFile;
}
public static File createDirectory(String path) {
var file = new File(path);
if (!file.exists()) {