net包优化

bin/restart.bat

@@ -0,0 +1,9 @@
+@ECHO OFF
+
+SET APP_HOME=%~dp0
+
+IF NOT EXIST "%APP_HOME%\conf\application.xml"  SET APP_HOME=%~dp0..
+
+call %APP_HOME%\bin\shutdown.bat 
+
+call %APP_HOME%\bin\start.bat 

bin/restart.sh

@@ -0,0 +1,20 @@
+#!/bin/sh
+
+export LC_ALL="zh_CN.UTF-8"
+
+APP_HOME=`dirname "$0"`
+
+cd "$APP_HOME"/..
+
+APP_HOME=`pwd`
+
+if [ ! -f "$APP_HOME"/conf/application.xml ]; then 
+     APP_HOME="$APP_HOME"/..  
+fi
+
+cd "$APP_HOME"
+
+./bin/shutdown.sh
+
+./bin/start.sh
+

src/META-INF/application-template.xml

@@ -37,9 +37,11 @@
             threads： 线程总数， 默认: <group>节点数*CPU核数*8
             bufferCapacity: ByteBuffer的初始化大小， 默认: 8K; 
             bufferPoolSize： ByteBuffer池的大小，默认: <group>节点数*CPU核数*8
+            readTimeoutSecond: TCP读取超时秒数, 默认为6秒， 为0表示无超时限制
+            writeTimeoutSecond: TCP写入超时秒数, 默认为6秒， 为0表示无超时限制
             strategy: 远程请求的负载均衡策略， 必须是org.redkale.net.TransportStrategy的实现类
         -->
-        <transport bufferCapacity="8K" bufferPoolSize="32" threads="32"/>
+        <transport bufferCapacity="8K" bufferPoolSize="32" threads="32" readTimeoutSecond="6" writeTimeoutSecond="6"/>
         
         <!--
             一个组包含多个node， 同一Service服务可以由多个进程提供，这些进程称为一个GROUP，且同一GROUP内的进程必须在同一机房或局域网内
@@ -64,7 +66,7 @@
         <!--
             全局的数据源设置, 可以是CacheSource、DataSource， JDBC的DataSource通常通过persistence.xml配置，此处多用于CacheSource的配置
             name:  资源名，用于依赖注入。
-            value：类名，必须是CacheSource或DataSource的子类，且必须实现Service接口。
+            value：类名，必须是CacheSource或DataSource的子类，且必须实现Service接口。如果是DataSource.class，系统自动映射成DataJdbcSource.class
             groups: 指定groups。
             xxx: 其他属性与子节点通过Service.init方法传入的AnyValue获取。
         -->
@@ -92,7 +94,8 @@
                System.setProperty("convert.json.writer.buffer.defsize", "4096");
                System.setProperty("convert.bson.writer.buffer.defsize", "4096");
                
-            <properties>节点下也可包含非<property>节点，其节点可以通过@Resource(name="properties.xxxxxx")进行注入, 被注解的字段类型只能是AnyValue、AnyValue[]
+            <properties>节点下也可包含非<property>节点.
+            非<property>其节点可以通过@Resource(name="properties.xxxxxx")进行注入, 被注解的字段类型只能是AnyValue、AnyValue[]
         -->
         <properties load="config.properties">
             <property name="system.property.yyyy" value="YYYYYY"/>
@@ -147,7 +150,7 @@
             <service value="com.xxx.XXX2Service" name="" groups="xxx;yyy"/>
             <!--   给Service增加配置属性 -->
             <service value="com.xxx.XXX1Service"> 
-                <!-- property节点值在 public void init(AnyValue conf) 方法中可以通过 AnyValue properties =  conf.getAnyValue("properties");获取 -->
+                <!-- property值在public void init(AnyValue conf)方法中可以通过AnyValue properties=conf.getAnyValue("properties")获取 -->
                 <property name="xxxxxx" value="XXXXXXXX"/>  
                 <property name="xxxxxx" value="XXXXXXXX"/>
             </service>
@@ -172,7 +175,7 @@
             
             <!--   给Filter增加配置属性 -->
             <filter value="com.xxx.XXX12Filter"> 
-                <!-- property节点值在 public void init(AnyValue conf) 方法中可以通过 AnyValue properties =  conf.getAnyValue("properties");获取 -->
+                <!-- property值在public void init(AnyValue conf)方法中可以通过AnyValue properties=conf.getAnyValue("properties")获取 -->
                 <property name="xxxxxx" value="XXXXXXXX"/>  
                 <property name="xxxxxx" value="XXXXXXXX"/>
             </filter>

src/module-info.java

@@ -7,10 +7,11 @@
 module org.redkale {
 
     requires java.se;
-    requires jdk.unsupported;
+    requires jdk.unsupported;  //sun.misc.Unsafe
 
     exports javax.annotation;
     exports javax.persistence;
+    exports org.redkale.asm;
     exports org.redkale.boot;
     exports org.redkale.boot.watch;
     exports org.redkale.convert;

src/org/redkale/asm/asm.txt

@@ -8,6 +8,7 @@ ClassReader.java
 ClassVisitor.java
 ClassWriter.java
 Context.java
+CurrentFrame.java
 Edge.java
 FieldVisitor.java
 FieldWriter.java
@@ -18,25 +19,8 @@ Item.java
 Label.java
 MethodVisitor.java
 MethodWriter.java
+ModuleVisitor.java
+ModuleWriter.java
 Opcodes.java
 Type.java
 TypePath.java
-
-
-
-public static void main(String[] args) throws Throwable {
-    File srcasmroot = new File("D:/JAVA/JDK源码/JDK9源码/java.base/jdk/internal/org/objectweb/asm");
-    File destasmroot = new File("D:/Java-Projects/RedkaleProject/src/org/redkale/asm");
-    String line = null;
-    LineNumberReader txtin = new LineNumberReader(new FileReader(new File(destasmroot, "asm.txt")));
-    while ((line = txtin.readLine()) != null) {
-        line = line.trim();
-        if (!line.endsWith(".java")) continue;
-        File srcfile = new File(srcasmroot, line);
-        File destfile = new File(destasmroot, line);
-        String content = Utility.readThenClose(new FileInputStream(srcfile));
-        FileOutputStream out = new FileOutputStream(destfile);
-        out.write(content.replace("jdk.internal.org.objectweb", "org.redkale").getBytes());
-        out.close();
-    }
-}

src/org/redkale/boot/Application.java

@@ -245,6 +245,8 @@ public final class Application {
         TransportStrategy strategy = null;
         int bufferCapacity = 8 * 1024;
         int bufferPoolSize = Runtime.getRuntime().availableProcessors() * 16;
+        int readTimeoutSecond = TransportFactory.DEFAULT_READTIMEOUTSECOND;
+        int writeTimeoutSecond = TransportFactory.DEFAULT_WRITETIMEOUTSECOND;
         AtomicLong createBufferCounter = new AtomicLong();
         AtomicLong cycleBufferCounter = new AtomicLong();
         if (resources != null) {
@@ -255,6 +257,8 @@ public final class Application {
                 //--------------transportBufferPool-----------
                 bufferCapacity = Math.max(parseLenth(transportConf.getValue("bufferCapacity"), bufferCapacity), 4 * 1024);
                 bufferPoolSize = parseLenth(transportConf.getValue("bufferPoolSize"), groupsize * Runtime.getRuntime().availableProcessors() * 8);
+                readTimeoutSecond = transportConf.getIntValue("readTimeoutSecond", readTimeoutSecond);
+                writeTimeoutSecond = transportConf.getIntValue("writeTimeoutSecond", writeTimeoutSecond);
                 final int threads = parseLenth(transportConf.getValue("threads"), groupsize * Runtime.getRuntime().availableProcessors() * 8);
                 final int capacity = bufferCapacity;
                 transportPool = new ObjectPool<>(createBufferCounter, cycleBufferCounter, bufferPoolSize,
@@ -306,7 +310,7 @@ public final class Application {
                     return true;
                 });
         }
-        this.sncpTransportFactory = TransportFactory.create(transportExec, transportPool, transportGroup, strategy);
+        this.sncpTransportFactory = TransportFactory.create(transportExec, transportPool, transportGroup, readTimeoutSecond, writeTimeoutSecond, strategy);
         DefaultAnyValue tarnsportConf = DefaultAnyValue.create(TransportFactory.NAME_PINGINTERVAL, System.getProperty("net.transport.pinginterval", "30"));
         this.sncpTransportFactory.init(tarnsportConf, Sncp.PING_BUFFER, Sncp.PONG_BUFFER.remaining());
         Thread.currentThread().setContextClassLoader(this.classLoader);

src/org/redkale/boot/NodeServer.java

@@ -179,6 +179,7 @@ public abstract class NodeServer {
             for (AnyValue sourceConf : resources.getAnyValues("source")) {
                 try {
                     Class type = serverClassLoader.loadClass(sourceConf.getValue("value"));
+                    if (type == DataSource.class) type = DataJdbcSource.class;
                     if (!Service.class.isAssignableFrom(type)) {
                         logger.log(Level.SEVERE, "load application source resource, but not Service error: " + sourceConf);
                     } else if (CacheSource.class.isAssignableFrom(type)) {

src/org/redkale/convert/CollectionDecoder.java

@@ -50,7 +50,7 @@ public final class CollectionDecoder<T> implements Decodeable<Reader, Collection
                 factory.register(type, this);
                 this.decoder = factory.loadDecoder(this.componentType);
             } else {
-                throw new ConvertException("collectiondecoder not support the type (" + type + ")");
+                throw new ConvertException("CollectionDecoder not support the type (" + type + ")");
             }
         } finally {
             inited = true;

src/org/redkale/convert/ConvertFactory.java

@@ -13,7 +13,7 @@ import java.nio.ByteBuffer;
 import java.nio.channels.CompletionHandler;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.*;
 import java.util.regex.Pattern;
 import java.util.stream.*;
 import org.redkale.convert.ext.*;
@@ -88,6 +88,7 @@ public abstract class ConvertFactory<R extends Reader, W extends Writer> {
             this.register(String.class, StringSimpledCoder.instance);
             this.register(CharSequence.class, CharSequenceSimpledCoder.instance);
             this.register(java.util.Date.class, DateSimpledCoder.instance);
+            this.register(AtomicInteger.class, AtomicIntegerSimpledCoder.instance);
             this.register(AtomicLong.class, AtomicLongSimpledCoder.instance);
             this.register(BigInteger.class, BigIntegerSimpledCoder.instance);
             this.register(BigDecimal.class, BigDecimalSimpledCoder.instance);
@@ -376,12 +377,17 @@ public abstract class ConvertFactory<R extends Reader, W extends Writer> {
         encoders.put(clazz, coder);
     }
 
+    public final <E> void register(final Type clazz, final Decodeable<R, E> decoder, final Encodeable<W, E> encoder) {
+        decoders.put(clazz, decoder);
+        encoders.put(clazz, encoder);
+    }
+
     public final <E> void register(final Type clazz, final Decodeable<R, E> decoder) {
         decoders.put(clazz, decoder);
     }
 
-    public final <E> void register(final Type clazz, final Encodeable<W, E> printer) {
-        encoders.put(clazz, printer);
+    public final <E> void register(final Type clazz, final Encodeable<W, E> encoder) {
+        encoders.put(clazz, encoder);
     }
 
     public final <E> Decodeable<R, E> findDecoder(final Type type) {
@@ -462,6 +468,8 @@ public abstract class ConvertFactory<R extends Reader, W extends Writer> {
             decoder = new StreamDecoder(this, type);
         } else if (Map.class.isAssignableFrom(clazz)) {
             decoder = new MapDecoder(this, type);
+        } else if (Optional.class == clazz) {
+            decoder = new OptionalCoder(this, type);
         } else if (clazz == Object.class) {
             od = new ObjectDecoder(type);
             decoder = od;
@@ -547,6 +555,8 @@ public abstract class ConvertFactory<R extends Reader, W extends Writer> {
             encoder = new StreamEncoder(this, type);
         } else if (Map.class.isAssignableFrom(clazz)) {
             encoder = new MapEncoder(this, type);
+        } else if (Optional.class == clazz) {
+            encoder = new OptionalCoder(this, type);
         } else if (clazz == Object.class) {
             return (Encodeable<W, E>) this.anyEncoder;
         } else if (!clazz.getName().startsWith("java.") || java.net.HttpCookie.class == clazz) {

src/org/redkale/convert/OptionalCoder.java

@@ -0,0 +1,107 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.redkale.convert;
+
+import java.lang.reflect.*;
+import java.util.*;
+
+/**
+ * Optional 的SimpledCoder实现
+ *
+ * <p>
+ * 详情见: https://redkale.org
+ *
+ * @author zhangjx
+ * @param <R> Reader输入的子类型
+ * @param <W> Writer输出的子类型
+ */
+public class OptionalCoder<R extends Reader, W extends Writer, T> extends SimpledCoder<R, W, Optional<T>> {
+
+    private final Type type;
+
+    private final Type componentType;
+
+    protected final Class componentClass;
+
+    protected final Decodeable<Reader, T> decoder;
+
+    protected final Encodeable<Writer, T> encoder;
+
+    private boolean inited = false;
+
+    private final Object lock = new Object();
+
+    public OptionalCoder(final ConvertFactory factory, final Type type) {
+        this.type = type;
+        try {
+            if (type instanceof ParameterizedType) {
+                final ParameterizedType pt = (ParameterizedType) type;
+                this.componentType = pt.getActualTypeArguments()[0];
+                factory.register(type, this);
+                this.decoder = factory.loadDecoder(this.componentType);
+                if (this.componentType instanceof TypeVariable) {
+                    this.encoder = factory.getAnyEncoder();
+                    this.componentClass = Object.class;
+                } else {
+                    if (componentType instanceof ParameterizedType) {
+                        final ParameterizedType pt2 = (ParameterizedType) componentType;
+                        this.componentClass = (Class) pt2.getRawType();
+                    } else {
+                        this.componentClass = (Class) componentType;
+                    }
+                    this.encoder = factory.loadEncoder(this.componentType);
+                }
+            } else {
+                this.componentType = Object.class;
+                this.componentClass = Object.class;
+                this.decoder = factory.loadDecoder(this.componentType);
+                this.encoder = factory.getAnyEncoder();
+            }
+        } finally {
+            inited = true;
+            synchronized (lock) {
+                lock.notifyAll();
+            }
+        }
+    }
+
+    @Override
+    public void convertTo(W out, Optional<T> value) {
+        if (value == null || !value.isPresent()) {
+            out.writeObjectNull(null);
+            return;
+        }
+        if (this.encoder == null) {
+            if (!this.inited) {
+                synchronized (lock) {
+                    try {
+                        lock.wait();
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                    }
+                }
+            }
+        }
+        this.encoder.convertTo(out, value.get());
+    }
+
+    @Override
+    public Optional<T> convertFrom(R in) {
+        if (this.decoder == null) {
+            if (!this.inited) {
+                synchronized (lock) {
+                    try {
+                        lock.wait();
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                    }
+                }
+            }
+        }
+        return Optional.ofNullable(this.decoder.convertFrom(in));
+    }
+
+}

src/org/redkale/convert/bson/BsonConvert.java

@@ -73,7 +73,7 @@ public final class BsonConvert extends BinaryConvert<BsonReader, BsonWriter> {
     }
 
     public void offerBsonReader(final BsonReader in) {
-        if (in != null) readerPool.offer(in);
+        if (in != null) readerPool.accept(in);
     }
 
     //------------------------------ writer -----------------------------------------------------------
@@ -90,7 +90,7 @@ public final class BsonConvert extends BinaryConvert<BsonReader, BsonWriter> {
     }
 
     public void offerBsonWriter(final BsonWriter out) {
-        if (out != null) writerPool.offer(out);
+        if (out != null) writerPool.accept(out);
     }
 
     //------------------------------ convertFrom -----------------------------------------------------------
@@ -106,7 +106,7 @@ public final class BsonConvert extends BinaryConvert<BsonReader, BsonWriter> {
         in.setBytes(bytes, start, len);
         @SuppressWarnings("unchecked")
         T rs = (T) factory.loadDecoder(type).convertFrom(in);
-        readerPool.offer(in);
+        readerPool.accept(in);
         return rs;
     }
 
@@ -145,7 +145,7 @@ public final class BsonConvert extends BinaryConvert<BsonReader, BsonWriter> {
             final BsonWriter out = writerPool.get().tiny(tiny);
             out.writeNull();
             byte[] result = out.toArray();
-            writerPool.offer(out);
+            writerPool.accept(out);
             return result;
         }
         return convertTo(value.getClass(), value);
@@ -157,7 +157,7 @@ public final class BsonConvert extends BinaryConvert<BsonReader, BsonWriter> {
         final BsonWriter out = writerPool.get().tiny(tiny);
         factory.loadEncoder(type).convertTo(out, value);
         byte[] result = out.toArray();
-        writerPool.offer(out);
+        writerPool.accept(out);
         return result;
     }
 
@@ -167,7 +167,7 @@ public final class BsonConvert extends BinaryConvert<BsonReader, BsonWriter> {
         final BsonWriter out = writerPool.get().tiny(tiny);
         ((AnyEncoder) factory.getAnyEncoder()).convertMapTo(out, values);
         byte[] result = out.toArray();
-        writerPool.offer(out);
+        writerPool.accept(out);
         return result;
     }
 

src/org/redkale/convert/ext/AtomicIntegerSimpledCoder.java

@@ -0,0 +1,35 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.redkale.convert.ext;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import org.redkale.convert.*;
+
+/**
+ * AtomicInteger 的SimpledCoder实现
+ *
+ * <p>
+ * 详情见: https://redkale.org
+ *
+ * @author zhangjx
+ * @param <R> Reader输入的子类型
+ * @param <W> Writer输出的子类型
+ */
+public class AtomicIntegerSimpledCoder<R extends Reader, W extends Writer> extends SimpledCoder<R, W, AtomicInteger> {
+
+    public static final AtomicIntegerSimpledCoder instance = new AtomicIntegerSimpledCoder();
+
+    @Override
+    public void convertTo(W out, AtomicInteger value) {
+        out.writeInt(value == null ? 0 : value.get());
+    }
+
+    @Override
+    public AtomicInteger convertFrom(R in) {
+        return new AtomicInteger(in.readInt());
+    }
+
+}

src/org/redkale/convert/json/JsonConvert.java

@@ -23,7 +23,7 @@ import org.redkale.util.*;
 @SuppressWarnings("unchecked")
 public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
 
-    public static final Type TYPE_MAP_STRING_STRING = new TypeToken<java.util.LinkedHashMap<String, String>>() {
+    public static final Type TYPE_MAP_STRING_STRING = new TypeToken<java.util.HashMap<String, String>>() {
     }.getType();
 
     private static final ObjectPool<JsonReader> readerPool = JsonReader.createPool(Integer.getInteger("convert.json.pool.size", 16));
@@ -60,7 +60,7 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
     }
 
     public void offerJsonReader(final JsonReader in) {
-        if (in != null) readerPool.offer(in);
+        if (in != null) readerPool.accept(in);
     }
 
     //------------------------------ writer -----------------------------------------------------------
@@ -81,7 +81,7 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
     }
 
     public void offerJsonWriter(final JsonWriter out) {
-        if (out != null) writerPool.offer(out);
+        if (out != null) writerPool.accept(out);
     }
 
     //------------------------------ convertFrom -----------------------------------------------------------
@@ -100,7 +100,7 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
         final JsonReader in = readerPool.get();
         in.setText(text, start, len);
         T rs = (T) factory.loadDecoder(type).convertFrom(in);
-        readerPool.offer(in);
+        readerPool.accept(in);
         return rs;
     }
 
@@ -142,7 +142,7 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
         final JsonWriter out = writerPool.get().tiny(tiny);
         factory.loadEncoder(type).convertTo(out, value);
         String result = out.toString();
-        writerPool.offer(out);
+        writerPool.accept(out);
         return result;
     }
 
@@ -152,7 +152,7 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
         final JsonWriter out = writerPool.get().tiny(tiny);
         ((AnyEncoder) factory.getAnyEncoder()).convertMapTo(out, values);
         String result = out.toString();
-        writerPool.offer(out);
+        writerPool.accept(out);
         return result;
     }
 
@@ -160,7 +160,7 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
         if (value == null) {
             new JsonStreamWriter(tiny, out).writeNull();
         } else {
-            factory.loadEncoder(value.getClass()).convertTo(new JsonStreamWriter(tiny, out), value);
+            convertTo(out, value.getClass(), value);
         }
     }
 
@@ -169,7 +169,15 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
         if (value == null) {
             new JsonStreamWriter(tiny, out).writeNull();
         } else {
-            factory.loadEncoder(type).convertTo(new JsonStreamWriter(tiny, out), value);
+            final JsonWriter writer = writerPool.get().tiny(tiny);
+            factory.loadEncoder(type).convertTo(writer, value);
+            byte[] bs = writer.toBytes();
+            writerPool.accept(writer);
+            try {
+                out.write(bs);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
         }
     }
 
@@ -177,7 +185,15 @@ public final class JsonConvert extends TextConvert<JsonReader, JsonWriter> {
         if (values == null) {
             new JsonStreamWriter(tiny, out).writeNull();
         } else {
-            ((AnyEncoder) factory.getAnyEncoder()).convertMapTo(new JsonStreamWriter(tiny, out), values);
+            final JsonWriter writer = writerPool.get().tiny(tiny);
+            ((AnyEncoder) factory.getAnyEncoder()).convertMapTo(writer, values);
+            byte[] bs = writer.toBytes();
+            writerPool.accept(writer);
+            try {
+                out.write(bs);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
         }
     }
 

src/org/redkale/convert/json/JsonWriter.java

@@ -59,6 +59,7 @@ public class JsonWriter extends Writer {
      * 返回指定至少指定长度的缓冲区
      *
      * @param len
+     *
      * @return
      */
     private char[] expand(int len) {
@@ -108,6 +109,10 @@ public class JsonWriter extends Writer {
         return new ByteBuffer[]{ByteBuffer.wrap(Utility.encodeUTF8(content, 0, count))};
     }
 
+    public byte[] toBytes() {
+        return Utility.encodeUTF8(content, 0, count);
+    }
+
     public int count() {
         return this.count;
     }

src/org/redkale/net/AsyncConnection.java

@@ -31,10 +31,10 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
     protected volatile long writetime;
 
     //关闭数
-    AtomicLong closedCounter;
+    protected AtomicLong closedCounter;
 
     //在线数
-    AtomicLong livingCounter;
+    protected AtomicLong livingCounter;
 
     public final long getLastReadTime() {
         return readtime;

src/org/redkale/net/Context.java

@@ -130,13 +130,13 @@ public class Context {
     }
 
     public void offerBuffer(ByteBuffer buffer) {
-        bufferPool.offer(buffer);
+        bufferPool.accept(buffer);
     }
 
     public void offerBuffer(ByteBuffer... buffers) {
         if (buffers == null) return;
         for (ByteBuffer buffer : buffers) {
-            bufferPool.offer(buffer);
+            bufferPool.accept(buffer);
         }
     }
 

src/org/redkale/net/PrepareRunner.java

@@ -48,14 +48,16 @@ public final class PrepareRunner implements Runnable {
             }
             return;
         }
-        final ByteBuffer buffer = context.pollBuffer();
+        final Response response = responsePool.get();
+        final ByteBuffer buffer = response.request.pollReadBuffer();
         try {
             channel.read(buffer, null, new CompletionHandler<Integer, Void>() {
                 @Override
                 public void completed(Integer count, Void attachment1) {
                     if (count < 1 && buffer.remaining() == buffer.limit()) {
                         try {
-                            context.offerBuffer(buffer);
+                            response.request.offerReadBuffer(buffer);
+                            response.finish(true);
                             channel.close();
                         } catch (Exception e) {
                             context.logger.log(Level.FINEST, "PrepareRunner close channel erroneous on no read bytes", e);
@@ -69,7 +71,6 @@ public final class PrepareRunner implements Runnable {
 //                        System.println(new String(bs));
 //                    }
                     buffer.flip();
-                    final Response response = responsePool.get();
                     response.init(channel);
                     try {
                         prepare.prepare(buffer, response.request, response);
@@ -81,7 +82,8 @@ public final class PrepareRunner implements Runnable {
 
                 @Override
                 public void failed(Throwable exc, Void attachment2) {
-                    context.offerBuffer(buffer);
+                    response.request.offerReadBuffer(buffer);
+                    response.finish(true);
                     try {
                         channel.close();
                     } catch (Exception e) {
@@ -90,7 +92,8 @@ public final class PrepareRunner implements Runnable {
                 }
             });
         } catch (Exception te) {
-            context.offerBuffer(buffer);
+            response.request.offerReadBuffer(buffer);
+            response.finish(true);
             try {
                 channel.close();
             } catch (Exception e) {

src/org/redkale/net/PrepareServlet.java

@@ -214,11 +214,11 @@ public abstract class PrepareServlet<K extends Serializable, C extends Context,
         executeCounter.incrementAndGet();
         final int rs = request.readHeader(buffer);
         if (rs < 0) {
-            response.context.offerBuffer(buffer);
+            request.offerReadBuffer(buffer);
             if (rs != Integer.MIN_VALUE) illRequestCounter.incrementAndGet();
             response.finish(true);
         } else if (rs == 0) {
-            response.context.offerBuffer(buffer);
+            request.offerReadBuffer(buffer);
             request.prepare();
             response.filter = this.headFilter;
             response.servlet = this;
@@ -236,7 +236,7 @@ public abstract class PrepareServlet<K extends Serializable, C extends Context,
                         buffer.clear();
                         request.channel.read(buffer, buffer, this);
                     } else {
-                        response.context.offerBuffer(buffer);
+                        request.offerReadBuffer(buffer);
                         request.prepare();
                         try {
                             response.filter = PrepareServlet.this.headFilter;
@@ -253,7 +253,7 @@ public abstract class PrepareServlet<K extends Serializable, C extends Context,
                 @Override
                 public void failed(Throwable exc, ByteBuffer attachment) {
                     illRequestCounter.incrementAndGet();
-                    response.context.offerBuffer(buffer);
+                    request.offerReadBuffer(buffer);
                     response.finish(true);
                     if (exc != null) request.context.logger.log(Level.FINER, "Servlet read channel erroneous, forece to close channel ", exc);
                 }

src/org/redkale/net/ProtocolServer.java

@@ -217,7 +217,7 @@ public abstract class ProtocolServer {
                     AsyncConnection conn = AsyncConnection.create(channel, null, context.readTimeoutSecond, context.writeTimeoutSecond);
                     conn.livingCounter = livingCounter;
                     conn.closedCounter = closedCounter;
-                    context.submitAsync(new PrepareRunner(context, conn, null));
+                    context.runAsync(new PrepareRunner(context, conn, null));
                 }
 
                 @Override

src/org/redkale/net/Request.java

@@ -33,6 +33,8 @@ public abstract class Request<C extends Context> {
 
     protected AsyncConnection channel;
 
+    protected ByteBuffer readBuffer;
+
     /**
      * properties 与 attributes 的区别在于：调用recycle时， attributes会被清空而properties会保留;
      * properties 通常存放需要永久绑定在request里的一些对象
@@ -43,10 +45,28 @@ public abstract class Request<C extends Context> {
 
     protected Request(C context) {
         this.context = context;
+        this.readBuffer = context.pollBuffer();
         this.bsonConvert = context.getBsonConvert();
         this.jsonConvert = context.getJsonConvert();
     }
 
+    protected ByteBuffer pollReadBuffer() {
+        ByteBuffer buffer = this.readBuffer;
+        this.readBuffer = null;
+        if (buffer == null) buffer = context.pollBuffer();
+        return buffer;
+    }
+
+    protected void offerReadBuffer(ByteBuffer buffer) {
+        if (buffer == null) return;
+        if (this.readBuffer == null) {
+            buffer.clear();
+            this.readBuffer = buffer;
+        } else {
+            context.offerBuffer(buffer);
+        }
+    }
+
     /**
      * 返回值：Integer.MIN_VALUE: 帧数据； -1：数据不合法； 0：解析完毕； &gt;0: 需再读取的字节数。
      *

src/org/redkale/net/Response.java

@@ -8,7 +8,7 @@ package org.redkale.net;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.CompletionHandler;
-import java.util.function.BiConsumer;
+import java.util.function.*;
 import java.util.logging.Level;
 
 /**
@@ -30,6 +30,10 @@ public abstract class Response<C extends Context, R extends Request<C>> {
 
     protected AsyncConnection channel;
 
+    protected ByteBuffer writeHeadBuffer;
+
+    protected ByteBuffer writeBodyBuffer;
+
     private boolean inited = true;
 
     protected Object output; //输出的结果对象
@@ -40,6 +44,8 @@ public abstract class Response<C extends Context, R extends Request<C>> {
 
     protected Servlet<C, R, ? extends Response<C, R>> servlet;
 
+    private Supplier<ByteBuffer> bodyBufferSupplier;
+
     private final CompletionHandler finishHandler = new CompletionHandler<Integer, ByteBuffer>() {
 
         @Override
@@ -47,17 +53,31 @@ public abstract class Response<C extends Context, R extends Request<C>> {
             if (attachment.hasRemaining()) {
                 channel.write(attachment, attachment, this);
             } else {
-                context.offerBuffer(attachment);
+                offerResponseBuffer(attachment);
                 finish();
             }
         }
 
         @Override
         public void failed(Throwable exc, ByteBuffer attachment) {
-            context.offerBuffer(attachment);
+            offerResponseBuffer(attachment);
             finish(true);
         }
 
+        private void offerResponseBuffer(ByteBuffer attachment) {
+            if (writeHeadBuffer == null) {
+                if (context.bufferPool.getRecyclerPredicate().test(attachment)) {
+                    writeHeadBuffer = attachment;
+                }
+            } else if (writeBodyBuffer == null) {
+                if (context.bufferPool.getRecyclerPredicate().test(attachment)) {
+                    writeBodyBuffer = attachment;
+                }
+            } else {
+                context.offerBuffer(attachment);
+            }
+        }
+
     };
 
     private final CompletionHandler finishHandler2 = new CompletionHandler<Integer, ByteBuffer[]>() {
@@ -74,26 +94,66 @@ public abstract class Response<C extends Context, R extends Request<C>> {
             if (index >= 0) {
                 channel.write(attachments, index, attachments.length - index, attachments, this);
             } else {
-                for (ByteBuffer attachment : attachments) {
-                    context.offerBuffer(attachment);
-                }
+                offerResponseBuffer(attachments);
                 finish();
             }
         }
 
         @Override
         public void failed(Throwable exc, final ByteBuffer[] attachments) {
-            for (ByteBuffer attachment : attachments) {
-                context.offerBuffer(attachment);
-            }
+            offerResponseBuffer(attachments);
             finish(true);
         }
 
+        private void offerResponseBuffer(ByteBuffer[] attachments) {
+            int start = 0;
+            if (writeHeadBuffer == null && attachments.length > start) {
+                if (context.bufferPool.getRecyclerPredicate().test(attachments[start])) {
+                    writeHeadBuffer = attachments[start];
+                    start++;
+                }
+            }
+            if (writeBodyBuffer == null && attachments.length > start) {
+                if (context.bufferPool.getRecyclerPredicate().test(attachments[start])) {
+                    writeBodyBuffer = attachments[start];
+                    start++;
+                }
+            }
+            for (int i = start; i < attachments.length; i++) {
+                context.offerBuffer(attachments[i]);
+            }
+        }
     };
 
     protected Response(C context, final R request) {
         this.context = context;
         this.request = request;
+        this.writeHeadBuffer = context.pollBuffer();
+        this.writeBodyBuffer = context.pollBuffer();
+        this.bodyBufferSupplier = () -> {
+            ByteBuffer buffer = writeBodyBuffer;
+            if (buffer == null) return context.pollBuffer();
+            writeBodyBuffer = null;
+            return buffer;
+        };
+    }
+
+    protected ByteBuffer pollWriteReadBuffer() {
+        ByteBuffer buffer = this.writeHeadBuffer;
+        this.writeHeadBuffer = null;
+        if (buffer == null) buffer = context.pollBuffer();
+        return buffer;
+    }
+
+    protected ByteBuffer pollWriteBodyBuffer() {
+        ByteBuffer buffer = this.writeBodyBuffer;
+        this.writeBodyBuffer = null;
+        if (buffer == null) buffer = context.pollBuffer();
+        return buffer;
+    }
+
+    protected Supplier<ByteBuffer> getBodyBufferSupplier() {
+        return bodyBufferSupplier;
     }
 
     protected AsyncConnection removeChannel() {
@@ -195,7 +255,7 @@ public abstract class Response<C extends Context, R extends Request<C>> {
         if (!this.inited) return; //避免重复关闭
         //System.println("耗时: " + (System.currentTimeMillis() - request.createtime));
         if (kill) refuseAlive();
-        this.context.responsePool.offer(this);
+        this.context.responsePool.accept(this);
     }
 
     public void finish(final byte[] bs) {

src/org/redkale/net/Transport.java

@@ -178,7 +178,7 @@ public final class Transport {
     }
 
     public void offerBuffer(ByteBuffer buffer) {
-        bufferPool.offer(buffer);
+        bufferPool.accept(buffer);
     }
 
     public void offerBuffer(ByteBuffer... buffers) {

src/org/redkale/net/TransportFactory.java

@@ -29,6 +29,12 @@ import org.redkale.util.*;
  */
 public class TransportFactory {
 
+    @Comment("默认TCP读取超时秒数")
+    public static int DEFAULT_READTIMEOUTSECOND = 6;
+
+    @Comment("默认TCP写入超时秒数")
+    public static int DEFAULT_WRITETIMEOUTSECOND = 6;
+
     public static final String NAME_PINGINTERVAL = "pinginterval";
 
     protected static final Logger logger = Logger.getLogger(TransportFactory.class.getSimpleName());
@@ -55,6 +61,12 @@ public class TransportFactory {
     //心跳周期， 单位：秒
     protected int pinginterval;
 
+    //TCP读取超时秒数
+    protected int readTimeoutSecond;
+
+    //TCP写入超时秒数
+    protected int writeTimeoutSecond;
+
     //ping的定时器
     private ScheduledThreadPoolExecutor pingScheduler;
 
@@ -68,15 +80,18 @@ public class TransportFactory {
     protected final TransportStrategy strategy;
 
     protected TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
-        final TransportStrategy strategy) {
+        int readTimeoutSecond, int writeTimeoutSecond, final TransportStrategy strategy) {
         this.executor = executor;
         this.bufferPool = bufferPool;
         this.channelGroup = channelGroup;
+        this.readTimeoutSecond = readTimeoutSecond;
+        this.writeTimeoutSecond = writeTimeoutSecond;
         this.strategy = strategy;
     }
 
-    protected TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup) {
-        this(executor, bufferPool, channelGroup, null);
+    protected TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
+        int readTimeoutSecond, int writeTimeoutSecond) {
+        this(executor, bufferPool, channelGroup, readTimeoutSecond, writeTimeoutSecond, null);
     }
 
     public void init(AnyValue conf, ByteBuffer pingBuffer, int pongLength) {
@@ -100,10 +115,14 @@ public class TransportFactory {
     }
 
     public static TransportFactory create(int threads) {
-        return create(threads, threads * 2, 8 * 1024);
+        return create(threads, threads * 2, 8 * 1024, DEFAULT_READTIMEOUTSECOND, DEFAULT_WRITETIMEOUTSECOND);
     }
 
     public static TransportFactory create(int threads, int bufferPoolSize, int bufferCapacity) {
+        return create(threads, bufferPoolSize, bufferCapacity, DEFAULT_READTIMEOUTSECOND, DEFAULT_WRITETIMEOUTSECOND);
+    }
+
+    public static TransportFactory create(int threads, int bufferPoolSize, int bufferCapacity, int readTimeoutSecond, int writeTimeoutSecond) {
         final ObjectPool<ByteBuffer> transportPool = new ObjectPool<>(new AtomicLong(), new AtomicLong(), bufferPoolSize,
             (Object... params) -> ByteBuffer.allocateDirect(bufferCapacity), null, (e) -> {
                 if (e == null || e.isReadOnly() || e.capacity() != bufferCapacity) return false;
@@ -123,16 +142,21 @@ public class TransportFactory {
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
-        return create(transportExec, transportPool, transportGroup);
+        return create(transportExec, transportPool, transportGroup, readTimeoutSecond, writeTimeoutSecond);
     }
 
     public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup) {
-        return new TransportFactory(executor, bufferPool, channelGroup, null);
+        return new TransportFactory(executor, bufferPool, channelGroup, DEFAULT_READTIMEOUTSECOND, DEFAULT_WRITETIMEOUTSECOND, null);
     }
 
     public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
-        final TransportStrategy strategy) {
-        return new TransportFactory(executor, bufferPool, channelGroup, strategy);
+        int readTimeoutSecond, int writeTimeoutSecond) {
+        return new TransportFactory(executor, bufferPool, channelGroup, readTimeoutSecond, writeTimeoutSecond, null);
+    }
+
+    public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
+        int readTimeoutSecond, int writeTimeoutSecond, final TransportStrategy strategy) {
+        return new TransportFactory(executor, bufferPool, channelGroup, readTimeoutSecond, writeTimeoutSecond, strategy);
     }
 
     public Transport createTransportTCP(String name, final InetSocketAddress clientAddress, final Collection<InetSocketAddress> addresses) {
@@ -308,7 +332,7 @@ public class TransportFactory {
                                     @Override
                                     public void completed(Integer result, ByteBuffer attachment) {
                                         if (counter > 3) {
-                                            bufferPool.offer(attachment);
+                                            bufferPool.accept(attachment);
                                             localconn.dispose();
                                             return;
                                         }
@@ -317,7 +341,7 @@ public class TransportFactory {
                                             localconn.read(pongBuffer, pongBuffer, this);
                                             return;
                                         }
-                                        bufferPool.offer(attachment);
+                                        bufferPool.accept(attachment);
                                         localqueue.offer(localconn);
                                     }
 

src/org/redkale/net/http/HttpResponse.java

@@ -15,6 +15,7 @@ import java.text.*;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiFunction;
 import java.util.logging.Level;
 import org.redkale.convert.*;
 import org.redkale.convert.json.JsonConvert;
@@ -34,15 +35,6 @@ import org.redkale.util.*;
  */
 public class HttpResponse extends Response<HttpContext, HttpRequest> {
 
-    /**
-     * HttpResponse.finish 方法内调用
-     * 主要给@HttpCacheable使用
-     */
-    protected static interface BufferHandler {
-
-        public ByteBuffer[] execute(final HttpResponse response, final ByteBuffer[] buffers);
-    }
-
     private static final ByteBuffer buffer304 = ByteBuffer.wrap("HTTP/1.1 304 Not Modified\r\nContent-Length:0\r\n\r\n".getBytes()).asReadOnlyBuffer();
 
     private static final ByteBuffer buffer404 = ByteBuffer.wrap("HTTP/1.1 404 Not Found\r\nContent-Length:0\r\n\r\n".getBytes()).asReadOnlyBuffer();
@@ -117,7 +109,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
 
     private boolean headsended = false;
 
-    private BufferHandler bufferHandler;
+    private BiFunction<HttpResponse, ByteBuffer[], ByteBuffer[]> bufferHandler;
     //------------------------------------------------
 
     private final DefaultAnyValue header = new DefaultAnyValue();
@@ -148,6 +140,10 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
         return super.removeChannel();
     }
 
+    protected AsyncConnection getChannel() {
+        return channel;
+    }
+
     @Override
     protected boolean recycle() {
         boolean rs = super.recycle();
@@ -229,7 +225,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
         return Utility.createAsyncHandler((v, a) -> {
             finish(v);
         }, (t, a) -> {
-            request.getContext().getLogger().log(Level.WARNING, "Servlet occur, forece to close channel. request = " + request + ", result is CompletionHandler", (Throwable)t);
+            request.getContext().getLogger().log(Level.WARNING, "Servlet occur, forece to close channel. request = " + request + ", result is CompletionHandler", (Throwable) t);
             finish(500, null);
         });
     }
@@ -258,7 +254,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     public void finishJson(final Object obj) {
         this.contentType = "text/plain; charset=utf-8";
         if (this.recycleListener != null) this.output = obj;
-        finish(request.getJsonConvert().convertTo(context.getBufferSupplier(), obj));
+        finish(request.getJsonConvert().convertTo(getBodyBufferSupplier(), obj));
     }
 
     /**
@@ -270,7 +266,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     public void finishMapJson(final Object... objs) {
         this.contentType = "text/plain; charset=utf-8";
         if (this.recycleListener != null) this.output = objs;
-        finish(request.getJsonConvert().convertMapTo(context.getBufferSupplier(), objs));
+        finish(request.getJsonConvert().convertMapTo(getBodyBufferSupplier(), objs));
     }
 
     /**
@@ -282,7 +278,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     public void finishJson(final JsonConvert convert, final Object obj) {
         this.contentType = "text/plain; charset=utf-8";
         if (this.recycleListener != null) this.output = obj;
-        finish(convert.convertTo(context.getBufferSupplier(), obj));
+        finish(convert.convertTo(getBodyBufferSupplier(), obj));
     }
 
     /**
@@ -295,7 +291,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     public void finishMapJson(final JsonConvert convert, final Object... objs) {
         this.contentType = "text/plain; charset=utf-8";
         if (this.recycleListener != null) this.output = objs;
-        finish(convert.convertMapTo(context.getBufferSupplier(), objs));
+        finish(convert.convertMapTo(getBodyBufferSupplier(), objs));
     }
 
     /**
@@ -307,7 +303,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     public void finishJson(final Type type, final Object obj) {
         this.contentType = "text/plain; charset=utf-8";
         this.output = obj;
-        finish(request.getJsonConvert().convertTo(context.getBufferSupplier(), type, obj));
+        finish(request.getJsonConvert().convertTo(getBodyBufferSupplier(), type, obj));
     }
 
     /**
@@ -320,7 +316,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     public void finishJson(final JsonConvert convert, final Type type, final Object obj) {
         this.contentType = "text/plain; charset=utf-8";
         if (this.recycleListener != null) this.output = obj;
-        finish(convert.convertTo(context.getBufferSupplier(), type, obj));
+        finish(convert.convertTo(getBodyBufferSupplier(), type, obj));
     }
 
     /**
@@ -331,7 +327,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     public void finishJson(final Object... objs) {
         this.contentType = "text/plain; charset=utf-8";
         if (this.recycleListener != null) this.output = objs;
-        finish(request.getJsonConvert().convertTo(context.getBufferSupplier(), objs));
+        finish(request.getJsonConvert().convertTo(getBodyBufferSupplier(), objs));
     }
 
     /**
@@ -346,7 +342,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
             this.header.addValue("retcode", String.valueOf(ret.getRetcode()));
             this.header.addValue("retinfo", ret.getRetinfo());
         }
-        finish(request.getJsonConvert().convertTo(context.getBufferSupplier(), ret));
+        finish(request.getJsonConvert().convertTo(getBodyBufferSupplier(), ret));
     }
 
     /**
@@ -362,7 +358,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
             this.header.addValue("retcode", String.valueOf(ret.getRetcode()));
             this.header.addValue("retinfo", ret.getRetinfo());
         }
-        finish(convert.convertTo(context.getBufferSupplier(), ret));
+        finish(convert.convertTo(getBodyBufferSupplier(), ret));
     }
 
     /**
@@ -432,7 +428,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
         } else if (obj instanceof CompletableFuture) {
             ((CompletableFuture) obj).whenComplete((v, e) -> {
                 if (e != null) {
-                    context.getLogger().log(Level.WARNING, "Servlet occur, forece to close channel. request = " + request + ", result is CompletableFuture", (Throwable)e);
+                    context.getLogger().log(Level.WARNING, "Servlet occur, forece to close channel. request = " + request + ", result is CompletableFuture", (Throwable) e);
                     finish(500, null);
                     return;
                 }
@@ -471,8 +467,8 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
                     this.header.addValue("retcode", String.valueOf(ret.getRetcode())).addValue("retinfo", ret.getRetinfo());
                 }
             }
-            ByteBuffer[] buffers = type == null ? convert.convertTo(context.getBufferSupplier(), obj)
-                : convert.convertTo(context.getBufferSupplier(), type, obj);
+            ByteBuffer[] buffers = type == null ? convert.convertTo(getBodyBufferSupplier(), obj)
+                : convert.convertTo(getBodyBufferSupplier(), type, obj);
             finish(buffers);
         }
     }
@@ -494,7 +490,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
         }
         if (context.getCharset() == null) {
             if (bufferHandler != null) {
-                bufferHandler.execute(this, new ByteBuffer[]{ByteBuffer.wrap(Utility.encodeUTF8(obj))});
+                bufferHandler.apply(this, new ByteBuffer[]{ByteBuffer.wrap(Utility.encodeUTF8(obj))});
             }
             final char[] chars = Utility.charArray(obj);
             this.contentLength = Utility.encodeUTF8Length(chars);
@@ -509,7 +505,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
         } else {
             ByteBuffer buffer = context.getCharset().encode(obj);
             if (bufferHandler != null) {
-                ByteBuffer[] bufs = bufferHandler.execute(this, new ByteBuffer[]{buffer});
+                ByteBuffer[] bufs = bufferHandler.apply(this, new ByteBuffer[]{buffer});
                 if (bufs != null) buffer = bufs[0];
             }
             this.contentLength = buffer.remaining();
@@ -554,7 +550,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     @Override
     public void finish(final byte[] bs) {
         if (isClosed()) return; //避免重复关闭
-        if (this.context.getBufferCapacity() == bs.length) {
+        if (this.context.getBufferCapacity() >= bs.length) {
             ByteBuffer buffer = this.context.pollBuffer();
             buffer.put(bs);
             buffer.flip();
@@ -617,7 +613,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     public void finish(boolean kill, ByteBuffer... buffers) {
         if (isClosed()) return; //避免重复关闭
         if (bufferHandler != null) {
-            ByteBuffer[] bufs = bufferHandler.execute(this, buffers);
+            ByteBuffer[] bufs = bufferHandler.apply(this, buffers);
             if (bufs != null) buffers = bufs;
         }
         if (kill) refuseAlive();
@@ -797,9 +793,10 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
         this.channel.write(hbuffer, hbuffer, new TransferFileHandler(file, offset, length));
     }
 
-    private ByteBuffer createHeader() {
+    //Header大小不能超过一个ByteBuffer的容量
+    protected ByteBuffer createHeader() {
         this.headsended = true;
-        ByteBuffer buffer = this.context.pollBuffer();
+        ByteBuffer buffer = this.pollWriteReadBuffer();
         buffer.put(("HTTP/1.1 " + this.status + " " + (this.status == 200 ? "OK" : httpCodes.get(this.status)) + "\r\n").getBytes());
 
         buffer.put(("Content-Type: " + (this.contentType == null ? "text/plain; charset=utf-8" : this.contentType) + "\r\n").getBytes());
@@ -1011,7 +1008,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
      *
      * @return 拦截器
      */
-    protected BufferHandler getBufferHandler() {
+    protected BiFunction<HttpResponse, ByteBuffer[], ByteBuffer[]> getBufferHandler() {
         return bufferHandler;
     }
 
@@ -1020,7 +1017,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
      *
      * @param bufferHandler 拦截器
      */
-    protected void setBufferHandler(BufferHandler bufferHandler) {
+    protected void setBufferHandler(BiFunction<HttpResponse, ByteBuffer[], ByteBuffer[]> bufferHandler) {
         this.bufferHandler = bufferHandler;
     }
 

src/org/redkale/net/http/HttpServlet.java

@@ -10,6 +10,7 @@ import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiFunction;
 import jdk.internal.org.objectweb.asm.*;
 import static jdk.internal.org.objectweb.asm.ClassWriter.COMPUTE_FRAMES;
 import static jdk.internal.org.objectweb.asm.Opcodes.*;
@@ -315,7 +316,7 @@ public class HttpServlet extends Servlet<HttpContext, HttpRequest, HttpResponse>
             return false;
         }
 
-        public final HttpResponse.BufferHandler cacheHandler;
+        public final BiFunction<HttpResponse, ByteBuffer[], ByteBuffer[]> cacheHandler;
 
         public final ConcurrentHashMap<String, CacheEntry> cache;
 

src/org/redkale/net/http/Rest.java

@@ -956,7 +956,7 @@ public final class Rest {
                 av1.visitEnd();
                 av0.visitEnd();
             }
-
+            int uploadLocal = 0;
             if (mupload != null) { //存在文件上传
                 if (muploadType == byte[].class) {
                     mv.visitVarInsn(ALOAD, 1);
@@ -966,6 +966,7 @@ public final class Rest {
                     mv.visitLdcInsn(mupload.contentTypeReg());
                     mv.visitMethodInsn(INVOKEVIRTUAL, "org/redkale/net/http/MultiContext", "partsFirstBytes", "(JLjava/lang/String;Ljava/lang/String;)[B", false);
                     mv.visitVarInsn(ASTORE, maxLocals);
+                    uploadLocal = maxLocals;
                 } else if (muploadType == File.class) {
                     mv.visitVarInsn(ALOAD, 1);
                     mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getMultiContext", "()Lorg/redkale/net/http/MultiContext;", false);
@@ -976,6 +977,7 @@ public final class Rest {
                     mv.visitLdcInsn(mupload.contentTypeReg());
                     mv.visitMethodInsn(INVOKEVIRTUAL, "org/redkale/net/http/MultiContext", "partsFirstFile", "(Ljava/io/File;JLjava/lang/String;Ljava/lang/String;)Ljava/io/File;", false);
                     mv.visitVarInsn(ASTORE, maxLocals);
+                    uploadLocal = maxLocals;
                 } else if (muploadType == File[].class) { //File[]
                     mv.visitVarInsn(ALOAD, 1);
                     mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getMultiContext", "()Lorg/redkale/net/http/MultiContext;", false);
@@ -986,6 +988,7 @@ public final class Rest {
                     mv.visitLdcInsn(mupload.contentTypeReg());
                     mv.visitMethodInsn(INVOKEVIRTUAL, "org/redkale/net/http/MultiContext", "partsFiles", "(Ljava/io/File;JLjava/lang/String;Ljava/lang/String;)[Ljava/io/File;", false);
                     mv.visitVarInsn(ASTORE, maxLocals);
+                    uploadLocal = maxLocals;
                 }
                 maxLocals++;
             }
@@ -1402,7 +1405,7 @@ public final class Rest {
                     } while ((loop = loop.getSuperclass()) != Object.class);
 
                     if (!attrParaNames.isEmpty()) { //参数存在 RestHeader、RestCookie、RestSessionid、RestAddress、RestBody字段
-                        mv.visitVarInsn(ALOAD, maxLocals);
+                        mv.visitVarInsn(ALOAD, maxLocals); //加载JsonBean
                         Label lif = new Label();
                         mv.visitJumpInsn(IFNULL, lif);  //if(bean != null) {
                         for (Map.Entry<String, Object[]> en : attrParaNames.entrySet()) {
@@ -1410,12 +1413,22 @@ public final class Rest {
                             mv.visitVarInsn(ALOAD, 0);
                             mv.visitFieldInsn(GETFIELD, newDynName, en.getKey(), attrDesc);
                             mv.visitVarInsn(ALOAD, maxLocals);
-                            boolean upload = en.getKey().contains("_upload");
-                            mv.visitVarInsn(ALOAD, upload ? (maxLocals - 1) : 1);
+                            mv.visitVarInsn(ALOAD, en.getKey().contains("_upload") ? uploadLocal : 1);
                             if (en.getKey().contains("_header_")) {
-                                mv.visitLdcInsn(en.getValue()[0].toString());
-                                mv.visitLdcInsn("");
-                                mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getHeader", "(Ljava/lang/String;Ljava/lang/String;)Ljava/lang/String;", false);
+                                String headerkey = en.getValue()[0].toString();
+                                if ("Host".equalsIgnoreCase(headerkey)) {
+                                    mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getHost", "()Ljava/lang/String;", false);
+                                } else if ("Content-Type".equalsIgnoreCase(headerkey)) {
+                                    mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getContentType", "()Ljava/lang/String;", false);
+                                } else if ("Connection".equalsIgnoreCase(headerkey)) {
+                                    mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getConnection", "()Ljava/lang/String;", false);
+                                } else if ("Method".equalsIgnoreCase(headerkey)) {
+                                    mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getMethod", "()Ljava/lang/String;", false);
+                                } else {
+                                    mv.visitLdcInsn(headerkey);
+                                    mv.visitLdcInsn("");
+                                    mv.visitMethodInsn(INVOKEVIRTUAL, reqInternalName, "getHeader", "(Ljava/lang/String;Ljava/lang/String;)Ljava/lang/String;", false);
+                                }
                             } else if (en.getKey().contains("_cookie_")) {
                                 mv.visitLdcInsn(en.getValue()[0].toString());
                                 mv.visitLdcInsn("");
@@ -1562,7 +1575,7 @@ public final class Rest {
                     mv.visitVarInsn(ALOAD, 0);
                     mv.visitFieldInsn(GETFIELD, newDynName, REST_JSONCONVERT_FIELD_PREFIX + restConverts.size(), jsonConvertDesc);
                     mv.visitVarInsn(ALOAD, maxLocals);
-                    mv.visitMethodInsn(INVOKEVIRTUAL, respInternalName, "finish", "(" + convertDesc + typeDesc + "Ljava/lang/Object;)V", false);
+                    mv.visitMethodInsn(INVOKEVIRTUAL, respInternalName, "finish", "(" + convertDesc + "Ljava/lang/Object;)V", false);
                 } else {
                     mv.visitVarInsn(ALOAD, maxLocals);
                     mv.visitMethodInsn(INVOKEVIRTUAL, respInternalName, "finish", "(Ljava/lang/Object;)V", false);

src/org/redkale/net/http/WebSocket.java

@@ -64,6 +64,9 @@ public abstract class WebSocket<G extends Serializable, T> {
     @Comment("WebSocket已离线")
     public static final int RETCODE_WSOFFLINE = 1 << 8; //256
 
+    @Comment("WebSocket将延迟发送")
+    public static final int RETCODE_DEAYSEND = 1 << 9; //512
+
     WebSocketRunner _runner; //不可能为空 
 
     WebSocketEngine _engine; //不可能为空 
@@ -90,6 +93,8 @@ public abstract class WebSocket<G extends Serializable, T> {
 
     private Map<String, Object> attributes = new HashMap<>(); //非线程安全
 
+    List<WebSocketPacket> delayPackets;
+
     protected WebSocket() {
     }
 
@@ -225,6 +230,11 @@ public abstract class WebSocket<G extends Serializable, T> {
      * @return 0表示成功， 非0表示错误码
      */
     CompletableFuture<Integer> sendPacket(WebSocketPacket packet) {
+        if (this._runner == null) {
+            if (delayPackets == null) delayPackets = new ArrayList<>();
+            delayPackets.add(packet);
+            return CompletableFuture.completedFuture(RETCODE_DEAYSEND);
+        }
         CompletableFuture<Integer> rs = this._runner.sendMessage(packet);
         if (_engine.logger.isLoggable(Level.FINEST) && packet != WebSocketPacket.DEFAULT_PING_PACKET) {
             _engine.logger.finest("userid:" + getUserid() + " send websocket message(" + packet + ")" + " on " + this);

src/org/redkale/net/http/WebSocketNode.java

@@ -349,6 +349,7 @@ public abstract class WebSocketNode {
      */
     public final CompletableFuture<Integer> sendMessage(final Convert convert, final Object message0, final boolean last, final Serializable... userids) {
         if (userids == null || userids.length < 1) return CompletableFuture.completedFuture(RETCODE_GROUP_EMPTY);
+        if (message0 instanceof CompletableFuture) return ((CompletableFuture) message0).thenApply(msg -> sendMessage(convert, msg, last, userids));
         final Object message = (convert == null || message0 instanceof WebSocketPacket) ? message0 : ((convert instanceof TextConvert) ? new WebSocketPacket(((TextConvert) convert).convertTo(message0), last) : new WebSocketPacket(((BinaryConvert) convert).convertTo(message0), last));
         if (this.localEngine != null && this.sncpNodeAddresses == null) { //本地模式且没有分布式
             return this.localEngine.sendMessage(message, last, userids);
@@ -406,6 +407,7 @@ public abstract class WebSocketNode {
      * @return 为0表示成功， 其他值表示部分发送异常
      */
     public final CompletableFuture<Integer> broadcastMessage(final Convert convert, final Object message0, final boolean last) {
+        if (message0 instanceof CompletableFuture) return ((CompletableFuture) message0).thenApply(msg -> broadcastMessage(convert, msg, last));
         final Object message = (convert == null || message0 instanceof WebSocketPacket) ? message0 : ((convert instanceof TextConvert) ? new WebSocketPacket(((TextConvert) convert).convertTo(message0), last) : new WebSocketPacket(((BinaryConvert) convert).convertTo(message0), last));
         if (this.localEngine != null && this.sncpNodeAddresses == null) { //本地模式且没有分布式
             return this.localEngine.broadcastMessage(message, last);
@@ -429,7 +431,7 @@ public abstract class WebSocketNode {
 
     private CompletableFuture<Integer> sendOneMessage(final Object message, final boolean last, final Serializable userid) {
         if (message instanceof CompletableFuture) return ((CompletableFuture) message).thenApply(msg -> sendOneMessage(msg, last, userid));
-        if (logger.isLoggable(Level.FINEST)) logger.finest("websocket want send message {userid:" + userid + ", content:'" + message + "'} from locale node to " + ((this.localEngine != null) ? "locale" : "remote") + " engine");
+        if (logger.isLoggable(Level.FINEST)) logger.finest("websocket want send message {userid:" + userid + ", content:'" + JsonConvert.root().convertTo(message) + "'} from locale node to " + ((this.localEngine != null) ? "locale" : "remote") + " engine");
         CompletableFuture<Integer> localFuture = null;
         if (this.localEngine != null) localFuture = localEngine.sendMessage(message, last, userid);
         if (this.sncpNodeAddresses == null || this.remoteNode == null) {
@@ -442,7 +444,7 @@ public abstract class WebSocketNode {
         CompletableFuture<Integer> remoteFuture = addrsFuture.thenCompose((Collection<InetSocketAddress> addrs) -> {
             if (addrs == null || addrs.isEmpty()) {
                 if (logger.isLoggable(Level.FINER)) logger.finer("websocket not found userid:" + userid + " on any node ");
-                return CompletableFuture.completedFuture(0);
+                return CompletableFuture.completedFuture(RETCODE_GROUP_EMPTY);
             }
             if (logger.isLoggable(Level.FINEST)) logger.finest("websocket(localaddr=" + localSncpAddress + ") found userid:" + userid + " on " + addrs);
             CompletableFuture<Integer> future = null;
@@ -452,7 +454,7 @@ public abstract class WebSocketNode {
                 future = future == null ? remoteNode.sendMessage(addr, remoteMessage, last, userid)
                     : future.thenCombine(remoteNode.sendMessage(addr, remoteMessage, last, userid), (a, b) -> a | b);
             }
-            return future == null ? CompletableFuture.completedFuture(0) : future;
+            return future == null ? CompletableFuture.completedFuture(RETCODE_GROUP_EMPTY) : future;
         });
         return localFuture == null ? remoteFuture : localFuture.thenCombine(remoteFuture, (a, b) -> a | b);
     }

src/org/redkale/net/http/WebSocketRunner.java

@@ -433,6 +433,10 @@ class WebSocketRunner implements Runnable {
         return futureResult;
     }
 
+    public boolean isClosed() {
+        return closed;
+    }
+
     public void closeRunner(int code) {
         if (closed) return;
         synchronized (this) {

src/org/redkale/net/http/WebSocketServlet.java

@@ -178,7 +178,7 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl
             return;
         }
         sessionFuture.whenComplete((sessionid, ex) -> {
-            if (sessionid == null || ex != null) {
+            if ((sessionid == null && webSocket.delayPackets == null) || ex != null) {
                 if (debug || ex != null) logger.log(ex == null ? Level.FINEST : Level.FINE, "WebSocket connect abort, Not found sessionid or occur error. request=" + request, ex);
                 response.finish(true);
                 return;
@@ -195,39 +195,93 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl
             response.addHeader("Sec-WebSocket-Accept", Base64.getEncoder().encodeToString(bytes));
             response.sendBody((ByteBuffer) null, null, new CompletionHandler<Integer, Void>() {
 
+                WebSocketRunner temprunner = null;
+
                 @Override
                 public void completed(Integer result, Void attachment) {
                     HttpContext context = response.getContext();
-                    CompletableFuture<Serializable> userFuture = webSocket.createUserid();
-                    if (userFuture == null) {
-                        if (debug) logger.finest("WebSocket connect abort, Create userid abort. request = " + request);
-                        response.finish(true);
-                        return;
-                    }
-                    userFuture.whenComplete((userid, ex2) -> {
-                        if (userid == null || ex2 != null) {
-                            if (debug || ex2 != null) logger.log(ex2 == null ? Level.FINEST : Level.FINE, "WebSocket connect abort, Create userid abort. request = " + request, ex2);
+
+                    Runnable createUseridHandler = () -> {
+                        CompletableFuture<Serializable> userFuture = webSocket.createUserid();
+                        if (userFuture == null) {
+                            if (debug) logger.finest("WebSocket connect abort, Create userid abort. request = " + request);
                             response.finish(true);
                             return;
                         }
-                        webSocket._userid = userid;
-                        if (single && !anyuser) {
-                            WebSocketServlet.this.node.existsWebSocket(userid).whenComplete((rs, ex) -> {
-                                if (rs) webSocket.onSingleRepeatConnect();
-                                WebSocketServlet.this.node.localEngine.add(webSocket);
-                                WebSocketRunner runner = new WebSocketRunner(context, webSocket, restMessageConsumer, response.removeChannel());
-                                webSocket._runner = runner;
-                                context.runAsync(runner);
+                        userFuture.whenComplete((userid, ex2) -> {
+                            if ((userid == null && webSocket.delayPackets == null) || ex2 != null) {
+                                if (debug || ex2 != null) logger.log(ex2 == null ? Level.FINEST : Level.FINE, "WebSocket connect abort, Create userid abort. request = " + request, ex2);
                                 response.finish(true);
-                            });
-                        } else {
-                            WebSocketServlet.this.node.localEngine.add(webSocket);
-                            WebSocketRunner runner = new WebSocketRunner(context, webSocket, restMessageConsumer, response.removeChannel());
-                            webSocket._runner = runner;
-                            context.runAsync(runner);
-                            response.finish(true);
+                                return;
+                            }
+                            Runnable runHandler = () -> {
+                                temprunner = null;
+                                webSocket._userid = userid;
+                                if (single && !anyuser) {
+                                    WebSocketServlet.this.node.existsWebSocket(userid).whenComplete((rs, ex) -> {
+                                        if (rs) webSocket.onSingleRepeatConnect();
+                                        WebSocketServlet.this.node.localEngine.add(webSocket);
+                                        WebSocketRunner runner = new WebSocketRunner(context, webSocket, restMessageConsumer, response.removeChannel());
+                                        webSocket._runner = runner;
+                                        context.runAsync(runner);
+                                        response.finish(true);
+                                    });
+                                } else {
+                                    WebSocketServlet.this.node.localEngine.add(webSocket);
+                                    WebSocketRunner runner = new WebSocketRunner(context, webSocket, restMessageConsumer, response.removeChannel());
+                                    webSocket._runner = runner;
+                                    context.runAsync(runner);
+                                    response.finish(true);
+                                }
+                            };
+                            if (webSocket.delayPackets != null) { //存在待发送的消息
+                                if (temprunner == null) temprunner = new WebSocketRunner(context, webSocket, restMessageConsumer, response.getChannel());
+                                List<WebSocketPacket> delayPackets = webSocket.delayPackets;
+                                webSocket.delayPackets = null;
+                                CompletableFuture<Integer> cf = null;
+                                for (WebSocketPacket packet : delayPackets) {
+                                    if (cf == null) {
+                                        cf = temprunner.sendMessage(packet);
+                                    } else {
+                                        cf = cf.thenCombine(temprunner.sendMessage(packet), (a, b) -> a | b);
+                                    }
+                                }
+                                cf.whenComplete((Integer v, Throwable t) -> {
+                                    if (userid == null || t != null || (temprunner != null && temprunner.isClosed())) {
+                                        if (t != null) logger.log(Level.FINEST, "WebSocket connect abort, Response send delayPackets abort. request = " + request, t);
+                                        response.finish(true);
+                                    } else {
+                                        runHandler.run();
+                                    }
+                                });
+                            } else {
+                                runHandler.run();
+                            }
+                        });
+                    };
+                    if (webSocket.delayPackets != null) { //存在待发送的消息
+                        if (temprunner == null) temprunner = new WebSocketRunner(context, webSocket, restMessageConsumer, response.getChannel());
+                        List<WebSocketPacket> delayPackets = webSocket.delayPackets;
+                        webSocket.delayPackets = null;
+                        CompletableFuture<Integer> cf = null;
+                        for (WebSocketPacket packet : delayPackets) {
+                            if (cf == null) {
+                                cf = temprunner.sendMessage(packet);
+                            } else {
+                                cf = cf.thenCombine(temprunner.sendMessage(packet), (a, b) -> a | b);
+                            }
                         }
-                    });
+                        cf.whenComplete((Integer v, Throwable t) -> {
+                            if (sessionid == null || t != null || (temprunner != null && temprunner.isClosed())) {
+                                if (t != null) logger.log(Level.FINEST, "WebSocket connect abort, Response send delayPackets abort. request = " + request, t);
+                                response.finish(true);
+                            } else {
+                                createUseridHandler.run();
+                            }
+                        });
+                    } else {
+                        createUseridHandler.run();
+                    }
                 }
 
                 @Override

src/org/redkale/net/sncp/SncpResponse.java

@@ -54,7 +54,7 @@ public final class SncpResponse extends Response<SncpContext, SncpRequest> {
 
     public void finish(final int retcode, final BsonWriter out) {
         if (out == null) {
-            final ByteBuffer buffer = context.pollBuffer();
+            final ByteBuffer buffer = pollWriteReadBuffer();
             fillHeader(buffer, 0, retcode);
             finish(buffer);
             return;

src/org/redkale/service/RetResult.java

@@ -124,7 +124,8 @@ public class RetResult<T> {
      */
     public RetResult<T> attach(String key, Object value) {
         if (this.attach == null) this.attach = new HashMap<>();
-        this.attach.put(key, value == null ? null : String.valueOf(value));
+        boolean canstr = value != null && (value instanceof CharSequence || value.getClass().isPrimitive());
+        this.attach.put(key, value == null ? null : (canstr ? String.valueOf(value) : JsonConvert.root().convertTo(value)));
         return this;
     }
 

src/org/redkale/source/CacheMemorySource.java

@@ -667,6 +667,39 @@ public class CacheMemorySource<V extends Object> extends AbstractService impleme
         return (Collection<String>) getAndRefresh(key, expireSeconds);
     }
 
+    @Override
+    public boolean existsSetItem(final String key, final V value) {
+        Collection<V> list = getCollection(key);
+        return list != null && list.contains(value);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> existsSetItemAsync(final String key, final V value) {
+        return CompletableFuture.supplyAsync(() -> existsSetItem(key, value), getExecutor());
+    }
+
+    @Override
+    public boolean existsStringSetItem(final String key, final String value) {
+        Collection<String> list = getStringCollection(key);
+        return list != null && list.contains(value);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> existsStringSetItemAsync(final String key, final String value) {
+        return CompletableFuture.supplyAsync(() -> existsStringSetItem(key, value), getExecutor());
+    }
+
+    @Override
+    public boolean existsLongSetItem(final String key, final long value) {
+        Collection<Long> list = getLongCollection(key);
+        return list != null && list.contains(value);
+    }
+
+    @Override
+    public CompletableFuture<Boolean> existsLongSetItemAsync(final String key, final long value) {
+        return CompletableFuture.supplyAsync(() -> existsLongSetItem(key, value), getExecutor());
+    }
+
     @Override
     @RpcMultiRun
     public Collection<Long> getLongCollectionAndRefresh(final String key, final int expireSeconds) {

src/org/redkale/source/CacheSource.java

@@ -66,6 +66,8 @@ public interface CacheSource<V extends Object> {
 
     public void removeListItem(final String key, final V value);
 
+    public boolean existsSetItem(final String key, final V value);
+
     public void appendSetItem(final String key, final V value);
 
     public void removeSetItem(final String key, final V value);
@@ -92,6 +94,8 @@ public interface CacheSource<V extends Object> {
 
     public void removeStringListItem(final String key, final String value);
 
+    public boolean existsStringSetItem(final String key, final String value);
+
     public void appendStringSetItem(final String key, final String value);
 
     public void removeStringSetItem(final String key, final String value);
@@ -112,6 +116,8 @@ public interface CacheSource<V extends Object> {
 
     public void removeLongListItem(final String key, final long value);
 
+    public boolean existsLongSetItem(final String key, final long value);
+
     public void appendLongSetItem(final String key, final long value);
 
     public void removeLongSetItem(final String key, final long value);
@@ -151,6 +157,8 @@ public interface CacheSource<V extends Object> {
 
     public CompletableFuture<Void> removeListItemAsync(final String key, final V value);
 
+    public CompletableFuture<Boolean> existsSetItemAsync(final String key, final V value);
+
     public CompletableFuture<Void> appendSetItemAsync(final String key, final V value);
 
     public CompletableFuture<Void> removeSetItemAsync(final String key, final V value);
@@ -177,6 +185,8 @@ public interface CacheSource<V extends Object> {
 
     public CompletableFuture<Void> removeStringListItemAsync(final String key, final String value);
 
+    public CompletableFuture<Boolean> existsStringSetItemAsync(final String key, final String value);
+
     public CompletableFuture<Void> appendStringSetItemAsync(final String key, final String value);
 
     public CompletableFuture<Void> removeStringSetItemAsync(final String key, final String value);
@@ -197,6 +207,8 @@ public interface CacheSource<V extends Object> {
 
     public CompletableFuture<Void> removeLongListItemAsync(final String key, final long value);
 
+    public CompletableFuture<Boolean> existsLongSetItemAsync(final String key, final long value);
+
     public CompletableFuture<Void> appendLongSetItemAsync(final String key, final long value);
 
     public CompletableFuture<Void> removeLongSetItemAsync(final String key, final long value);

src/org/redkale/source/DataJdbcSource.java

@@ -109,27 +109,33 @@ public class DataJdbcSource extends AbstractService implements DataSource, DataC
     }
 
     @Override
+    @Local
     public void close() throws Exception {
         readPool.close();
         writePool.close();
     }
 
+    @Local
     public PoolJdbcSource getReadPoolJdbcSource() {
         return readPool;
     }
 
+    @Local
     public PoolJdbcSource getWritePoolJdbcSource() {
         return writePool;
     }
 
+    @Local
     public Connection createReadSQLConnection() {
         return readPool.poll();
     }
 
+    @Local
     public <T> Connection createWriteSQLConnection() {
         return writePool.poll();
     }
 
+    @Local
     public void closeSQLConnection(final Connection sqlconn) {
         if (sqlconn == null) return;
         try {
@@ -140,6 +146,7 @@ public class DataJdbcSource extends AbstractService implements DataSource, DataC
     }
 
     @Override
+    @Local
     public EntityInfo apply(Class t) {
         return loadEntityInfo(t);
     }
@@ -972,9 +979,8 @@ public class DataJdbcSource extends AbstractService implements DataSource, DataC
                 }
                 String sql = "UPDATE " + info.getTable(node) + " a " + (join1 == null ? "" : (", " + join1)) + " SET " + setsql
                     + ((where == null || where.length() == 0) ? (join2 == null ? "" : (" WHERE " + join2))
-                    : (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))));
-                //注：LIMIT 仅支持MySQL 且在多表关联式会异常， 该BUG尚未解决
-                sql += info.createSQLOrderby(flipper) + ((flipper == null || flipper.getLimit() < 1) ? "" : (" LIMIT " + flipper.getLimit()));
+                    : (" WHERE " + where + (join2 == null ? "" : (" AND " + join2))))
+                    + info.createSQLOrderby(flipper);
                 if (info.isLoggable(logger, Level.FINEST)) logger.finest(info.getType().getSimpleName() + " update sql=" + sql);
                 conn.setReadOnly(false);
                 if (blobs != null) {
@@ -2369,7 +2375,8 @@ public class DataJdbcSource extends AbstractService implements DataSource, DataC
      *
      * @return 结果数组
      */
-    public final int[] directExecute(String... sqls) {
+    @Local
+    public int[] directExecute(String... sqls) {
         Connection conn = createWriteSQLConnection();
         try {
             return directExecute(conn, sqls);
@@ -2385,7 +2392,8 @@ public class DataJdbcSource extends AbstractService implements DataSource, DataC
      * @param sql      SQL语句
      * @param consumer 回调函数
      */
-    public final void directQuery(String sql, Consumer<ResultSet> consumer) {
+    @Local
+    public void directQuery(String sql, Consumer<ResultSet> consumer) {
         final Connection conn = createReadSQLConnection();
         try {
             if (logger.isLoggable(Level.FINEST)) logger.finest("direct query sql=" + sql);

src/org/redkale/source/DataSource.java

@@ -300,7 +300,7 @@ public interface DataSource {
      * 更新符合过滤条件的记录的指定字段   <br>
      * Flipper中offset字段将被忽略   <br>
      * <b>注意</b>：Entity类中标记为&#064;Column(updatable=false)不会被更新   <br>
-     * 等价SQL: UPDATE {table} SET {column1} = {value1}, {column2} += {value2}, {column3} *= {value3}, &#183;&#183;&#183; WHERE {filter node} ORDER BY {flipper.sort} LIMIT {flipper.limit}  <br>
+     * 等价SQL: UPDATE {table} SET {column1} = {value1}, {column2} += {value2}, {column3} *= {value3}, &#183;&#183;&#183; WHERE {filter node} ORDER BY {flipper.sort}  <br>
      *
      * @param <T>     Entity泛型
      * @param clazz   Entity类
@@ -316,7 +316,7 @@ public interface DataSource {
      * 更新符合过滤条件的记录的指定字段   <br>
      * Flipper中offset字段将被忽略   <br>
      * <b>注意</b>：Entity类中标记为&#064;Column(updatable=false)不会被更新   <br>
-     * 等价SQL: UPDATE {table} SET {column1} = {value1}, {column2} += {value2}, {column3} *= {value3}, &#183;&#183;&#183; WHERE {filter node} ORDER BY {flipper.sort} LIMIT {flipper.limit}  <br>
+     * 等价SQL: UPDATE {table} SET {column1} = {value1}, {column2} += {value2}, {column3} *= {value3}, &#183;&#183;&#183; WHERE {filter node} ORDER BY {flipper.sort}  <br>
      *
      * @param <T>     Entity泛型
      * @param clazz   Entity类

src/org/redkale/source/EntityCache.java

@@ -286,7 +286,7 @@ public final class EntityCache<T> {
             rs = rs2;
         } else if (func == DISTINCTCOUNT) {
             Map rs2 = new LinkedHashMap();
-            rs.forEach((x, y) -> rs2.put(x, ((Set) y).size()));
+            rs.forEach((x, y) -> rs2.put(x, ((Set) y).size() + 0L));
             rs = rs2;
         }
         return rs;

src/org/redkale/util/ObjectPool.java

@@ -19,7 +19,7 @@ import java.util.logging.*;
  * @author zhangjx
  * @param <T> 对象池元素的数据类型
  */
-public final class ObjectPool<T> implements Supplier<T> {
+public final class ObjectPool<T> implements Supplier<T>, Consumer<T> {
 
     private static final Logger logger = Logger.getLogger(ObjectPool.class.getSimpleName());
 
@@ -67,6 +67,10 @@ public final class ObjectPool<T> implements Supplier<T> {
         this.creator = creator;
     }
 
+    public Predicate<T> getRecyclerPredicate() {
+        return recycler;
+    }
+
     @Override
     public T get() {
         T result = queue.poll();
@@ -78,21 +82,27 @@ public final class ObjectPool<T> implements Supplier<T> {
         return result;
     }
 
-    public void offer(final T e) {
+    @Override
+    public void accept(final T e) {
         if (e != null && recycler.test(e)) {
             if (cycleCounter != null) cycleCounter.incrementAndGet();
-            if (debug) {
-                for (T t : queue) {
-                    if (t == e) {
-                        logger.log(Level.WARNING, "[" + Thread.currentThread().getName() + "] repeat offer the same object(" + e + ")", new Exception());
-                        return;
-                    }
-                }
-            }
+//            if (debug) {
+//                for (T t : queue) {
+//                    if (t == e) {
+//                        logger.log(Level.WARNING, "[" + Thread.currentThread().getName() + "] repeat offer the same object(" + e + ")", new Exception());
+//                        return;
+//                    }
+//                }
+//            }
             queue.offer(e);
         }
     }
 
+    @Deprecated
+    public void offer(final T e) {
+        accept(e);
+    }
+
     public long getCreatCount() {
         return creatCounter.longValue();
     }

src/org/redkale/util/Redkale.java

@@ -17,7 +17,7 @@ public final class Redkale {
     }
 
     public static String getDotedVersion() {
-        return "1.8.7";
+        return "1.8.9";
     }
 
     public static int getMajorVersion() {

src/org/redkale/util/Utility.java

@@ -230,6 +230,191 @@ public final class Utility {
         return news;
     }
 
+    /**
+     * 获取int数组之和
+     *
+     * @param array 数组
+     *
+     * @return int
+     */
+    public static int sum(final int... array) {
+        if (array == null || array.length == 0) throw new NullPointerException("array is null or empty");
+        int sum = 0;
+        for (int i : array) {
+            sum += i;
+        }
+        return sum;
+    }
+
+    /**
+     * 获取long数组之和
+     *
+     * @param array 数组
+     *
+     * @return long
+     */
+    public static long sum(final long... array) {
+        if (array == null || array.length == 0) throw new NullPointerException("array is null or empty");
+        long sum = 0L;
+        for (long i : array) {
+            sum += i;
+        }
+        return sum;
+    }
+
+    /**
+     * 获取int数组最大值
+     *
+     * @param array 数组
+     *
+     * @return int
+     */
+    public static int max(final int... array) {
+        if (array == null || array.length == 0) throw new NullPointerException("array is null or empty");
+        int max = array[0];
+        for (int i : array) {
+            if (i > max) i = max;
+        }
+        return max;
+    }
+
+    /**
+     * 获取long数组最大值
+     *
+     * @param array 数组
+     *
+     * @return long
+     */
+    public static long max(final long... array) {
+        if (array == null || array.length == 0) throw new NullPointerException("array is null or empty");
+        long max = array[0];
+        for (long i : array) {
+            if (i > max) i = max;
+        }
+        return max;
+    }
+
+    /**
+     * 获取int数组最小值
+     *
+     * @param array 数组
+     *
+     * @return int
+     */
+    public static long min(final int... array) {
+        if (array == null || array.length == 0) throw new NullPointerException("array is null or empty");
+        int min = array[0];
+        for (int i : array) {
+            if (i < min) i = min;
+        }
+        return min;
+    }
+
+    /**
+     * 获取long数组最小值
+     *
+     * @param array 数组
+     *
+     * @return long
+     */
+    public static long min(final long... array) {
+        if (array == null || array.length == 0) throw new NullPointerException("array is null or empty");
+        long min = array[0];
+        for (long i : array) {
+            if (i < min) i = min;
+        }
+        return min;
+    }
+
+    /**
+     * 将int数组用分隔符拼接成字符串
+     *
+     * @param array     数组
+     * @param delimiter 分隔符
+     *
+     * @return String
+     */
+    public static String joining(final int[] array, final String delimiter) {
+        if (array == null || array.length == 0) return "";
+        StringBuilder sb = new StringBuilder();
+        for (int i : array) {
+            if (sb.length() > 0) sb.append(delimiter);
+            sb.append(i);
+        }
+        return sb.toString();
+    }
+
+    /**
+     * 将long数组用分隔符拼接成字符串
+     *
+     * @param array     数组
+     * @param delimiter 分隔符
+     *
+     * @return String
+     */
+    public static String joining(final long[] array, final String delimiter) {
+        if (array == null || array.length == 0) return "";
+        StringBuilder sb = new StringBuilder();
+        for (long i : array) {
+            if (sb.length() > 0) sb.append(delimiter);
+            sb.append(i);
+        }
+        return sb.toString();
+    }
+
+    /**
+     * 将对象数组用分隔符拼接成字符串
+     *
+     * @param <T>       泛型
+     * @param array     数组
+     * @param delimiter 分隔符
+     *
+     * @return String
+     */
+    public static <T> String joining(final T[] array, final String delimiter) {
+        if (array == null || array.length == 0) return "";
+        StringBuilder sb = new StringBuilder();
+        for (T i : array) {
+            if (sb.length() > 0) sb.append(delimiter);
+            sb.append(i);
+        }
+        return sb.toString();
+    }
+
+    /**
+     * 将一个或多个int新元素添加到int数组结尾
+     *
+     * @param array 原数组
+     * @param objs  待追加数据
+     *
+     * @return 新数组
+     */
+    public static int[] append(final int[] array, final int... objs) {
+        if (array == null || array.length == 0) return objs;
+        if (objs == null || objs.length == 0) return array;
+        final int[] news = new int[array.length + objs.length];
+        System.arraycopy(array, 0, news, 0, array.length);
+        System.arraycopy(objs, 0, news, array.length, objs.length);
+        return news;
+    }
+
+    /**
+     * 将一个或多个long新元素添加到long数组结尾
+     *
+     * @param array 原数组
+     * @param objs  待追加数据
+     *
+     * @return 新数组
+     */
+    public static long[] append(final long[] array, final long... objs) {
+        if (array == null || array.length == 0) return objs;
+        if (objs == null || objs.length == 0) return array;
+        final long[] news = new long[array.length + objs.length];
+        System.arraycopy(array, 0, news, 0, array.length);
+        System.arraycopy(objs, 0, news, array.length, objs.length);
+        return news;
+    }
+
     /**
      * 将一个或多个新元素添加到数组结尾
      *

test/org/redkale/test/asm/AsmCreator.java

@@ -0,0 +1,35 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package org.redkale.test.asm;
+
+import java.io.*;
+import org.redkale.util.Utility;
+
+/**
+ *
+ * @author zhangjx
+ */
+public class AsmCreator {
+
+    public static void main(String[] args) throws Throwable {
+        boolean realasm = true; //从http://forge.ow2.org/projects/asm/ 下载最新asm的src放在 srcasmroot 目录下
+        File srcasmroot = new File("D:/JAVA/JDK源码/JDK9源码/java.base/jdk/internal/org/objectweb/asm");
+        if(realasm)  srcasmroot = new File("D:/JAVA/JDK源码/org/objectweb/asm");
+        File destasmroot = new File("D:/Java-Projects/RedkaleProject/src/org/redkale/asm");
+        String line = null;
+        LineNumberReader txtin = new LineNumberReader(new FileReader(new File(destasmroot, "asm.txt")));
+        while ((line = txtin.readLine()) != null) {
+            line = line.trim();
+            if (!line.endsWith(".java")) continue;
+            File srcfile = new File(srcasmroot, line);
+            File destfile = new File(destasmroot, line);
+            String content = Utility.readThenClose(new FileInputStream(srcfile));
+            FileOutputStream out = new FileOutputStream(destfile);
+            out.write(content.replace("jdk.internal.org.objectweb", "org.redkale").replace("org.objectweb", "org.redkale").getBytes());
+            out.close();
+        }
+    }
+}

test/org/redkale/test/convert/BsonTestMain.java

@@ -13,6 +13,7 @@ import org.redkale.convert.bson.BsonConvert;
 import java.nio.*;
 import java.util.*;
 import org.redkale.convert.json.*;
+import org.redkale.util.*;
 
 /**
  *
@@ -31,6 +32,7 @@ public class BsonTestMain {
         main3(args);
         main4(args);
         main5(args);
+        main6(args);
     }
 
     public static void main2(String[] args) throws Exception {
@@ -98,4 +100,20 @@ public class BsonTestMain {
         Object mapobj = convert.convertFrom(Object.class, bs);
         System.out.println(mapobj);
     }
+
+    public static void main6(String[] args) throws Exception {
+        final BsonConvert convert = BsonFactory.root().getConvert();
+
+        Optional<String> val = Optional.ofNullable("haha");
+        byte[] bs = convert.convertTo(val);
+        Object obj = convert.convertFrom(Optional.class, bs);
+        System.out.println(obj);
+        bs = convert.convertTo(Object.class, val);
+        obj = convert.convertFrom(Object.class, bs);
+        System.out.println(obj);
+        bs = convert.convertTo(new TypeToken<Optional<String>>(){}.getType(), val);
+        obj = convert.convertFrom(new TypeToken<Optional<String>>(){}.getType(), bs);
+        System.out.println(obj);
+        System.out.println(JsonConvert.root().convertTo(val)); 
+    }
 }

test/org/redkale/test/rest/HelloService.java

@@ -60,6 +60,11 @@ public class HelloService implements Service {
         if (source != null) source.update(entity);
     }
 
+    //修改记录
+    public void update2Hello(@RestAddress String clientAddr, @RestUploadFile byte[] fs) { //通过 /pipes/hello/update2?bean={...} 修改对象
+        System.out.println("修改记录2-" + nodeid + ": clientAddr = " + clientAddr + ", fs =" + fs);
+    }
+    
     //修改记录
     @RestMapping(name = "partupdate")
     public void updateHello(HelloEntity entity, @RestParam(name = "cols") String[] columns) { //通过 /pipes/hello/partupdate?bean={...}&cols=... 修改对象
@@ -74,6 +79,7 @@ public class HelloService implements Service {
 
     //查询List列表
     @RestMapping(name = "list")
+    @RestConvert(type = HelloEntity.class, ignoreColumns = {"createtime"})
     public List<HelloEntity> queryHello(HelloBean bean) { //通过 /pipes/hello/list?bean={...} 查询List列表
         return source.queryList(HelloEntity.class, bean);
     }

test/org/redkale/test/rest/SimpleRestServlet.java

@@ -14,11 +14,11 @@ public class SimpleRestServlet extends HttpServlet {
     protected static final RetResult RET_AUTHILLEGAL = RetCodes.retResult(RetCodes.RET_USER_AUTH_ILLEGAL);
 
     @Resource
-    private UserService userService;
+    private UserService userService = new UserService();
 
     @Override
     public void preExecute(HttpRequest request, HttpResponse response) throws IOException {
-        final String sessionid = request.getSessionid(false);
+        final String sessionid = request.getSessionid(true);
         if (sessionid != null) request.setCurrentUser(userService.current(sessionid));
         response.nextEvent();
     }

test/org/redkale/test/rest/UserService.java

@@ -16,6 +16,6 @@ public class UserService implements Service {
 
     //根据登录态获取当前用户信息
     public UserInfo current(String sessionid) {
-        return null;
+        return new UserInfo();
     }
 }

test/org/redkale/test/rest/_DynHelloRestServlet1.java

@@ -38,6 +38,8 @@ public class _DynHelloRestServlet1 extends SimpleRestServlet {
         //headers.put(Rest.REST_HEADER_RESOURCE_NAME, "my-res");
         String url = "http://127.0.0.1:" + port + "/pipes/hello/update?entity={}&bean2={}";
         System.out.println(Utility.postHttpContent(url, headers, null));
+        url = "http://127.0.0.1:" + port + "/pipes/hello/update2?entity={}&bean2={}";
+        System.out.println(Utility.postHttpContent(url, headers, null));
 
         url = "http://127.0.0.1:" + port + "/pipes/hello/asyncfind/1234";
         System.out.println("异步查找: " + Utility.postHttpContent(url, headers, null));