Response优化

src/main/java/org/redkale/net/AsyncConnection.java

@@ -256,6 +256,39 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
     protected abstract <A> void writeImpl(
             ByteBuffer[] srcs, int offset, int length, A attachment, CompletionHandler<Integer, ? super A> handler);
 
+    /**
+     * src写完才会回调
+     *
+     * @see org.redkale.net.AsyncNioConnection#writeImpl(java.nio.ByteBuffer, java.util.function.Consumer, java.lang.Object, java.nio.channels.CompletionHandler)
+     * @param <A> A
+     * @param src ByteBuffer
+     * @param consumer Consumer
+     * @param attachment A
+     * @param handler CompletionHandler
+     */
+    protected abstract <A> void writeImpl(
+            ByteBuffer src, Consumer<ByteBuffer> consumer, A attachment, CompletionHandler<Integer, ? super A> handler);
+
+    /**
+     *  srcs写完才会回调
+     *
+     * @see org.redkale.net.AsyncNioConnection#writeImpl(java.nio.ByteBuffer[], int, int, java.util.function.Consumer, java.lang.Object, java.nio.channels.CompletionHandler)
+     * @param <A> A
+     * @param srcs ByteBuffer[]
+     * @param offset offset
+     * @param length length
+     * @param consumer Consumer
+     * @param attachment A
+     * @param handler  CompletionHandler
+     */
+    protected abstract <A> void writeImpl(
+            ByteBuffer[] srcs,
+            int offset,
+            int length,
+            Consumer<ByteBuffer> consumer,
+            A attachment,
+            CompletionHandler<Integer, ? super A> handler);
+
     protected void startRead(CompletionHandler<Integer, ByteBuffer> handler) {
         read(handler);
     }
@@ -301,7 +334,7 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
      * @param handler  回调函数
      */
     public final void write(ByteTuple array, CompletionHandler<Integer, Void> handler) {
-        write(array.content(), array.offset(), array.length(), (byte[]) null, 0, 0, handler);
+        write(array.content(), array.offset(), array.length(), null, handler);
     }
 
     /**
@@ -318,7 +351,7 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
 
     <A> void write(ByteBuffer src, A attachment, CompletionHandler<Integer, ? super A> handler) {
         if (sslEngine == null) {
-            writeImpl(src, attachment, handler);
+            writeImpl(src, (Consumer) null, attachment, handler);
         } else {
             try {
                 int remain = src.remaining();
@@ -338,7 +371,7 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
     <A> void write(
             ByteBuffer[] srcs, int offset, int length, A attachment, CompletionHandler<Integer, ? super A> handler) {
         if (sslEngine == null) {
-            writeImpl(srcs, offset, length, attachment, handler);
+            writeImpl(srcs, offset, length, (Consumer) null, attachment, handler);
         } else {
             try {
                 int remain = ByteBufferReader.remaining(srcs, offset, length);
@@ -355,44 +388,10 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
         }
     }
 
-    <A> void write(ByteBuffer[] srcs, A attachment, CompletionHandler<Integer, ? super A> handler) {
-        write(srcs, 0, srcs.length, attachment, handler);
-    }
-
-    void write(byte[] bytes, CompletionHandler<Integer, Void> handler) {
-        write(bytes, 0, bytes.length, (byte[]) null, 0, 0, handler);
-    }
-
-    void write(byte[] bytes, int offset, int length, CompletionHandler<Integer, Void> handler) {
-        write(bytes, offset, length, (byte[]) null, 0, 0, handler);
-    }
-
-    void write(
-            byte[] headerContent,
-            int headerOffset,
-            int headerLength,
-            byte[] bodyContent,
-            int bodyOffset,
-            int bodyLength,
-            CompletionHandler<Integer, Void> handler) {
-        write(headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength, null, handler);
-    }
-
-    void write(
-            byte[] headerContent,
-            int headerOffset,
-            int headerLength,
-            byte[] bodyContent,
-            int bodyOffset,
-            int bodyLength,
-            Object handlerAttachment,
-            CompletionHandler handler) {
+    void write(byte[] bytes, int offset, int length, Object attachment, CompletionHandler handler) {
         final ByteBuffer buffer = sslEngine == null ? pollWriteBuffer() : pollWriteSSLBuffer();
-        if (buffer.remaining() >= headerLength + bodyLength) {
-            buffer.put(headerContent, headerOffset, headerLength);
-            if (bodyLength > 0) {
-                buffer.put(bodyContent, bodyOffset, bodyLength);
-            }
+        if (buffer.remaining() >= length) {
+            buffer.put(bytes, offset, length);
             buffer.flip();
             CompletionHandler<Integer, Object> newHandler = new CompletionHandler<Integer, Object>() {
                 @Override
@@ -407,14 +406,11 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
                     handler.failed(exc, attachment);
                 }
             };
-            write(buffer, handlerAttachment, newHandler);
+            write(buffer, attachment, newHandler);
         } else {
             ByteBufferWriter writer =
                     ByteBufferWriter.create(sslEngine == null ? writeBufferSupplier : this::pollWriteSSLBuffer, buffer);
-            writer.put(headerContent, headerOffset, headerLength);
-            if (bodyLength > 0) {
-                writer.put(bodyContent, bodyOffset, bodyLength);
-            }
+            writer.put(bytes, offset, length);
             final ByteBuffer[] buffers = writer.toBuffers();
             CompletionHandler<Integer, Object> newHandler = new CompletionHandler<Integer, Object>() {
                 @Override
@@ -429,11 +425,19 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
                     handler.failed(exc, attachment);
                 }
             };
-            write(buffers, handlerAttachment, newHandler);
+            write(buffers, 0, buffers.length, attachment, newHandler);
         }
     }
 
-    // src写完才会回调
+    /**
+     * src写完才会回调
+     *
+     * @see Response#finish(boolean, byte[], int, int)
+     * @param <A> 泛型
+     * @param src ByteBuffer
+     * @param attachment 附件
+     * @param handler 回调函数
+     */
     public final <A> void writeInIOThread(ByteBuffer src, A attachment, CompletionHandler<Integer, ? super A> handler) {
         if (inCurrWriteThread()) {
             write(src, attachment, handler);
@@ -441,6 +445,22 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
             executeWrite(() -> write(src, attachment, handler));
         }
     }
+    /**
+     * src写完才会回调
+     *
+     * @see Response#finish(boolean, byte[], int, int)
+     * @param bytes 内容
+     * @param offset 培偏移量
+     * @param length 长度
+     * @param handler 回调函数
+     */
+    public final void writeInIOThread(byte[] bytes, int offset, int length, CompletionHandler<Integer, Void> handler) {
+        if (inCurrWriteThread()) {
+            write(bytes, offset, length, null, handler);
+        } else {
+            executeWrite(() -> write(bytes, offset, length, null, handler));
+        }
+    }
 
     public final <A> void writeInIOThread(
             ByteBuffer[] srcs, int offset, int length, A attachment, CompletionHandler<Integer, ? super A> handler) {
@@ -454,17 +474,17 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
     public final <A> void writeInIOThread(
             ByteBuffer[] srcs, A attachment, CompletionHandler<Integer, ? super A> handler) {
         if (inCurrWriteThread()) {
-            write(srcs, attachment, handler);
+            write(srcs, 0, srcs.length, attachment, handler);
         } else {
-            executeWrite(() -> write(srcs, attachment, handler));
+            executeWrite(() -> write(srcs, 0, srcs.length, attachment, handler));
         }
     }
 
     public final void writeInIOThread(byte[] bytes, CompletionHandler<Integer, Void> handler) {
         if (inCurrWriteThread()) {
-            write(bytes, handler);
+            write(bytes, 0, bytes.length, null, handler);
         } else {
-            executeWrite(() -> write(bytes, handler));
+            executeWrite(() -> write(bytes, 0, bytes.length, null, handler));
         }
     }
 
@@ -476,30 +496,6 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
         }
     }
 
-    public final void writeInIOThread(byte[] bytes, int offset, int length, CompletionHandler<Integer, Void> handler) {
-        if (inCurrWriteThread()) {
-            write(bytes, offset, length, handler);
-        } else {
-            executeWrite(() -> write(bytes, offset, length, handler));
-        }
-    }
-
-    public final void writeInIOThread(
-            byte[] headerContent,
-            int headerOffset,
-            int headerLength,
-            byte[] bodyContent,
-            int bodyOffset,
-            int bodyLength,
-            CompletionHandler<Integer, Void> handler) {
-        if (inCurrWriteThread()) {
-            write(headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength, handler);
-        } else {
-            executeWrite(() ->
-                    write(headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength, handler));
-        }
-    }
-
     public void setReadBuffer(ByteBuffer buffer) {
         if (this.readBuffer != null) {
             throw new RedkaleException("repeat AsyncConnection.setReadBuffer");
@@ -539,7 +535,7 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
             if (srcs.length == 1) {
                 write(srcs[0], attachment, newHandler);
             } else {
-                write(srcs, attachment, newHandler);
+                write(srcs, 0, srcs.length, attachment, newHandler);
             }
         }
     }
@@ -566,7 +562,7 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
     }
 
     // 返回pipelineCount个数数据是否全部写入完毕
-    public boolean appendPipeline(int pipelineIndex, int pipelineCount, byte[] bs, int offset, int length) {
+    public boolean appendPipeline(int pipelineIndex, int pipelineCount, byte[] bytes, int offset, int length) {
         writeLock.lock();
         try {
             ByteBufferWriter writer = this.pipelineWriter;
@@ -575,7 +571,7 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
                 this.pipelineWriter = writer;
             }
             if (this.pipelineDataNode == null && pipelineIndex == writer.getWriteBytesCounter() + 1) {
-                writer.put(bs, offset, length);
+                writer.put(bytes, offset, length);
                 return (pipelineIndex == pipelineCount);
             } else {
                 PipelineDataNode dataNode = this.pipelineDataNode;
@@ -586,65 +582,7 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
                 if (pipelineIndex == pipelineCount) { // 此时pipelineCount为最大值
                     dataNode.pipelineCount = pipelineCount;
                 }
-                dataNode.put(pipelineIndex, bs, offset, length);
-                if (writer.getWriteBytesCounter() + dataNode.itemsize == dataNode.pipelineCount) {
-                    for (PipelineDataItem item : dataNode.arrayItems()) {
-                        writer.put(item.data);
-                    }
-                    this.pipelineDataNode = null;
-                    return true;
-                }
-                return false;
-            }
-        } finally {
-            writeLock.unlock();
-        }
-    }
-
-    // 返回pipelineCount个数数据是否全部写入完毕
-    public final boolean appendPipeline(int pipelineIndex, int pipelineCount, ByteTuple header, ByteTuple body) {
-        return appendPipeline(
-                pipelineIndex,
-                pipelineCount,
-                header.content(),
-                header.offset(),
-                header.length(),
-                body == null ? null : body.content(),
-                body == null ? 0 : body.offset(),
-                body == null ? 0 : body.length());
-    }
-
-    // 返回pipelineCount个数数据是否全部写入完毕
-    public boolean appendPipeline(
-            int pipelineIndex,
-            int pipelineCount,
-            byte[] headerContent,
-            int headerOffset,
-            int headerLength,
-            byte[] bodyContent,
-            int bodyOffset,
-            int bodyLength) {
-        writeLock.lock();
-        try {
-            ByteBufferWriter writer = this.pipelineWriter;
-            if (writer == null) {
-                writer = ByteBufferWriter.create(getWriteBufferSupplier());
-                this.pipelineWriter = writer;
-            }
-            if (this.pipelineDataNode == null && pipelineIndex == writer.getWriteBytesCounter() + 1) {
-                writer.put(headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength);
-                return (pipelineIndex == pipelineCount);
-            } else {
-                PipelineDataNode dataNode = this.pipelineDataNode;
-                if (dataNode == null) {
-                    dataNode = new PipelineDataNode();
-                    this.pipelineDataNode = dataNode;
-                }
-                if (pipelineIndex == pipelineCount) { // 此时pipelineCount为最大值
-                    dataNode.pipelineCount = pipelineCount;
-                }
-                dataNode.put(
-                        pipelineIndex, headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength);
+                dataNode.put(pipelineIndex, bytes, offset, length);
                 if (writer.getWriteBytesCounter() + dataNode.itemsize == dataNode.pipelineCount) {
                     for (PipelineDataItem item : dataNode.arrayItems()) {
                         writer.put(item.data);
@@ -694,27 +632,6 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
             }
             itemsize++;
         }
-
-        public void put(
-                int pipelineIndex,
-                byte[] headerContent,
-                int headerOffset,
-                int headerLength,
-                byte[] bodyContent,
-                int bodyOffset,
-                int bodyLength) {
-            if (tail == null) {
-                head = new PipelineDataItem(
-                        pipelineIndex, headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength);
-                tail = head;
-            } else {
-                PipelineDataItem item = new PipelineDataItem(
-                        pipelineIndex, headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength);
-                tail.next = item;
-                tail = item;
-            }
-            itemsize++;
-        }
     }
 
     private static class PipelineDataItem implements Comparable<PipelineDataItem> {
@@ -730,33 +647,6 @@ public abstract class AsyncConnection implements Channel, AutoCloseable {
             this.data = Arrays.copyOfRange(bs, offset, offset + length);
         }
 
-        public PipelineDataItem(
-                int index,
-                byte[] headerContent,
-                int headerOffset,
-                int headerLength,
-                byte[] bodyContent,
-                int bodyOffset,
-                int bodyLength) {
-            this.index = index;
-            this.data = bodyLength > 0
-                    ? copyOfRange(headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength)
-                    : Arrays.copyOfRange(headerContent, headerOffset, headerOffset + headerLength);
-        }
-
-        private static byte[] copyOfRange(
-                byte[] headerContent,
-                int headerOffset,
-                int headerLength,
-                byte[] bodyContent,
-                int bodyOffset,
-                int bodyLength) {
-            byte[] result = new byte[headerLength + bodyLength];
-            System.arraycopy(headerContent, headerOffset, result, 0, headerLength);
-            System.arraycopy(bodyContent, bodyOffset, result, headerLength, bodyLength);
-            return result;
-        }
-
         @Override
         public int compareTo(PipelineDataItem o) {
             return this.index - o.index;

src/main/java/org/redkale/net/AsyncNioConnection.java

@@ -152,61 +152,160 @@ abstract class AsyncNioConnection extends AsyncConnection {
         doRead(this.ioReadThread.inCurrThread());
     }
 
+    public void doRead(boolean direct) {
+        try {
+            this.readTime = System.currentTimeMillis();
+            int readCount = 0;
+            if (direct) {
+                if (this.readByteBuffer == null) {
+                    this.readByteBuffer = sslEngine == null ? pollReadBuffer() : pollReadSSLBuffer();
+                }
+                readCount = implRead(readByteBuffer);
+            }
+
+            if (readCount != 0) {
+                handleRead(readCount, null);
+            } else if (readKey == null) {
+                ioReadThread.register(selector -> {
+                    try {
+                        if (readKey == null) {
+                            readKey = keyFor(selector);
+                        }
+                        if (readKey == null) {
+                            readKey = implRegister(selector, SelectionKey.OP_READ);
+                            readKey.attach(this);
+                        } else {
+                            readKey.interestOps(readKey.interestOps() | SelectionKey.OP_READ);
+                        }
+                    } catch (ClosedChannelException e) {
+                        handleRead(0, e);
+                    }
+                });
+            } else {
+                ioReadThread.interestOpsOr(readKey, SelectionKey.OP_READ);
+            }
+        } catch (Exception e) {
+            handleRead(0, e);
+        }
+    }
+
+    protected void handleRead(final int totalCount, Throwable t) {
+        CompletionHandler<Integer, ByteBuffer> handler = this.readCompletionHandler;
+        ByteBuffer attach = this.readByteBuffer;
+        // 清空读参数
+        this.readCompletionHandler = null;
+        this.readByteBuffer = null;
+        this.readPending = false; // 必须放最后
+
+        if (handler == null) {
+            if (t == null) {
+                protocolCodec.completed(totalCount, attach);
+            } else {
+                protocolCodec.failed(t, attach);
+            }
+        } else {
+            if (t == null) {
+                handler.completed(totalCount, attach);
+            } else {
+                handler.failed(t, attach);
+            }
+        }
+    }
+
     @Override
-    public void write(
-            byte[] headerContent,
-            int headerOffset,
-            int headerLength,
-            byte[] bodyContent,
-            int bodyOffset,
-            int bodyLength,
-            CompletionHandler<Integer, Void> handler) {
-
-        if (sslEngine != null) {
-            super.write(headerContent, headerOffset, headerLength, bodyContent, bodyOffset, bodyLength, handler);
-            return;
-        }
-        Objects.requireNonNull(headerContent);
+    public <A> void writeImpl(
+            ByteBuffer src,
+            Consumer<ByteBuffer> consumer,
+            A attachment,
+            CompletionHandler<Integer, ? super A> handler) {
+        Objects.requireNonNull(src);
         Objects.requireNonNull(handler);
-        if (!this.isConnected()) {
-            handler.failed(new NotYetConnectedException(), null);
-            return;
-        }
-        if (this.writePending) {
-            handler.failed(new WritePendingException(), null);
-            return;
-        }
-        this.writePending = true;
-        this.writeByteTuple1Array = headerContent;
-        this.writeByteTuple1Offset = headerOffset;
-        this.writeByteTuple1Length = headerLength;
-        this.writeByteTuple2Array = bodyContent;
-        this.writeByteTuple2Offset = bodyOffset;
-        this.writeByteTuple2Length = bodyLength;
-        this.writeAttachment = null;
-        CompletionHandler<Integer, Void> newHandler = new CompletionHandler<Integer, Void>() {
-            @Override
-            public void completed(Integer result, Void attachment) {
-                if (writeByteBuffers != null) {
-                    offerWriteBuffers(writeByteBuffers);
-                } else {
-                    offerWriteBuffer(writeByteBuffer);
-                }
-                handler.completed(result, attachment);
+        int total = 0;
+        Exception t = null;
+        try {
+            if (this.writePending) {
+                handler.failed(new WritePendingException(), attachment);
+                return;
             }
+            this.writePending = true;
+            while (src.hasRemaining()) { // 必须要将buffer写完为止
+                int c = implWrite(src);
+                if (c < 0) {
+                    t = new ClosedChannelException();
+                    total = c;
+                    break;
+                }
+                total += c;
+            }
+        } catch (Exception e) {
+            t = e;
+        }
+        this.writePending = false;
+        if (consumer != null) {
+            consumer.accept(src);
+        }
+        if (t != null) {
+            handler.failed(t, attachment);
+        } else {
+            handler.completed(total, attachment);
+        }
+    }
 
-            @Override
-            public void failed(Throwable exc, Void attachment) {
-                if (writeByteBuffers != null) {
-                    offerWriteBuffers(writeByteBuffers);
-                } else {
-                    offerWriteBuffer(writeByteBuffer);
-                }
-                handler.failed(exc, attachment);
+    @Override
+    public <A> void writeImpl(
+            ByteBuffer[] srcs,
+            int offset,
+            int length,
+            Consumer<ByteBuffer> consumer,
+            A attachment,
+            CompletionHandler<Integer, ? super A> handler) {
+        Objects.requireNonNull(srcs);
+        Objects.requireNonNull(handler);
+        int total = 0;
+        Exception t = null;
+        int batchOffset = offset;
+        int batchLength = length;
+        ByteBuffer[] batchBuffers = srcs;
+        try {
+            if (this.writePending) {
+                handler.failed(new WritePendingException(), attachment);
+                return;
             }
-        };
-        this.writeCompletionHandler = newHandler;
-        doWrite(); // 如果不是true，则bodyCallback的执行可能会切换线程
+            this.writePending = true;
+            boolean hasRemain = true;
+            while (hasRemain) { // 必须要将buffer写完为止
+                int c = implWrite(batchBuffers, batchOffset, batchLength);
+                if (c < 0) {
+                    t = new ClosedChannelException();
+                    total = c;
+                    break;
+                }
+                boolean remain = false;
+                for (int i = 0; i < batchLength; i++) {
+                    if (batchBuffers[batchOffset + i].hasRemaining()) {
+                        remain = true;
+                        batchOffset += i;
+                        batchLength -= i;
+                        break;
+                    }
+                }
+                hasRemain = remain;
+                total += c;
+            }
+        } catch (Exception e) {
+            t = e;
+        }
+        this.writePending = false;
+        if (consumer != null) {
+            for (int i = 0; i < length; i++) {
+                consumer.accept(srcs[offset + i]);
+            }
+        }
+        if (t != null) {
+            handler.failed(t, attachment);
+        } else {
+            handler.completed(total, attachment);
+        }
     }
 
     @Override
@@ -250,43 +349,6 @@ abstract class AsyncNioConnection extends AsyncConnection {
         doWrite();
     }
 
-    public void doRead(boolean direct) {
-        try {
-            this.readTime = System.currentTimeMillis();
-            int readCount = 0;
-            if (direct) {
-                if (this.readByteBuffer == null) {
-                    this.readByteBuffer = sslEngine == null ? pollReadBuffer() : pollReadSSLBuffer();
-                }
-                readCount = implRead(readByteBuffer);
-            }
-
-            if (readCount != 0) {
-                handleRead(readCount, null);
-            } else if (readKey == null) {
-                ioReadThread.register(selector -> {
-                    try {
-                        if (readKey == null) {
-                            readKey = keyFor(selector);
-                        }
-                        if (readKey == null) {
-                            readKey = implRegister(selector, SelectionKey.OP_READ);
-                            readKey.attach(this);
-                        } else {
-                            readKey.interestOps(readKey.interestOps() | SelectionKey.OP_READ);
-                        }
-                    } catch (ClosedChannelException e) {
-                        handleRead(0, e);
-                    }
-                });
-            } else {
-                ioReadThread.interestOpsOr(readKey, SelectionKey.OP_READ);
-            }
-        } catch (Exception e) {
-            handleRead(0, e);
-        }
-    }
-
     public void doWrite() {
         try {
             this.writeTime = System.currentTimeMillis();
@@ -372,9 +434,6 @@ abstract class AsyncNioConnection extends AsyncConnection {
                 handleWrite(totalCount, new ClosedChannelException());
             } else if (writeCompleted && (totalCount != 0 || !hasRemain)) {
                 handleWrite(this.writeTotal + totalCount, null);
-                //                if (fastWriteCount.get() > 0) {
-                //                    doWrite();
-                //                }
             } else if (writeKey == null) {
                 ioWriteThread.register(selector -> {
                     try {
@@ -399,6 +458,26 @@ abstract class AsyncNioConnection extends AsyncConnection {
         }
     }
 
+    protected void handleWrite(final int totalCount, Throwable t) {
+        CompletionHandler handler = this.writeCompletionHandler;
+        Object attach = this.writeAttachment;
+        // 清空写参数
+        this.writeCompletionHandler = null;
+        this.writeAttachment = null;
+        this.writeByteBuffer = null;
+        this.writeByteBuffers = null;
+        this.writeBuffersOffset = 0;
+        this.writeBuffersLength = 0;
+        this.writeTotal = 0;
+        this.writePending = false; // 必须放最后
+
+        if (t == null) {
+            handler.completed(totalCount, attach);
+        } else {
+            handler.failed(t, attach);
+        }
+    }
+
     protected void handleConnect(Throwable t) {
         if (connectKey != null) {
             connectKey.cancel();
@@ -420,49 +499,6 @@ abstract class AsyncNioConnection extends AsyncConnection {
         }
     }
 
-    protected void handleRead(final int totalCount, Throwable t) {
-        CompletionHandler<Integer, ByteBuffer> handler = this.readCompletionHandler;
-        ByteBuffer attach = this.readByteBuffer;
-        // 清空读参数
-        this.readCompletionHandler = null;
-        this.readByteBuffer = null;
-        this.readPending = false; // 必须放最后
-
-        if (handler == null) {
-            if (t == null) {
-                protocolCodec.completed(totalCount, attach);
-            } else {
-                protocolCodec.failed(t, attach);
-            }
-        } else {
-            if (t == null) {
-                handler.completed(totalCount, attach);
-            } else {
-                handler.failed(t, attach);
-            }
-        }
-    }
-
-    protected void handleWrite(final int totalCount, Throwable t) {
-        CompletionHandler handler = this.writeCompletionHandler;
-        Object attach = this.writeAttachment;
-        // 清空写参数
-        this.writeCompletionHandler = null;
-        this.writeAttachment = null;
-        this.writeByteBuffer = null;
-        this.writeByteBuffers = null;
-        this.writeBuffersOffset = 0;
-        this.writeBuffersLength = 0;
-        this.writeTotal = 0;
-        this.writePending = false; // 必须放最后
-
-        if (t == null) {
-            handler.completed(totalCount, attach);
-        } else {
-            handler.failed(t, attach);
-        }
-    }
-
     @Deprecated(since = "2.5.0")
     protected abstract ReadableByteChannel readableByteChannel();
 

src/main/java/org/redkale/net/Response.java

@@ -345,27 +345,27 @@ public abstract class Response<C extends Context, R extends Request<C>> {
         }
     }
 
-    public final void finish(final byte[] bs) {
+    public final void finish(byte[] bs) {
         finish(false, bs, 0, bs.length);
     }
 
-    public final void finish(final byte[] bs, int offset, int length) {
+    public final void finish(byte[] bs, int offset, int length) {
         finish(false, bs, offset, length);
     }
 
-    public final void finish(final ByteTuple array) {
+    public final void finish(ByteTuple array) {
         finish(false, array.content(), array.offset(), array.length());
     }
 
-    public final void finish(boolean kill, final byte[] bs) {
+    public final void finish(boolean kill, byte[] bs) {
         finish(kill, bs, 0, bs.length);
     }
 
-    public final void finish(boolean kill, final ByteTuple array) {
+    public final void finish(boolean kill, ByteTuple array) {
         finish(kill, array.content(), array.offset(), array.length());
     }
 
-    public void finish(boolean kill, final byte[] bs, int offset, int length) {
+    public void finish(boolean kill, byte[] bs, int offset, int length) {
         if (kill) {
             refuseAlive();
         }
@@ -395,104 +395,8 @@ public abstract class Response<C extends Context, R extends Request<C>> {
         }
     }
 
-    public void finish(boolean kill, byte[] bs1, int offset1, int length1, byte[] bs2, int offset2, int length2) {
-        if (kill) {
-            refuseAlive();
-        }
-        if (request.pipelineIndex > 0) {
-            boolean allCompleted = this.channel.appendPipeline(
-                    request.pipelineIndex, request.pipelineCount, bs1, offset1, length1, bs2, offset2, length2);
-            if (allCompleted) {
-                request.pipelineCompleted = true;
-                this.channel.writePipelineInIOThread(this.finishBytesIOThreadHandler);
-            } else {
-                removeChannel();
-                this.responseConsumer.accept(this);
-            }
-        } else if (this.channel.hasPipelineData()) {
-            this.channel.appendPipeline(
-                    request.pipelineIndex, request.pipelineCount, bs1, offset1, length1, bs2, offset2, length2);
-            this.channel.writePipelineInIOThread(this.finishBytesIOThreadHandler);
-        } else {
-            this.channel.writeInIOThread(bs1, offset1, length1, bs2, offset2, length2, finishBytesIOThreadHandler);
-        }
-    }
-
-    protected void send(final ByteTuple array, final CompletionHandler<Integer, Void> handler) {
-        ByteBuffer buffer = this.writeBuffer;
-        if (buffer != null && buffer.capacity() >= array.length()) {
-            buffer.clear();
-            buffer.put(array.content(), array.offset(), array.length());
-            buffer.flip();
-            this.channel.write(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
-
-                @Override
-                public void completed(Integer result, ByteBuffer attachment) {
-                    attachment.clear();
-                    handler.completed(result, null);
-                }
-
-                @Override
-                public void failed(Throwable exc, ByteBuffer attachment) {
-                    attachment.clear();
-                    handler.failed(exc, null);
-                }
-            });
-        } else {
-            this.channel.write(array, handler);
-        }
-    }
-
-    protected <A> void send(ByteBuffer buffer, A attachment, CompletionHandler<Integer, A> handler) {
-        this.channel.write(buffer, attachment, new CompletionHandler<Integer, A>() {
-
-            @Override
-            public void completed(Integer result, A attachment) {
-                if (buffer != writeBuffer) {
-                    channel.offerWriteBuffer(buffer);
-                } else {
-                    buffer.clear();
-                }
-                if (handler != null) {
-                    handler.completed(result, attachment);
-                }
-            }
-
-            @Override
-            public void failed(Throwable exc, A attachment) {
-                if (buffer != writeBuffer) {
-                    channel.offerWriteBuffer(buffer);
-                } else {
-                    buffer.clear();
-                }
-                if (handler != null) {
-                    handler.failed(exc, attachment);
-                }
-            }
-        });
-    }
-
-    protected <A> void send(ByteBuffer[] buffers, A attachment, CompletionHandler<Integer, A> handler) {
-        this.channel.write(buffers, attachment, new CompletionHandler<Integer, A>() {
-
-            @Override
-            public void completed(Integer result, A attachment) {
-                channel.offerWriteBuffers(buffers);
-                if (handler != null) {
-                    handler.completed(result, attachment);
-                }
-            }
-
-            @Override
-            public void failed(Throwable exc, A attachment) {
-                for (ByteBuffer buffer : buffers) {
-                    channel.offerWriteBuffer(buffer);
-                }
-                if (handler != null) {
-                    handler.failed(exc, attachment);
-                }
-            }
-        });
+    protected void send(ByteTuple array, CompletionHandler<Integer, Void> handler) {
+        this.channel.writeInIOThread(array, handler);
     }
 
     public C getContext() {

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

@@ -1241,28 +1241,14 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
     }
 
     /**
-     * 异步输出指定内容, 供WebSocketServlet使用
+     * 异步输出header
      *
-     * @param buffer 输出内容
      * @param handler 异步回调函数
      */
-    protected void sendBody(ByteBuffer buffer, CompletionHandler<Integer, Void> handler) {
-        if (this.headWritedSize < 0) {
-            if (this.contentLength < 0) {
-                this.contentLength = buffer == null ? 0 : buffer.remaining();
-            }
-            createHeader();
-            if (buffer == null) { // 只发header
-                super.send(headerArray, handler);
-            } else {
-                ByteBuffer headBuf = channel.pollWriteBuffer();
-                headBuf.put(headerArray.content(), 0, headerArray.length());
-                headBuf.flip();
-                super.send(new ByteBuffer[] {headBuf, buffer}, null, handler);
-            }
-        } else {
-            super.send(buffer, null, handler);
-        }
+    protected void sendHeader(CompletionHandler<Integer, Void> handler) {
+        this.contentLength = 0;
+        createHeader();
+        super.send(headerArray, handler);
     }
 
     /**

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

@@ -9,7 +9,6 @@ import java.io.*;
 import java.lang.reflect.*;
 import java.math.BigInteger;
 import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
 import java.nio.channels.CompletionHandler;
 import java.security.MessageDigest;
 import java.util.*;
@@ -322,7 +321,7 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl
                 response.addHeader("Sec-WebSocket-Extensions", "permessage-deflate");
             }
 
-            response.sendBody((ByteBuffer) null, new CompletionHandler<Integer, Void>() {
+            response.sendHeader(new CompletionHandler<Integer, Void>() {
 
                 @Override
                 public void completed(Integer result, Void attachment) {