优化ProtocolCodec的续读功能

This commit is contained in:
redkale
2023-11-29 22:59:13 +08:00
parent ba788eec9c
commit a4b277e875
6 changed files with 21 additions and 2 deletions

View File

@@ -159,6 +159,8 @@ class ProtocolCodec implements CompletionHandler<Integer, ByteBuffer> {
context.dispatcher.incrExecuteCounter();
int pindex = pipelineIndex;
boolean pipeline = false;
boolean seted = false;
boolean completed = request.completed;
Request hreq = lastReq;
if (buffer.hasRemaining()) {
pipeline = true;
@@ -176,6 +178,7 @@ class ProtocolCodec implements CompletionHandler<Integer, ByteBuffer> {
request.pipeline(pindex, pindex);
}
channel.setReadBuffer(buffer.clear());
seted = true;
}
context.executeDispatch(request, response);
if (pipeline) {
@@ -186,6 +189,11 @@ class ProtocolCodec implements CompletionHandler<Integer, ByteBuffer> {
context.logger.log(Level.WARNING, "dispatch pipeline servlet abort, force to close channel ", t);
pipelineResponse.codecError(t);
}
} else if (completed) {
if (!seted) {
channel.setReadBuffer(buffer.clear());
}
channel.readRegister(this);
}
} else {
channel.setReadBuffer(buffer);

View File

@@ -33,6 +33,10 @@ public abstract class Request<C extends Context> {
protected boolean keepAlive;
//请求包是否完成读取完毕用于ProtocolCodec继续读的判断条件
//需要在readHeader方法中设置
protected boolean completed;
protected int pipelineIndex;
protected int pipelineCount;
@@ -103,6 +107,7 @@ public abstract class Request<C extends Context> {
pipelineIndex = 0;
pipelineCount = 0;
pipelineCompleted = false;
completed = false;
keepAlive = false;
attributes.clear();
channel = null; // close it by response

View File

@@ -329,11 +329,14 @@ public abstract class Response<C extends Context, R extends Request<C>> {
}
this.recycleListener = null;
}
boolean completed = request.completed;
if (request.keepAlive && (request.pipelineIndex == 0 || request.pipelineCompleted)) {
AsyncConnection conn = removeChannel();
if (conn != null && conn.protocolCodec != null) {
this.responseConsumer.accept(this);
conn.readRegister(conn.protocolCodec);
if (!completed) {
conn.readRegister(conn.protocolCodec);
}
} else {
Supplier<Response> poolSupplier = this.responseSupplier;
Consumer<Response> poolConsumer = this.responseConsumer;

View File

@@ -295,6 +295,7 @@ public class HttpRequest extends Request<HttpContext> {
this.readState = READ_STATE_HEADER;
}
if (this.readState == READ_STATE_HEADER) {
this.completed = true;
if (last != null && ((HttpRequest) last).headerLength > 0) {
final HttpRequest httplast = (HttpRequest) last;
int bufremain = buffer.remaining();
@@ -350,6 +351,7 @@ public class HttpRequest extends Request<HttpContext> {
this.boundary = true;
}
if (this.boundary) {
this.completed = false; //completed=true时ProtocolCodec会继续读下一个request
this.keepAlive = false; //文件上传必须设置keepAlive为false因为文件过大时用户不一定会skip掉多余的数据
}
this.readState = READ_STATE_BODY;

View File

@@ -83,6 +83,7 @@ public class SncpRequest extends Request<SncpContext> {
}
//---------------------head----------------------------------
if (this.readState == READ_STATE_HEADER) {
this.completed = true;
int remain = buffer.remaining();
int expect = halfArray == null ? this.headerSize - 2 : this.headerSize - 2 - halfArray.length();
if (remain < expect) {

View File

@@ -66,6 +66,6 @@ public class SncpSleepTest {
System.out.println("耗时: " + e + " ms");
server.shutdown();
workExecutor.shutdown();
Assertions.assertTrue(e < 900);
Assertions.assertTrue(e < 600);
}
}