ref[lpmap]: move lpmap from orm to protocol

2026-08-23 02:24:37 +00:00 · 2022-10-09 12:16:53 +08:00
parent b1ac3ed538
commit 46aeeb1d5a
15 changed files with 46 additions and 40 deletions
@@ -0,0 +1,114 @@
+/*
+ * 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.protocol.collection.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;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+public class ConcurrentFileChannelHeapMap<V extends IPacket> implements LpMap<V>, Closeable {
+
+    private final ReentrantLock fileChannelLock = new ReentrantLock();
+
+    private final FileChannelMap<V> fileChannelMap;
+
+    private final ConcurrentHeapMap<V> concurrentHeapMap;
+
+    public ConcurrentFileChannelHeapMap(String dbPath, Class<V> clazz) {
+        fileChannelMap = new FileChannelMap<>(dbPath, clazz);
+        concurrentHeapMap = new ConcurrentHeapMap<>();
+
+        fileChannelMap.forEach((key, v) -> concurrentHeapMap.put(key, v));
+    }
+
+    @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 putIfAbsent(long key, V packet) {
+        var previousValue = concurrentHeapMap.putIfAbsent(key, packet);
+        if (previousValue == null) {
+            previousValue = put(key, packet);
+        }
+        return previousValue;
+    }
+
+    @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();
+    }
+
+    @Override
+    public void clear() {
+        fileChannelLock.lock();
+        try {
+            fileChannelMap.clear();
+            concurrentHeapMap.clear();
+        } finally {
+            fileChannelLock.unlock();
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        fileChannelLock.lock();
+        try {
+            fileChannelMap.close();
+            concurrentHeapMap.clear();
+        } finally {
+            fileChannelLock.unlock();
+        }
+    }
+
+    @Override
+    public void forEach(BiConsumer<Long, V> biConsumer) {
+        concurrentHeapMap.forEach(biConsumer);
+    }
+
+}
@@ -0,0 +1,84 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.IPacket;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.function.BiConsumer;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+public class ConcurrentFileChannelMap<V extends IPacket> implements LpMap<V>, Closeable {
+
+
+    private final FileChannelMap<V> fileChannelMap;
+
+    public ConcurrentFileChannelMap(String dbPath, Class<V> clazz) {
+        fileChannelMap = new FileChannelMap<>(dbPath, clazz);
+    }
+
+    @Override
+    public synchronized V put(long key, V value) {
+        return fileChannelMap.put(key, value);
+    }
+
+    @Override
+    public synchronized V putIfAbsent(long key, V packet) {
+        return fileChannelMap.put(key, packet);
+    }
+
+    @Override
+    public synchronized V delete(long key) {
+        return fileChannelMap.delete(key);
+    }
+
+    @Override
+    public synchronized V get(long key) {
+        return fileChannelMap.get(key);
+    }
+
+    public synchronized List<V> getFrom(long startKey, long endKey) {
+        return fileChannelMap.getFrom(startKey, endKey);
+    }
+
+    @Override
+    public synchronized long getMaxIndex() {
+        return fileChannelMap.getMaxIndex();
+    }
+
+    @Override
+    public synchronized long getIncrementIndex() {
+        return fileChannelMap.getIncrementIndex();
+    }
+
+    @Override
+    public synchronized void clear() {
+        fileChannelMap.clear();
+    }
+
+    @Override
+    public synchronized void close() throws IOException {
+        fileChannelMap.close();
+    }
+
+    @Override
+    public synchronized void forEach(BiConsumer<Long, V> biConsumer) {
+        fileChannelMap.forEach(biConsumer);
+    }
+
+}
@@ -0,0 +1,99 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.IPacket;
+
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+public class ConcurrentHeapMap<V extends IPacket> implements LpMap<V> {
+
+    private final ConcurrentNavigableMap<Long, V> map = new ConcurrentSkipListMap<>();
+
+    private final AtomicLong maxIndexAtomic = new AtomicLong(0);
+
+
+    @Override
+    public V put(long key, V value) {
+        checkKey(key);
+
+        while (true) {
+            var maxIndex = maxIndexAtomic.get();
+
+            if (key <= maxIndex) {
+                break;
+            }
+
+            maxIndexAtomic.compareAndSet(maxIndex, key);
+        }
+
+        return map.put(key, value);
+    }
+
+    @Override
+    public V putIfAbsent(long key, V packet) {
+        var previousValue = map.putIfAbsent(key, packet);
+        if (previousValue == null) {
+            while (true) {
+                var maxIndex = maxIndexAtomic.get();
+
+                if (key <= maxIndex) {
+                    break;
+                }
+
+                maxIndexAtomic.compareAndSet(maxIndex, key);
+            }
+        }
+        return previousValue;
+    }
+
+    @Override
+    public V delete(long key) {
+        checkKey(key);
+        return map.remove(key);
+    }
+
+    @Override
+    public V get(long key) {
+        checkKey(key);
+        return map.get(key);
+    }
+
+    @Override
+    public long getMaxIndex() {
+        return maxIndexAtomic.get();
+    }
+
+    @Override
+    public long getIncrementIndex() {
+        return maxIndexAtomic.incrementAndGet();
+    }
+
+    @Override
+    public void clear() {
+        map.clear();
+        maxIndexAtomic.set(0);
+    }
+
+    @Override
+    public void forEach(BiConsumer<Long, V> biConsumer) {
+        map.forEach(biConsumer);
+    }
+}
@@ -0,0 +1,82 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.IPacket;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.function.BiConsumer;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+public class FileChannelHeapMap<V extends IPacket> implements LpMap<V>, Closeable {
+
+    private final FileChannelMap<V> fileChannelMap;
+
+    private final HeapMap<V> heapMap;
+
+    public FileChannelHeapMap(String dbPath, int initialCapacity, Class<V> clazz) {
+        fileChannelMap = new FileChannelMap<>(dbPath, clazz);
+        heapMap = new HeapMap<>(initialCapacity);
+
+        fileChannelMap.forEach((key, v) -> heapMap.put(key, v));
+    }
+
+    @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);
+    }
+
+    @Override
+    public long getMaxIndex() {
+        return heapMap.getMaxIndex();
+    }
+
+    @Override
+    public long getIncrementIndex() {
+        return heapMap.getIncrementIndex();
+    }
+
+    @Override
+    public void clear() {
+        fileChannelMap.clear();
+        heapMap.clear();
+    }
+
+    @Override
+    public void close() throws IOException {
+        fileChannelMap.close();
+        heapMap.clear();
+    }
+
+    @Override
+    public void forEach(BiConsumer<Long, V> biConsumer) {
+        heapMap.forEach(biConsumer);
+    }
+
+}
@@ -0,0 +1,244 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.IPacket;
+import com.zfoo.protocol.ProtocolManager;
+import com.zfoo.protocol.exception.RunException;
+import com.zfoo.protocol.registration.IProtocolRegistration;
+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;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.BiConsumer;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+public class FileChannelMap<V extends IPacket> implements LpMap<V>, Closeable {
+
+    private final File dbFile;
+    protected RandomAccessFile dbFileRandomAccess;
+    protected FileChannel dbFileChannel;
+
+    private final File indexFile;
+    protected RandomAccessFile indexFileRandomAccess;
+    protected FileChannel indexFileChannel;
+
+    protected long maxIndex;
+
+    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();
+
+            var protocolId = ProtocolManager.protocolId(clazz);
+            protocolRegistration = ProtocolManager.getProtocol(protocolId);
+
+            indexBuffer = ByteBufAllocator.DEFAULT.ioBuffer(16);
+            dbBuffer = ByteBufAllocator.DEFAULT.ioBuffer(100);
+
+            maxIndex = indexFileChannel.size() / 16;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+
+    @Override
+    public V put(long key, V packet) {
+        checkKey(key);
+
+        V previousValue = null;
+        if (key <= maxIndex) {
+            previousValue = get(key);
+        } else {
+            maxIndex = key;
+        }
+
+        setKeyValue(key, packet);
+        return previousValue;
+    }
+
+    @Override
+    public V delete(long key) {
+        checkKey(key);
+
+        if (key <= maxIndex) {
+            var previousValue = get(key);
+            resetKey(key);
+            return previousValue;
+        } else {
+            return null;
+        }
+    }
+
+    @Override
+    public V get(long key) {
+        checkKey(key);
+
+        if (key > maxIndex) {
+            return null;
+        }
+
+        try {
+            clearByteBuf();
+            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;
+        }
+    }
+
+    /**
+     * 获取从startKey到endKey的值
+     *
+     * @param startKey inclusive
+     * @param endKey   exclusive
+     * @return list
+     */
+    public List<V> getFrom(long startKey, long endKey) {
+        checkKey(startKey);
+        checkKey(endKey);
+
+        if (startKey >= endKey) {
+            throw new RunException("range error startKey < endKey");
+        }
+        if (startKey > maxIndex) {
+            return Collections.emptyList();
+        }
+
+        var list = new ArrayList<V>();
+        for (var i = startKey; i < endKey; i++) {
+            var value = get(i);
+            if (value != null) {
+                list.add(value);
+            }
+        }
+        return list;
+    }
+
+
+    @Override
+    public long getMaxIndex() {
+        return maxIndex;
+    }
+
+    @Override
+    public long getIncrementIndex() {
+        maxIndex++;
+        return maxIndex;
+    }
+
+    @Override
+    public void forEach(BiConsumer<Long, V> biConsumer) {
+        for (var i = 0L; i <= getMaxIndex(); i++) {
+            var value = get(i);
+            if (value != null) {
+                biConsumer.accept(i, value);
+            }
+        }
+    }
+
+    @Override
+    public void clear() {
+        try {
+            maxIndex = 0;
+            indexFileRandomAccess.setLength(0);
+            dbFileRandomAccess.setLength(0);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
+    }
+
+    @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 {
+            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 clearByteBuf() {
+        indexBuffer.clear();
+        dbBuffer.clear();
+    }
+
+}
@@ -0,0 +1,151 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.IPacket;
+import com.zfoo.protocol.ProtocolManager;
+import com.zfoo.protocol.buffer.ByteBufUtils;
+import com.zfoo.protocol.registration.IProtocolRegistration;
+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.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.util.function.BiConsumer;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+public class FileHeapMap<V extends IPacket> implements LpMap<V> {
+
+    private final File dbFile;
+
+    private final IProtocolRegistration protocolRegistration;
+
+    private final HeapMap<V> heapMap;
+
+    public FileHeapMap(String dbPath, Class<V> clazz) {
+        try {
+            this.dbFile = FileUtils.getOrCreateFile(dbPath, StringUtils.format("{}.db", clazz.getSimpleName()));
+
+            var protocolId = ProtocolManager.protocolId(clazz);
+            protocolRegistration = ProtocolManager.getProtocol(protocolId);
+            heapMap = new HeapMap<>();
+
+            load();
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+
+    @Override
+    public V put(long key, V value) {
+        return heapMap.put(key, value);
+    }
+
+    @Override
+    public V delete(long key) {
+        return heapMap.delete(key);
+    }
+
+    @Override
+    public V get(long key) {
+        return heapMap.get(key);
+    }
+
+    @Override
+    public long getMaxIndex() {
+        return heapMap.getMaxIndex();
+    }
+
+    @Override
+    public long getIncrementIndex() {
+        return heapMap.getIncrementIndex();
+    }
+
+    @Override
+    public void forEach(BiConsumer<Long, V> biConsumer) {
+        heapMap.forEach(biConsumer);
+    }
+
+    @Override
+    public void clear() {
+        heapMap.clear();
+        save();
+    }
+
+    private void load() {
+        FileInputStream fileInputStream = null;
+        FileChannel fileChannel = null;
+        ByteBuf buffer = null;
+        try {
+            fileInputStream = FileUtils.openInputStream(dbFile);
+            fileChannel = fileInputStream.getChannel();
+
+            if (fileChannel.size() <= 0) {
+                return;
+            }
+
+            buffer = ByteBufAllocator.DEFAULT.ioBuffer(1000);
+            buffer.writeBytes(fileChannel, 0L, (int) dbFile.length());
+
+            var size = ByteBufUtils.readLong(buffer);
+            for (var i = 0; i < size; i++) {
+                var key = ByteBufUtils.readLong(buffer);
+                var value = (V) protocolRegistration.read(buffer);
+                put(key, value);
+            }
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        } finally {
+            IOUtils.closeIO(fileChannel, fileInputStream);
+            ReferenceCountUtil.release(buffer);
+        }
+    }
+
+    public void save() {
+        FileOutputStream fileOutputStream = null;
+        ByteBuf buffer = null;
+        try {
+            fileOutputStream = FileUtils.openOutputStream(dbFile, false);
+            buffer = ByteBufAllocator.DEFAULT.heapBuffer(1000);
+
+            // 写入长度
+            ByteBufUtils.writeLong(buffer, heapMap.map.size());
+            buffer.readBytes(fileOutputStream, buffer.readableBytes());
+
+            for (var entry : heapMap.map.entries()) {
+                buffer.clear();
+
+                var key = entry.key();
+                var value = entry.value();
+                ByteBufUtils.writeLong(buffer, key);
+                protocolRegistration.write(buffer, value);
+
+                buffer.readBytes(fileOutputStream, buffer.readableBytes());
+            }
+        } catch (IOException e) {
+            IOUtils.closeIO(fileOutputStream);
+            ReferenceCountUtil.release(buffer);
+        }
+    }
+}
@@ -0,0 +1,87 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.IPacket;
+import io.netty.util.collection.LongObjectHashMap;
+
+import java.util.function.BiConsumer;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+public class HeapMap<V extends IPacket> implements LpMap<V> {
+
+    protected LongObjectHashMap<V> map;
+
+    protected long maxIndex = 0;
+
+    public HeapMap() {
+        this(128);
+    }
+
+    public HeapMap(int initialCapacity) {
+        map = new LongObjectHashMap<>(initialCapacity);
+    }
+
+
+    @Override
+    public V put(long key, V value) {
+        checkKey(key);
+
+        if (key > maxIndex) {
+            maxIndex = key;
+        }
+
+        return map.put(key, value);
+    }
+
+    @Override
+    public V delete(long key) {
+        checkKey(key);
+        if (key > maxIndex) {
+            return null;
+        }
+
+        return map.remove(key);
+    }
+
+    @Override
+    public V get(long key) {
+        checkKey(key);
+        return map.get(key);
+    }
+
+    @Override
+    public long getMaxIndex() {
+        return maxIndex;
+    }
+
+    @Override
+    public long getIncrementIndex() {
+        maxIndex++;
+        return maxIndex;
+    }
+
+    @Override
+    public void clear() {
+        maxIndex = 0;
+        map.clear();
+    }
+
+    @Override
+    public void forEach(BiConsumer<Long, V> biConsumer) {
+        map.forEach(biConsumer);
+    }
+}
@@ -0,0 +1,62 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.IPacket;
+import com.zfoo.protocol.exception.RunException;
+
+import java.util.function.BiConsumer;
+
+/**
+ * 类型固定的map，key为long，value为IPacket
+ * 其中long必须大于等于0，value可以为null
+ *
+ * @author godotg
+ * @version 3.0
+ */
+public interface LpMap<V extends IPacket> {
+
+    /**
+     * @param packet the previous value associated with key, or null if there was no mapping for key.
+     */
+    V put(long key, V packet);
+
+    default V putIfAbsent(long key, V packet) {
+        var v = get(key);
+        if (v == null) {
+            v = put(key, packet);
+        }
+        return v;
+    }
+
+    /**
+     * @return 返回被删除的那个值
+     */
+    V delete(long key);
+
+    V get(long key);
+
+    long getMaxIndex();
+
+    long getIncrementIndex();
+
+    void clear();
+
+    void forEach(BiConsumer<Long, V> biConsumer);
+
+    default void checkKey(long key) {
+        if (key < 0) {
+            throw new RunException("key[{}]只能为大于等于0的正数", key);
+        }
+    }
+}
@@ -26,13 +26,14 @@ import java.util.concurrent.CountDownLatch;
@Ignore
 public class ConcurrentTest {

+    private static final int EXECUTOR_SIZE = Runtime.getRuntime().availableProcessors();
+
    @Test
    public void test() throws InterruptedException {
        var map = new CopyOnWriteHashMapLongObject<Integer>();
        var num = 1_0000;
-        var executorSize = Runtime.getRuntime().availableProcessors();
-        var countDownLatch = new CountDownLatch(executorSize);
-        for (var i = 0; i < executorSize; i++) {
+        var countDownLatch = new CountDownLatch(EXECUTOR_SIZE);
+        for (var i = 0; i < EXECUTOR_SIZE; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
@@ -46,8 +47,8 @@ public class ConcurrentTest {
        countDownLatch.await();
        Assert.assertEquals(map.size(), num);

-        var countDownLatch2 = new CountDownLatch(executorSize);
-        for (var i = 0; i < executorSize; i++) {
+        var countDownLatch2 = new CountDownLatch(EXECUTOR_SIZE);
+        for (var i = 0; i < EXECUTOR_SIZE; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
@@ -0,0 +1,73 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.ProtocolManager;
+import com.zfoo.protocol.collection.lpmap.model.MyPacket;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+@Ignore
+public class ConcurrentFileChannelMapTest {
+
+    private static final int EXECUTOR_SIZE = Runtime.getRuntime().availableProcessors();
+
+    @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(EXECUTOR_SIZE);
+        for (int i = 0; i < EXECUTOR_SIZE; i++) {
+            new Thread(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();
+                }
+            }).start();
+        }
+        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);
+        }
+    }
+}
@@ -0,0 +1,82 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.collection.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;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+@Ignore
+public class ConcurrentHeapMapTest {
+
+    private static final int EXECUTOR_SIZE = Runtime.getRuntime().availableProcessors();
+
+    @Test
+    public void putIfAbsentTest() {
+        ProtocolManager.initProtocol(Set.of(MyPacket.class));
+        var myPacket = new MyPacket();
+        myPacket.setA(1);
+
+        var map = new ConcurrentHeapMap<MyPacket>();
+
+        var previous1 = map.put(1, myPacket);
+        var previous2 = map.put(2, myPacket);
+        var previous3 = map.put(2, new MyPacket());
+
+        Assert.assertNull(previous1);
+        Assert.assertNull(previous2);
+        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(EXECUTOR_SIZE);
+        for (int i = 0; i < EXECUTOR_SIZE; i++) {
+            new Thread(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();
+                }
+            }).start();
+        }
+        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);
+        }
+    }
+}
@@ -0,0 +1,91 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.collection.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;
+
+/**
+ * @author godotg
+ * @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.put(0, myPacket);
+        map.put(1, myPacket);
+        map.put(2, myPacket);
+        map.put(3, 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));
+    }
+
+    @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);
+        }
+    }
+}
@@ -0,0 +1,90 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.collection.lpmap.model.MyPacket;
+import com.zfoo.protocol.ProtocolManager;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.util.Set;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+@Ignore
+public class FileHeapMapTest {
+
+    @Test
+    public void test() {
+        ProtocolManager.initProtocol(Set.of(MyPacket.class));
+
+        var map = new FileHeapMap<MyPacket>("db", MyPacket.class);
+        var myPacket = new MyPacket();
+        myPacket.setA(9999);
+
+        var packet = map.put(1, myPacket);
+        Assert.assertNull(packet);
+
+        packet = map.put(2, myPacket);
+        Assert.assertNull(packet);
+
+        packet = map.delete(4);
+        Assert.assertNull(packet);
+
+        packet = map.delete(2);
+        Assert.assertEquals(packet, myPacket);
+
+        map.put(1, myPacket);
+        map.put(2, myPacket);
+        map.put(3, myPacket);
+        map.put(4, myPacket);
+        map.put(5, myPacket);
+        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);
+        }
+    }
+}
@@ -0,0 +1,62 @@
+/*
+ * 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.protocol.collection.lpmap;
+
+import com.zfoo.protocol.collection.lpmap.model.MyPacket;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+@Ignore
+public class HeapMapTest {
+
+    @Test
+    public void test() {
+        var map = new HeapMap<MyPacket>();
+        var myPacket = new MyPacket();
+
+        var packet = map.put(0, myPacket);
+        Assert.assertNull(packet);
+
+        packet = map.put(3, myPacket);
+        Assert.assertNull(packet);
+
+        packet = map.delete(4);
+        Assert.assertNull(packet);
+
+        packet = map.delete(3);
+        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);
+        }
+    }
+
+}
@@ -0,0 +1,80 @@
+/*
+ * 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.protocol.collection.lpmap.model;
+
+import com.zfoo.protocol.IPacket;
+
+import java.util.Objects;
+
+/**
+ * @author godotg
+ * @version 3.0
+ */
+public class MyPacket implements IPacket {
+
+    public static final transient short PROTOCOL_ID = 1;
+
+    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;
+    }
+
+    public void setA(int a) {
+        this.a = a;
+    }
+
+    public String getB() {
+        return b;
+    }
+
+    public void setB(String b) {
+        this.b = b;
+    }
+}