lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

net浼樺寲璁℃暟鍣?

Browse Source
This commit is contained in:
Redkale
2023-01-27 15:59:09 +08:00
parent 8d4ebe653d
commit a492cbd815
5 changed files with 20 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/main/java/org/redkale/net/AsyncConnection.java
View File

@@ -369,13 +369,13 @@ public abstract class AsyncConnection implements ChannelContext, Channel, AutoCl
CompletionHandler<Integer, Object> newHandler = new CompletionHandler<Integer, Object>() { CompletionHandler<Integer, Object> newHandler = new CompletionHandler<Integer, Object>() {
@Override @Override
public void completed(Integer result, Object attachment) { public void completed(Integer result, Object attachment) {
offerWriteBuffer(buffers); offerWriteBuffers(buffers);
handler.completed(result, attachment); handler.completed(result, attachment);
} }
@Override @Override
public void failed(Throwable exc, Object attachment) { public void failed(Throwable exc, Object attachment) {
offerWriteBuffer(buffers); offerWriteBuffers(buffers);
handler.failed(exc, attachment); handler.failed(exc, attachment);
} }
}; };
@@ -476,13 +476,13 @@ public abstract class AsyncConnection implements ChannelContext, Channel, AutoCl
CompletionHandler<Integer, ? super A> newHandler = new CompletionHandler<Integer, A>() { CompletionHandler<Integer, ? super A> newHandler = new CompletionHandler<Integer, A>() {
@Override @Override
public void completed(Integer result, A attachment) { public void completed(Integer result, A attachment) {
offerWriteBuffer(srcs); offerWriteBuffers(srcs);
handler.completed(result, attachment); handler.completed(result, attachment);
} }
@Override @Override
public void failed(Throwable exc, A attachment) { public void failed(Throwable exc, A attachment) {
offerWriteBuffer(srcs); offerWriteBuffers(srcs);
handler.failed(exc, attachment); handler.failed(exc, attachment);
} }
}; };
@@ -689,7 +689,7 @@ public abstract class AsyncConnection implements ChannelContext, Channel, AutoCl
readBufferConsumer.accept(buffer); readBufferConsumer.accept(buffer);
} }
public void offerReadBuffer(ByteBuffer... buffers) { public void offerReadBuffers(ByteBuffer... buffers) {
if (buffers == null) { if (buffers == null) {
return; return;
} }
@@ -706,7 +706,7 @@ public abstract class AsyncConnection implements ChannelContext, Channel, AutoCl
writeBufferConsumer.accept(buffer); writeBufferConsumer.accept(buffer);
} }
public void offerWriteBuffer(ByteBuffer... buffers) { public void offerWriteBuffers(ByteBuffer... buffers) {
if (buffers == null) { if (buffers == null) {
return; return;
} }
@@ -1009,20 +1009,20 @@ public abstract class AsyncConnection implements ChannelContext, Channel, AutoCl
writeImpl(netBuffers, 0, netBuffers.length, null, new CompletionHandler<Integer, Void>() { writeImpl(netBuffers, 0, netBuffers.length, null, new CompletionHandler<Integer, Void>() {
@Override @Override
public void completed(Integer count, Void attachment) { public void completed(Integer count, Void attachment) {
offerWriteBuffer(netBuffers); offerWriteBuffers(netBuffers);
callback.accept(null); callback.accept(null);
} }
@Override @Override
public void failed(Throwable t, Void attachment) { public void failed(Throwable t, Void attachment) {
offerWriteBuffer(netBuffers); offerWriteBuffers(netBuffers);
callback.accept(t); callback.accept(t);
} }
}); });
} }
return true; return true;
} else { } else {
offerWriteBuffer(netBuffers); offerWriteBuffers(netBuffers);
return false; return false;
} }
} }
@@ -1048,20 +1048,20 @@ public abstract class AsyncConnection implements ChannelContext, Channel, AutoCl
writeImpl(netBuffers, 0, netBuffers.length, null, new CompletionHandler<Integer, Void>() { writeImpl(netBuffers, 0, netBuffers.length, null, new CompletionHandler<Integer, Void>() {
@Override @Override
public void completed(Integer count, Void attachment) { public void completed(Integer count, Void attachment) {
offerWriteBuffer(netBuffers); offerWriteBuffers(netBuffers);
callback.accept(null); callback.accept(null);
} }
@Override @Override
public void failed(Throwable t, Void attachment) { public void failed(Throwable t, Void attachment) {
offerWriteBuffer(netBuffers); offerWriteBuffers(netBuffers);
callback.accept(t); callback.accept(t);
} }
}); });
} }
return true; return true;
} else { } else {
offerWriteBuffer(netBuffers); offerWriteBuffers(netBuffers);
return false; return false;
} }
} }

8
src/main/java/org/redkale/net/AsyncIOThread.java
View File

@@ -10,7 +10,6 @@ import java.nio.ByteBuffer;
import java.nio.channels.*; import java.nio.channels.*;
import java.util.*; import java.util.*;
import java.util.concurrent.*; import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.*; import java.util.function.*;
import java.util.logging.*; import java.util.logging.*;
import org.redkale.util.*; import org.redkale.util.*;
@@ -31,9 +30,6 @@ public class AsyncIOThread extends WorkThread {
final Selector selector; final Selector selector;
//如果有read/write两IOThread，只记readThread
final LongAdder connCounter = new LongAdder();
private final Supplier<ByteBuffer> bufferSupplier; private final Supplier<ByteBuffer> bufferSupplier;
private final Consumer<ByteBuffer> bufferConsumer; private final Consumer<ByteBuffer> bufferConsumer;
@@ -141,10 +137,6 @@ public class AsyncIOThread extends WorkThread {
return bufferConsumer; return bufferConsumer;
} }
public int currConnections() {
return connCounter.intValue();
}
@Override @Override
public void run() { public void run() {
final Queue<Runnable> commands = this.commandQueue; final Queue<Runnable> commands = this.commandQueue;

12
src/main/java/org/redkale/net/AsyncNioCompletionHandler.java
View File

@@ -74,9 +74,9 @@ class AsyncNioCompletionHandler<A> implements CompletionHandler<Integer, A>, Run
if (conn != null) { if (conn != null) {
if (buffers != null) { if (buffers != null) {
if (readMode) { if (readMode) {
conn.offerReadBuffer(buffers); conn.offerReadBuffers(buffers);
} else { } else {
conn.offerWriteBuffer(buffers); conn.offerWriteBuffers(buffers);
} }
} else if (buffer != null) { } else if (buffer != null) {
if (readMode) { if (readMode) {
@@ -102,9 +102,9 @@ class AsyncNioCompletionHandler<A> implements CompletionHandler<Integer, A>, Run
if (conn != null) { if (conn != null) {
if (buffers != null) { if (buffers != null) {
if (readMode) { if (readMode) {
conn.offerReadBuffer(buffers); conn.offerReadBuffers(buffers);
} else { } else {
conn.offerWriteBuffer(buffers); conn.offerWriteBuffers(buffers);
} }
} else if (buffer != null) { } else if (buffer != null) {
if (readMode) { if (readMode) {
@@ -125,9 +125,9 @@ class AsyncNioCompletionHandler<A> implements CompletionHandler<Integer, A>, Run
if (conn != null) { if (conn != null) {
if (buffers != null) { if (buffers != null) {
if (readMode) { if (readMode) {
conn.offerReadBuffer(buffers); conn.offerReadBuffers(buffers);
} else { } else {
conn.offerWriteBuffer(buffers); conn.offerWriteBuffers(buffers);
} }
} else if (buffer != null) { } else if (buffer != null) {
if (readMode) { if (readMode) {

4
src/main/java/org/redkale/net/AsyncNioTcpConnection.java
View File

@@ -39,7 +39,6 @@ class AsyncNioTcpConnection extends AsyncNioConnection {
} }
} }
this.remoteAddress = addr; this.remoteAddress = addr;
ioReadThread.connCounter.increment();
} }
@Override @Override
@@ -183,7 +182,7 @@ class AsyncNioTcpConnection extends AsyncNioConnection {
channel.write(netBuffers); channel.write(netBuffers);
} }
} }
offerWriteBuffer(netBuffers); offerWriteBuffers(netBuffers);
return len; return len;
} }
@@ -269,7 +268,6 @@ class AsyncNioTcpConnection extends AsyncNioConnection {
@Override @Override
public final void close() throws IOException { public final void close() throws IOException {
super.close(); super.close();
ioReadThread.connCounter.decrement();
channel.close(); channel.close();
if (this.connectKey != null) { if (this.connectKey != null) {
this.connectKey.cancel(); this.connectKey.cancel();

2
src/main/java/org/redkale/net/Response.java
View File

@@ -437,7 +437,7 @@ public abstract class Response<C extends Context, R extends Request<C>> {
@Override @Override
public void completed(Integer result, A attachment) { public void completed(Integer result, A attachment) {
channel.offerWriteBuffer(buffers); channel.offerWriteBuffers(buffers);
if (handler != null) { if (handler != null) {
handler.completed(result, attachment); handler.completed(result, attachment);
} }