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
2017-12-29 10:19:20 +08:00
parent 46a89d9847
commit 6413161dc8
7 changed files with 119 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/org/redkale/net/PrepareRunner.java
View File

@@ -48,14 +48,16 @@ public final class PrepareRunner implements Runnable {
}
return;
}
final ByteBuffer buffer = context.pollBuffer();
final Response response = responsePool.get();
final ByteBuffer buffer = response.request.pollReadBuffer();
try {
channel.read(buffer, null, new CompletionHandler<Integer, Void>() {
@Override
public void completed(Integer count, Void attachment1) {
if (count < 1 && buffer.remaining() == buffer.limit()) {
try {
context.offerBuffer(buffer);
response.request.offerReadBuffer(buffer);
response.finish(true);
channel.close();
} catch (Exception e) {
context.logger.log(Level.FINEST, "PrepareRunner close channel erroneous on no read bytes", e);
@@ -69,7 +71,6 @@ public final class PrepareRunner implements Runnable {
// System.println(new String(bs));
// }
buffer.flip();
final Response response = responsePool.get();
response.init(channel);
try {
prepare.prepare(buffer, response.request, response);
@@ -81,7 +82,8 @@ public final class PrepareRunner implements Runnable {
@Override
public void failed(Throwable exc, Void attachment2) {
context.offerBuffer(buffer);
response.request.offerReadBuffer(buffer);
response.finish(true);
try {
channel.close();
} catch (Exception e) {
@@ -90,7 +92,8 @@ public final class PrepareRunner implements Runnable {
}
});
} catch (Exception te) {
context.offerBuffer(buffer);
response.request.offerReadBuffer(buffer);
response.finish(true);
try {
channel.close();
} catch (Exception e) {

8
src/org/redkale/net/PrepareServlet.java
View File

@@ -214,11 +214,11 @@ public abstract class PrepareServlet<K extends Serializable, C extends Context,
executeCounter.incrementAndGet();
final int rs = request.readHeader(buffer);
if (rs < 0) {
response.context.offerBuffer(buffer);
request.offerReadBuffer(buffer);
if (rs != Integer.MIN_VALUE) illRequestCounter.incrementAndGet();
response.finish(true);
} else if (rs == 0) {
response.context.offerBuffer(buffer);
request.offerReadBuffer(buffer);
request.prepare();
response.filter = this.headFilter;
response.servlet = this;
@@ -236,7 +236,7 @@ public abstract class PrepareServlet<K extends Serializable, C extends Context,
buffer.clear();
request.channel.read(buffer, buffer, this);
} else {
response.context.offerBuffer(buffer);
request.offerReadBuffer(buffer);
request.prepare();
try {
response.filter = PrepareServlet.this.headFilter;
@@ -253,7 +253,7 @@ public abstract class PrepareServlet<K extends Serializable, C extends Context,
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
illRequestCounter.incrementAndGet();
response.context.offerBuffer(buffer);
request.offerReadBuffer(buffer);
response.finish(true);
if (exc != null) request.context.logger.log(Level.FINER, "Servlet read channel erroneous, forece to close channel ", exc);
}

20
src/org/redkale/net/Request.java
View File

@@ -33,6 +33,8 @@ public abstract class Request<C extends Context> {
protected AsyncConnection channel;
protected ByteBuffer readBuffer;
/**
* properties 与 attributes 的区别在于：调用recycle时， attributes会被清空而properties会保留;
* properties 通常存放需要永久绑定在request里的一些对象
@@ -43,10 +45,28 @@ public abstract class Request<C extends Context> {
protected Request(C context) {
this.context = context;
this.readBuffer = context.pollBuffer();
this.bsonConvert = context.getBsonConvert();
this.jsonConvert = context.getJsonConvert();
}
protected ByteBuffer pollReadBuffer() {
ByteBuffer buffer = this.readBuffer;
this.readBuffer = null;
if (buffer == null) buffer = context.pollBuffer();
return buffer;
}
protected void offerReadBuffer(ByteBuffer buffer) {
if (buffer == null) return;
if (this.readBuffer == null) {
buffer.clear();
this.readBuffer = buffer;
} else {
context.offerBuffer(buffer);
}
}
/**
* 返回值：Integer.MIN_VALUE: 帧数据； -1：数据不合法； 0：解析完毕； &gt;0: 需再读取的字节数。
*

78
src/org/redkale/net/Response.java
View File

@@ -8,7 +8,7 @@ package org.redkale.net;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.CompletionHandler;
import java.util.function.BiConsumer;
import java.util.function.*;
import java.util.logging.Level;
/**
@@ -30,6 +30,10 @@ public abstract class Response<C extends Context, R extends Request<C>> {
protected AsyncConnection channel;
protected ByteBuffer writeHeadBuffer;
protected ByteBuffer writeBodyBuffer;
private boolean inited = true;
protected Object output; //输出的结果对象
@@ -40,6 +44,8 @@ public abstract class Response<C extends Context, R extends Request<C>> {
protected Servlet<C, R, ? extends Response<C, R>> servlet;
private Supplier<ByteBuffer> bodyBufferSupplier;
private final CompletionHandler finishHandler = new CompletionHandler<Integer, ByteBuffer>() {
@Override
@@ -47,17 +53,31 @@ public abstract class Response<C extends Context, R extends Request<C>> {
if (attachment.hasRemaining()) {
channel.write(attachment, attachment, this);
} else {
context.offerBuffer(attachment);
offerResponseBuffer(attachment);
finish();
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
context.offerBuffer(attachment);
offerResponseBuffer(attachment);
finish(true);
}
private void offerResponseBuffer(ByteBuffer attachment) {
if (writeHeadBuffer == null) {
if (context.bufferPool.getRecyclerPredicate().test(attachment)) {
writeHeadBuffer = attachment;
}
} else if (writeBodyBuffer == null) {
if (context.bufferPool.getRecyclerPredicate().test(attachment)) {
writeBodyBuffer = attachment;
}
} else {
context.offerBuffer(attachment);
}
}
};
private final CompletionHandler finishHandler2 = new CompletionHandler<Integer, ByteBuffer[]>() {
@@ -74,26 +94,66 @@ public abstract class Response<C extends Context, R extends Request<C>> {
if (index >= 0) {
channel.write(attachments, index, attachments.length - index, attachments, this);
} else {
for (ByteBuffer attachment : attachments) {
context.offerBuffer(attachment);
}
offerResponseBuffer(attachments);
finish();
}
}
@Override
public void failed(Throwable exc, final ByteBuffer[] attachments) {
for (ByteBuffer attachment : attachments) {
context.offerBuffer(attachment);
}
offerResponseBuffer(attachments);
finish(true);
}
private void offerResponseBuffer(ByteBuffer[] attachments) {
int start = 0;
if (writeHeadBuffer == null && attachments.length > start) {
if (context.bufferPool.getRecyclerPredicate().test(attachments[start])) {
writeHeadBuffer = attachments[start];
start++;
}
}
if (writeBodyBuffer == null && attachments.length > start) {
if (context.bufferPool.getRecyclerPredicate().test(attachments[start])) {
writeBodyBuffer = attachments[start];
start++;
}
}
for (int i = start; i < attachments.length; i++) {
context.offerBuffer(attachments[i]);
}
}
};
protected Response(C context, final R request) {
this.context = context;
this.request = request;
this.writeHeadBuffer = context.pollBuffer();
this.writeBodyBuffer = context.pollBuffer();
this.bodyBufferSupplier = () -> {
ByteBuffer buffer = writeBodyBuffer;
if (buffer == null) return context.pollBuffer();
writeBodyBuffer = null;
return buffer;
};
}
protected ByteBuffer pollWriteReadBuffer() {
ByteBuffer buffer = this.writeHeadBuffer;
this.writeHeadBuffer = null;
if (buffer == null) buffer = context.pollBuffer();
return buffer;
}
protected ByteBuffer pollWriteBodyBuffer() {
ByteBuffer buffer = this.writeBodyBuffer;
this.writeBodyBuffer = null;
if (buffer == null) buffer = context.pollBuffer();
return buffer;
}
protected Supplier<ByteBuffer> getBodyBufferSupplier() {
return bodyBufferSupplier;
}
protected AsyncConnection removeChannel() {

26
src/org/redkale/net/http/HttpResponse.java
View File

@@ -254,7 +254,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
public void finishJson(final Object obj) {
this.contentType = "text/plain; charset=utf-8";
if (this.recycleListener != null) this.output = obj;
finish(request.getJsonConvert().convertTo(context.getBufferSupplier(), obj));
finish(request.getJsonConvert().convertTo(getBodyBufferSupplier(), obj));
}
/**
@@ -266,7 +266,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
public void finishMapJson(final Object... objs) {
this.contentType = "text/plain; charset=utf-8";
if (this.recycleListener != null) this.output = objs;
finish(request.getJsonConvert().convertMapTo(context.getBufferSupplier(), objs));
finish(request.getJsonConvert().convertMapTo(getBodyBufferSupplier(), objs));
}
/**
@@ -278,7 +278,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
public void finishJson(final JsonConvert convert, final Object obj) {
this.contentType = "text/plain; charset=utf-8";
if (this.recycleListener != null) this.output = obj;
finish(convert.convertTo(context.getBufferSupplier(), obj));
finish(convert.convertTo(getBodyBufferSupplier(), obj));
}
/**
@@ -291,7 +291,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
public void finishMapJson(final JsonConvert convert, final Object... objs) {
this.contentType = "text/plain; charset=utf-8";
if (this.recycleListener != null) this.output = objs;
finish(convert.convertMapTo(context.getBufferSupplier(), objs));
finish(convert.convertMapTo(getBodyBufferSupplier(), objs));
}
/**
@@ -303,7 +303,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
public void finishJson(final Type type, final Object obj) {
this.contentType = "text/plain; charset=utf-8";
this.output = obj;
finish(request.getJsonConvert().convertTo(context.getBufferSupplier(), type, obj));
finish(request.getJsonConvert().convertTo(getBodyBufferSupplier(), type, obj));
}
/**
@@ -316,7 +316,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
public void finishJson(final JsonConvert convert, final Type type, final Object obj) {
this.contentType = "text/plain; charset=utf-8";
if (this.recycleListener != null) this.output = obj;
finish(convert.convertTo(context.getBufferSupplier(), type, obj));
finish(convert.convertTo(getBodyBufferSupplier(), type, obj));
}
/**
@@ -327,7 +327,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
public void finishJson(final Object... objs) {
this.contentType = "text/plain; charset=utf-8";
if (this.recycleListener != null) this.output = objs;
finish(request.getJsonConvert().convertTo(context.getBufferSupplier(), objs));
finish(request.getJsonConvert().convertTo(getBodyBufferSupplier(), objs));
}
/**
@@ -342,7 +342,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
this.header.addValue("retcode", String.valueOf(ret.getRetcode()));
this.header.addValue("retinfo", ret.getRetinfo());
}
finish(request.getJsonConvert().convertTo(context.getBufferSupplier(), ret));
finish(request.getJsonConvert().convertTo(getBodyBufferSupplier(), ret));
}
/**
@@ -358,7 +358,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
this.header.addValue("retcode", String.valueOf(ret.getRetcode()));
this.header.addValue("retinfo", ret.getRetinfo());
}
finish(convert.convertTo(context.getBufferSupplier(), ret));
finish(convert.convertTo(getBodyBufferSupplier(), ret));
}
/**
@@ -467,8 +467,8 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
this.header.addValue("retcode", String.valueOf(ret.getRetcode())).addValue("retinfo", ret.getRetinfo());
}
}
ByteBuffer[] buffers = type == null ? convert.convertTo(context.getBufferSupplier(), obj)
: convert.convertTo(context.getBufferSupplier(), type, obj);
ByteBuffer[] buffers = type == null ? convert.convertTo(getBodyBufferSupplier(), obj)
: convert.convertTo(getBodyBufferSupplier(), type, obj);
finish(buffers);
}
}
@@ -550,7 +550,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
@Override
public void finish(final byte[] bs) {
if (isClosed()) return; //避免重复关闭
if (this.context.getBufferCapacity() == bs.length) {
if (this.context.getBufferCapacity() >= bs.length) {
ByteBuffer buffer = this.context.pollBuffer();
buffer.put(bs);
buffer.flip();
@@ -796,7 +796,7 @@ public class HttpResponse extends Response<HttpContext, HttpRequest> {
//Header大小不能超过一个ByteBuffer的容量
protected ByteBuffer createHeader() {
this.headsended = true;
ByteBuffer buffer = this.context.pollBuffer();
ByteBuffer buffer = this.pollWriteReadBuffer();
buffer.put(("HTTP/1.1 " + this.status + " " + (this.status == 200 ? "OK" : httpCodes.get(this.status)) + "\r\n").getBytes());
buffer.put(("Content-Type: " + (this.contentType == null ? "text/plain; charset=utf-8" : this.contentType) + "\r\n").getBytes());

2
src/org/redkale/net/sncp/SncpResponse.java
View File

@@ -54,7 +54,7 @@ public final class SncpResponse extends Response<SncpContext, SncpRequest> {
public void finish(final int retcode, final BsonWriter out) {
if (out == null) {
final ByteBuffer buffer = context.pollBuffer();
final ByteBuffer buffer = pollWriteReadBuffer();
fillHeader(buffer, 0, retcode);
finish(buffer);
return;

4
src/org/redkale/util/ObjectPool.java
View File

@@ -67,6 +67,10 @@ public final class ObjectPool<T> implements Supplier<T>, Consumer<T> {
this.creator = creator;
}
public Predicate<T> getRecyclerPredicate() {
return recycler;
}
@Override
public T get() {
T result = queue.poll();