src/org/redkale/net/TcpAioAsyncConnection.java

@@ -112,20 +112,28 @@ public class TcpAioAsyncConnection extends AsyncConnection {
 
     private <A> void nextWrite(Throwable exc, A attachment) {
         BlockingQueue<WriteEntry> queue = this.writeQueue;
-        WriteEntry entry = queue == null ? null : queue.poll();
+        if (exc != null && !isOpen()) {
-        if (entry != null) {
+            WriteEntry entry;
-            if (exc == null) {
+            while ((entry = queue.poll()) != null) {
                 try {
-                    if (entry.writeOneBuffer == null) {
+                    entry.writeHandler.failed(exc, entry.writeAttachment);
-                        write(false, entry.writeBuffers, entry.writeOffset, entry.writeLength, entry.writeAttachment, entry.writeHandler);
+                } catch (Throwable e) {
-                    } else {
+                    e.printStackTrace(System.err);
-                        write(false, entry.writeOneBuffer, entry.writeAttachment, entry.writeHandler);
-                    }
-                } catch (Exception e) {
-                    entry.writeHandler.failed(e, entry.writeAttachment);
                 }
-            } else {  //当连接已经关掉了，不需要调用write方法，直接报异常
+            }
-                entry.writeHandler.failed(exc, entry.writeAttachment);
+            return;
+        }
+        WriteEntry entry = queue == null ? null : queue.poll();
+
+        if (entry != null) {
+            try {
+                if (entry.writeOneBuffer == null) {
+                    write(false, entry.writeBuffers, entry.writeOffset, entry.writeLength, entry.writeAttachment, entry.writeHandler);
+                } else {
+                    write(false, entry.writeOneBuffer, entry.writeAttachment, entry.writeHandler);
+                }
+            } catch (Exception e) {
+                entry.writeHandler.failed(e, entry.writeAttachment);
             }
         } else {
             semaphore.release();
@@ -316,18 +324,27 @@ public class TcpAioAsyncConnection extends AsyncConnection {
                     failed(e, attachment);
                     return;
                 }
-                nextWrite(null, attachment);
+                try {
-                writeHandler.completed(writeCount, attachment);
+                    writeHandler.completed(writeCount, attachment);
+                } finally {
+                    nextWrite(null, attachment);
+                }
             } else {
-                nextWrite(null, attachment);
+                try {
-                writeHandler.completed(result.intValue(), attachment);
+                    writeHandler.completed(result.intValue(), attachment);
+                } finally {
+                    nextWrite(null, attachment);
+                }
             }
         }
 
         @Override
         public void failed(Throwable exc, A attachment) {
-            nextWrite(isOpen() ? null : exc, attachment);
+            try {
-            writeHandler.failed(exc, attachment);
+                writeHandler.failed(exc, attachment);
+            } finally {
+                nextWrite(exc, attachment);
+            }
         }
 
     }
@@ -354,14 +371,21 @@ public class TcpAioAsyncConnection extends AsyncConnection {
                 failed(e, attachment);
                 return;
             }
-            nextWrite(null, attachment);
+            try {
-            writeHandler.completed(result, attachment);
+                writeHandler.completed(result, attachment);
+            } finally {
+                nextWrite(null, attachment);
+            }
+
         }
 
         @Override
         public void failed(Throwable exc, A attachment) {
-            nextWrite(isOpen() ? null : exc, attachment);
+            try {
-            writeHandler.failed(exc, attachment);
+                writeHandler.failed(exc, attachment);
+            } finally {
+                nextWrite(exc, attachment);
+            }
         }
 
     }