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

ClientCodec优化

Browse Source
This commit is contained in:
redkale
2023-06-30 10:29:15 +08:00
parent f38adacbe4
commit 7c66446459
1 changed files with 24 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/main/java/org/redkale/net/client/ClientCodec.java
View File

@@ -123,11 +123,16 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
connection.preComplete(message, (R) request, exc); connection.preComplete(message, (R) request, exc);
if (exc != null) { if (exc != null) {
if (workThread.inIO() && workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
workThread.execute(() -> { if (workThread.inIO()) {
Traces.currentTraceid(request.traceid); Traces.currentTraceid(request.traceid);
respFuture.completeExceptionally(exc); respFuture.completeExceptionally(exc);
}); } else {
workThread.execute(() -> {
Traces.currentTraceid(request.traceid);
respFuture.completeExceptionally(exc);
});
}
} else { } else {
workThread.runWork(() -> { workThread.runWork(() -> {
Traces.currentTraceid(request.traceid); Traces.currentTraceid(request.traceid);
@@ -136,11 +141,16 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
} }
} else { } else {
final P rs = request.respTransfer == null ? message : (P) request.respTransfer.apply(message); final P rs = request.respTransfer == null ? message : (P) request.respTransfer.apply(message);
if (workThread.inIO() && workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
workThread.execute(() -> { if (workThread.inIO()) {
Traces.currentTraceid(request.traceid); Traces.currentTraceid(request.traceid);
respFuture.complete(rs); respFuture.complete(rs);
}); } else {
workThread.execute(() -> {
Traces.currentTraceid(request.traceid);
respFuture.complete(rs);
});
}
} else { } else {
workThread.runWork(() -> { workThread.runWork(() -> {
Traces.currentTraceid(request.traceid); Traces.currentTraceid(request.traceid);
@@ -149,11 +159,16 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
} }
} }
} catch (Throwable t) { } catch (Throwable t) {
if (workThread.inIO() && workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE if (workThread.getState() == Thread.State.RUNNABLE) { //fullCache时state不是RUNNABLE
workThread.execute(() -> { if (workThread.inIO()) {
Traces.currentTraceid(request.traceid); Traces.currentTraceid(request.traceid);
respFuture.completeExceptionally(t); respFuture.completeExceptionally(t);
}); } else {
workThread.execute(() -> {
Traces.currentTraceid(request.traceid);
respFuture.completeExceptionally(t);
});
}
} else { } else {
workThread.runWork(() -> { workThread.runWork(() -> {
Traces.currentTraceid(request.traceid); Traces.currentTraceid(request.traceid);