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 0a76d56b..ff1dbb2c 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java @@ -26,19 +26,19 @@ public class ConcurrentHeapMap implements LpMap { private ConcurrentNavigableMap 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 diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java new file mode 100644 index 00000000..f00f8fa9 --- /dev/null +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java @@ -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 implements LpMap { + + private FileChannelMap fileChannelMap; + + private HeapMap heapMap; + + public FileChannelHeapMap(String dbPath, int initialCapacity, Class 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); + } + + } + +} diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java new file mode 100644 index 00000000..6e3f7723 --- /dev/null +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java @@ -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 implements LpMap, 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 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); + } + +} 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 0a5d32b0..6feb52a9 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java @@ -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 implements LpMap { - private File file; + private File dbFile; private IProtocolRegistration protocolRegistration; private HeapMap heapMap; - public FileHeapMap(File file, int initialCapacity, Class clazz) { - this.file = file; - var protocolId = ProtocolAnalysis.getProtocolIdByClass(clazz); - protocolRegistration = ProtocolManager.getProtocol(protocolId); - heapMap = new HeapMap<>(initialCapacity); + public FileHeapMap(String dbPath, int initialCapacity, Class 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 implements LpMap { 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 implements LpMap { 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 implements LpMap { FileOutputStream fileOutputStream = null; ByteBuf buffer = null; try { - fileOutputStream = FileUtils.openOutputStream(file, false); + fileOutputStream = FileUtils.openOutputStream(dbFile, false); buffer = ByteBufAllocator.DEFAULT.heapBuffer(1000); // 写入长度 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 6c0d460b..ebf2a37b 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java @@ -29,37 +29,37 @@ public class HeapMap implements LpMap { /** * 没有被使用的key */ - private Queue freeKeyQueue = new LinkedList<>(); + protected Queue 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 implements LpMap { @Override public V delete(long key) { checkKey(key); - if (key > index) { + if (key > maxIndex) { return null; } else { var previousValue = map.remove(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 new file mode 100644 index 00000000..d72965db --- /dev/null +++ b/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java @@ -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("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("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("db", 1000, MyPacket.class); + + System.out.println(map.get(1)); + System.out.println(map.get(2)); + System.out.println(map.get(3)); + } +} 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 09ea41ad..f9d8890f 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java @@ -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(new File("myPacket.db"), 10, MyPacket.class); + var map = new FileHeapMap("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(new File("myPacket.db"), 10, MyPacket.class); + var map = new FileHeapMap("tc", 10, MyPacket.class); Assert.assertNotNull(map.get(5)); } } diff --git a/protocol/src/main/java/com/zfoo/protocol/util/FileUtils.java b/protocol/src/main/java/com/zfoo/protocol/util/FileUtils.java index 64d323c1..fd7cde86 100644 --- a/protocol/src/main/java/com/zfoo/protocol/util/FileUtils.java +++ b/protocol/src/main/java/com/zfoo/protocol/util/FileUtils.java @@ -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()) {