perf[orm]: 优化FileChannelHeap

net/src/main/java/com/zfoo/net/core/http/HttpServer.java

@@ -14,9 +14,9 @@
 package com.zfoo.net.core.http;
 
 import com.zfoo.net.core.AbstractServer;
-import com.zfoo.net.handler.ServerDispatcherHandler;
+import com.zfoo.net.handler.HttpDispatcherHandler;
 import com.zfoo.net.handler.codec.http.HttpCodecHandler;
-import com.zfoo.protocol.IPacket;
+import com.zfoo.net.packet.model.DecodedPacketInfo;
 import com.zfoo.util.net.HostAndPort;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
@@ -36,9 +36,9 @@ public class HttpServer extends AbstractServer {
     /**
      * http的地址解析器
      */
-    private Function<FullHttpRequest, IPacket> uriResolver;
+    private Function<FullHttpRequest, DecodedPacketInfo> uriResolver;
 
-    public HttpServer(HostAndPort host, Function<FullHttpRequest, IPacket> uriResolver) {
+    public HttpServer(HostAndPort host, Function<FullHttpRequest, DecodedPacketInfo> uriResolver) {
         super(host);
         this.uriResolver = uriResolver;
     }
@@ -56,7 +56,7 @@ public class HttpServer extends AbstractServer {
             channel.pipeline().addLast(new ChunkedWriteHandler());
             channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024));
             channel.pipeline().addLast(new HttpCodecHandler(uriResolver));
-            channel.pipeline().addLast(new ServerDispatcherHandler());
+            channel.pipeline().addLast(new HttpDispatcherHandler());
         }
     }
 }

net/src/main/java/com/zfoo/net/handler/HttpDispatcherHandler.java

@@ -0,0 +1,45 @@
+/*
+ * 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.net.handler;
+
+import com.zfoo.net.NetContext;
+import com.zfoo.net.packet.model.DecodedPacketInfo;
+import com.zfoo.net.packet.model.HttpPacketAttachment;
+import com.zfoo.net.util.SessionUtils;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ * @author jaysunxiao
+ * @version 3.0
+ */
+@ChannelHandler.Sharable
+public class HttpDispatcherHandler extends ServerDispatcherHandler {
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        var session = SessionUtils.getSession(ctx);
+        if (session == null) {
+            return;
+        }
+        DecodedPacketInfo decodedPacketInfo = (DecodedPacketInfo) msg;
+
+        var httpPacketAttachment = (HttpPacketAttachment) decodedPacketInfo.getPacketAttachment();
+        if (httpPacketAttachment.getUid() <= 0) {
+            httpPacketAttachment.useExecutorConsistentHash(session.getSid());
+        }
+
+        NetContext.getDispatcher().receive(session, decodedPacketInfo.getPacket(), decodedPacketInfo.getPacketAttachment());
+    }
+}

net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java

@@ -23,13 +23,15 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToMessageCodec;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
 import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.function.Function;
 
+
 /**
  * @author jaysunxiao
  * @version 3.0
@@ -38,9 +40,9 @@ public class HttpCodecHandler extends MessageToMessageCodec<FullHttpRequest, Enc
 
     private static final Logger logger = LoggerFactory.getLogger(HttpCodecHandler.class);
 
-    private Function<FullHttpRequest, IPacket> uriResolver;
+    private Function<FullHttpRequest, DecodedPacketInfo> uriResolver;
 
-    public HttpCodecHandler(Function<FullHttpRequest, IPacket> uriResolver) {
+    public HttpCodecHandler(Function<FullHttpRequest, DecodedPacketInfo> uriResolver) {
         super();
         this.uriResolver = uriResolver;
     }
@@ -48,9 +50,7 @@ public class HttpCodecHandler extends MessageToMessageCodec<FullHttpRequest, Enc
     @Override
     protected void decode(ChannelHandlerContext channelHandlerContext, FullHttpRequest fullHttpRequest, List<Object> list) {
         try {
-            var packet = uriResolver.apply(fullHttpRequest);
-            var attachment = HttpPacketAttachment.valueOf(fullHttpRequest, HttpResponseStatus.OK);
-            var decodedPacketInfo = DecodedPacketInfo.valueOf(packet, attachment);
+            var decodedPacketInfo = uriResolver.apply(fullHttpRequest);
             list.add(decodedPacketInfo);
         } catch (Exception e) {
             logger.error("exception异常", e);
@@ -63,7 +63,6 @@ public class HttpCodecHandler extends MessageToMessageCodec<FullHttpRequest, Enc
 
     @Override
     protected void encode(ChannelHandlerContext channelHandlerContext, EncodedPacketInfo out, List<Object> list) {
-
         try {
             var packet = (IPacket) out.getPacket();
             var attachment = (HttpPacketAttachment) out.getPacketAttachment();
@@ -71,13 +70,31 @@ public class HttpCodecHandler extends MessageToMessageCodec<FullHttpRequest, Enc
             var protocolVersion = attachment.getFullHttpRequest().protocolVersion();
             var httpResponseStatus = attachment.getHttpResponseStatus();
             if (packet.protocolId() == Message.PROTOCOL_ID) {
-                var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus);
-                list.add(fullHttpResponse);
+                var message = (Message) packet;
+                if (StringUtils.isEmpty(message.getMessage())) {
+                    var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus);
+                    fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
+                    fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0);
+                    list.add(fullHttpResponse);
+                } else {
+                    var byteBuf = channelHandlerContext.alloc().ioBuffer();
+                    byteBuf.writeCharSequence(message.getMessage(), StringUtils.DEFAULT_CHARSET);
+                    var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus, byteBuf);
+
+                    fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
+                    fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, byteBuf.readableBytes());
+
+                    list.add(fullHttpResponse);
+                }
             } else {
                 var byteBuf = channelHandlerContext.alloc().ioBuffer();
-                var jsonStr = JsonUtils.object2StringTurbo(packet);
+                var jsonStr = JsonUtils.object2String(packet);
                 byteBuf.writeBytes(StringUtils.bytes(jsonStr));
                 var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus, byteBuf);
+
+                fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
+                fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, byteBuf.readableBytes());
+
                 list.add(fullHttpResponse);
             }
         } catch (Exception e) {

net/src/main/java/com/zfoo/net/packet/common/Message.java

@@ -49,6 +49,12 @@ public class Message implements IPacket {
         return Message.valueOf(packet, code, null);
     }
 
+    public static Message valueOf(String message) {
+        var mess = new Message();
+        mess.message = message;
+        return mess;
+    }
+
     @Override
     public short protocolId() {
         return PROTOCOL_ID;

net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java

@@ -13,7 +13,7 @@
 
 package com.zfoo.net.packet.model;
 
-import com.zfoo.util.math.RandomUtils;
+import com.zfoo.util.math.HashUtils;
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpResponseStatus;
 
@@ -25,6 +25,12 @@ public class HttpPacketAttachment implements IPacketAttachment {
 
     public static final transient short PROTOCOL_ID = 3;
 
+    private long uid;
+
+    private boolean useExecutorConsistentHash;
+
+    private int executorConsistentHash;
+
     private transient FullHttpRequest fullHttpRequest;
 
     private transient HttpResponseStatus httpResponseStatus;
@@ -43,7 +49,11 @@ public class HttpPacketAttachment implements IPacketAttachment {
 
     @Override
     public int executorConsistentHash() {
-        return RandomUtils.randomInt();
+        if (useExecutorConsistentHash) {
+            return executorConsistentHash;
+        } else {
+            return HashUtils.fnvHash(uid);
+        }
     }
 
     @Override
@@ -51,6 +61,35 @@ public class HttpPacketAttachment implements IPacketAttachment {
         return PROTOCOL_ID;
     }
 
+    public void useExecutorConsistentHash(Object argument) {
+        this.useExecutorConsistentHash = true;
+        this.executorConsistentHash = HashUtils.fnvHash(argument);
+    }
+
+    public long getUid() {
+        return uid;
+    }
+
+    public void setUid(long uid) {
+        this.uid = uid;
+    }
+
+    public boolean isUseExecutorConsistentHash() {
+        return useExecutorConsistentHash;
+    }
+
+    public void setUseExecutorConsistentHash(boolean useExecutorConsistentHash) {
+        this.useExecutorConsistentHash = useExecutorConsistentHash;
+    }
+
+    public int getExecutorConsistentHash() {
+        return executorConsistentHash;
+    }
+
+    public void setExecutorConsistentHash(int executorConsistentHash) {
+        this.executorConsistentHash = executorConsistentHash;
+    }
+
     public FullHttpRequest getFullHttpRequest() {
         return fullHttpRequest;
     }

orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentHeapMap.java

@@ -52,4 +52,9 @@ public class ConcurrentHeapMap<V extends IPacket> implements LpMap<V> {
         checkKey(key);
         return map.get(key);
     }
+
+    @Override
+    public long getMaxIndex() {
+        return maxIndexAtomic.get();
+    }
 }

orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java

@@ -14,6 +14,8 @@ package com.zfoo.orm.lpmap;
 
 import com.zfoo.protocol.IPacket;
 
+import java.util.function.BiConsumer;
+
 /**
  * @author jaysunxiao
  * @version 3.0
@@ -55,6 +57,11 @@ public class FileChannelHeapMap<V extends IPacket> implements LpMap<V> {
         return heapMap.get(key);
     }
 
+    @Override
+    public long getMaxIndex() {
+        return heapMap.getMaxIndex();
+    }
+
     private void load() {
         var maxIndex = fileChannelMap.getMaxIndex();
         if (maxIndex <= 0) {
@@ -68,7 +75,10 @@ public class FileChannelHeapMap<V extends IPacket> implements LpMap<V> {
             }
             heapMap.put(key, value);
         }
-
     }
 
+    @Override
+    public void forEach(BiConsumer<Long, V> biConsumer) {
+        heapMap.forEach(biConsumer);
+    }
 }

orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java

@@ -145,6 +145,19 @@ public class FileChannelMap<V extends IPacket> implements LpMap<V>, Closeable {
         }
     }
 
+    @Override
+    public long getMaxIndex() {
+        try {
+            clearByteBuf();
+            indexBuffer.writeBytes(indexFileChannel, 0, 8);
+            return indexBuffer.readLong();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        } finally {
+            clearByteBuf();
+        }
+    }
+
     protected void setKeyValue(long key, V value) {
         try {
             clearByteBuf();
@@ -194,18 +207,6 @@ public class FileChannelMap<V extends IPacket> implements LpMap<V>, Closeable {
         }
     }
 
-    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();

orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java

@@ -77,6 +77,11 @@ public class FileHeapMap<V extends IPacket> implements LpMap<V> {
         return heapMap.get(key);
     }
 
+    @Override
+    public long getMaxIndex() {
+        return heapMap.getMaxIndex();
+    }
+
     private void load() {
         FileInputStream fileInputStream = null;
         FileChannel fileChannel = null;

orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java

@@ -17,6 +17,7 @@ import io.netty.util.collection.LongObjectHashMap;
 
 import java.util.LinkedList;
 import java.util.Queue;
+import java.util.function.BiConsumer;
 
 /**
  * @author jaysunxiao
@@ -81,4 +82,14 @@ public class HeapMap<V extends IPacket> implements LpMap<V> {
         checkKey(key);
         return map.get(key);
     }
+
+    @Override
+    public long getMaxIndex() {
+        return maxIndex;
+    }
+
+    @Override
+    public void forEach(BiConsumer<Long, V> biConsumer) {
+        map.forEach(biConsumer);
+    }
 }

orm/src/main/java/com/zfoo/orm/lpmap/LpMap.java

@@ -15,6 +15,8 @@ package com.zfoo.orm.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
@@ -41,10 +43,16 @@ public interface LpMap<V extends IPacket> {
 
     V get(long key);
 
+    long getMaxIndex();
+
     default void checkKey(long key) {
         if (key <= 0) {
             throw new RunException("key[{}]不能为负数", key);
         }
     }
 
+    default void forEach(BiConsumer<Long, V> biConsumer) {
+
+    }
+
 }