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

Browse Source
This commit is contained in:
Redkale
2018-04-17 18:30:41 +08:00
parent bc99f3f295
commit 8c0db6db53
15 changed files with 138 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/META-INF/application-template.xml
View File

@@ -37,11 +37,11 @@
threads： 线程总数， 默认: <group>节点数*CPU核数*2
bufferCapacity: ByteBuffer的初始化大小， 默认: 32K;
bufferPoolSize： ByteBuffer池的大小，默认: 线程总数*4
readTimeoutSecond: TCP读取超时秒数, 默认为6秒， 为0表示无超时限制
writeTimeoutSecond: TCP写入超时秒数, 默认为6秒， 为0表示无超时限制
readTimeoutSeconds: TCP读取超时秒数, 默认为6秒， 为0表示无超时限制
writeTimeoutSeconds: TCP写入超时秒数, 默认为6秒， 为0表示无超时限制
strategy: 远程请求的负载均衡策略， 必须是org.redkale.net.TransportStrategy的实现类
-->
<transport bufferCapacity="32K" bufferPoolSize="32" threads="32" readTimeoutSecond="6" writeTimeoutSecond="6"/>
<transport bufferCapacity="32K" bufferPoolSize="32" threads="32" readTimeoutSeconds="6" writeTimeoutSeconds="6"/>
<!--
一个组包含多个node， 同一Service服务可以由多个进程提供，这些进程称为一个GROUP，且同一GROUP内的进程必须在同一机房或局域网内
@@ -122,9 +122,9 @@
bufferCapacity: ByteBuffer的初始化大小， 默认: 32K; (HTTP 2.0、WebSocket，必须要16k以上)
bufferPoolSize： ByteBuffer池的大小，默认: 线程总数*4
responsePoolSize： Response池的大小，默认: 线程总数*2
aliveTimeoutSecond: KeepAlive读操作超时秒数， 默认0， 表示永久不超时,-1表示禁止KeepAlive
readTimeoutSecond: 读操作超时秒数， 默认0， 表示永久不超时
writeTimeoutSecond: 写操作超时秒数， 默认0， 表示永久不超时
aliveTimeoutSeconds: KeepAlive读操作超时秒数， 默认0， 表示永久不超时,-1表示禁止KeepAlive
readTimeoutSeconds: 读操作超时秒数， 默认0， 表示永久不超时
writeTimeoutSeconds: 写操作超时秒数， 默认0， 表示永久不超时
interceptor: 启动/关闭NodeServer时被调用的拦截器实现类，必须是org.redkale.boot.NodeInterceptor的子类，默认为null
-->
<server protocol="HTTP" host="127.0.0.1" port="6060" root="root" lib="">

10
src/org/redkale/boot/Application.java
View File

@@ -249,8 +249,8 @@ public final class Application {
TransportStrategy strategy = null;
int bufferCapacity = 32 * 1024;
int bufferPoolSize = Runtime.getRuntime().availableProcessors() * 8;
int readTimeoutSecond = TransportFactory.DEFAULT_READTIMEOUTSECOND;
int writeTimeoutSecond = TransportFactory.DEFAULT_WRITETIMEOUTSECOND;
int readTimeoutSeconds = TransportFactory.DEFAULT_READTIMEOUTSECONDS;
int writeTimeoutSeconds = TransportFactory.DEFAULT_WRITETIMEOUTSECONDS;
AtomicLong createBufferCounter = new AtomicLong();
AtomicLong cycleBufferCounter = new AtomicLong();
if (resources != null) {
@@ -260,8 +260,8 @@ public final class Application {
if (transportConf != null) {
//--------------transportBufferPool-----------
bufferCapacity = Math.max(parseLenth(transportConf.getValue("bufferCapacity"), bufferCapacity), 8 * 1024);
readTimeoutSecond = transportConf.getIntValue("readTimeoutSecond", readTimeoutSecond);
writeTimeoutSecond = transportConf.getIntValue("writeTimeoutSecond", writeTimeoutSecond);
readTimeoutSeconds = transportConf.getIntValue("readTimeoutSeconds", readTimeoutSeconds);
writeTimeoutSeconds = transportConf.getIntValue("writeTimeoutSeconds", writeTimeoutSeconds);
final int threads = parseLenth(transportConf.getValue("threads"), groupsize * Runtime.getRuntime().availableProcessors() * 2);
bufferPoolSize = parseLenth(transportConf.getValue("bufferPoolSize"), threads * 4);
final int capacity = bufferCapacity;
@@ -314,7 +314,7 @@ public final class Application {
return true;
});
}
this.sncpTransportFactory = TransportFactory.create(transportExec, transportPool, transportGroup, (SSLContext) null, readTimeoutSecond, writeTimeoutSecond, strategy);
this.sncpTransportFactory = TransportFactory.create(transportExec, transportPool, transportGroup, (SSLContext) null, readTimeoutSeconds, writeTimeoutSeconds, strategy);
DefaultAnyValue tarnsportConf = DefaultAnyValue.create(TransportFactory.NAME_POOLMAXCONNS, System.getProperty("net.transport.poolmaxconns", "100"))
.addValue(TransportFactory.NAME_PINGINTERVAL, System.getProperty("net.transport.pinginterval", "30"))
.addValue(TransportFactory.NAME_CHECKINTERVAL, System.getProperty("net.transport.checkinterval", "30"));

136
src/org/redkale/net/AsyncConnection.java
View File

@@ -53,13 +53,13 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
public abstract SocketAddress getLocalAddress();
public abstract int getReadTimeoutSecond();
public abstract int getReadTimeoutSeconds();
public abstract int getWriteTimeoutSecond();
public abstract int getWriteTimeoutSeconds();
public abstract void setReadTimeoutSecond(int readTimeoutSecond);
public abstract void setReadTimeoutSeconds(int readTimeoutSeconds);
public abstract void setWriteTimeoutSecond(int writeTimeoutSecond);
public abstract void setWriteTimeoutSeconds(int writeTimeoutSeconds);
@Override
public abstract Future<Integer> read(ByteBuffer dst);
@@ -144,14 +144,14 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
*
* @param address 连接点子
* @param group 连接AsynchronousChannelGroup
* @param readTimeoutSecond 读取超时秒数
* @param writeTimeoutSecond 写入超时秒数
* @param readTimeoutSeconds 读取超时秒数
* @param writeTimeoutSeconds 写入超时秒数
*
* @return 连接CompletableFuture
*/
public static CompletableFuture<AsyncConnection> createTCP(final AsynchronousChannelGroup group, final SocketAddress address,
final int readTimeoutSecond, final int writeTimeoutSecond) {
return createTCP(group, null, address, false, readTimeoutSecond, writeTimeoutSecond);
final int readTimeoutSeconds, final int writeTimeoutSeconds) {
return createTCP(group, null, address, false, readTimeoutSeconds, writeTimeoutSeconds);
}
/**
@@ -160,14 +160,14 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
* @param address 连接点子
* @param sslContext SSLContext
* @param group 连接AsynchronousChannelGroup
* @param readTimeoutSecond 读取超时秒数
* @param writeTimeoutSecond 写入超时秒数
* @param readTimeoutSeconds 读取超时秒数
* @param writeTimeoutSeconds 写入超时秒数
*
* @return 连接CompletableFuture
*/
public static CompletableFuture<AsyncConnection> createTCP(final AsynchronousChannelGroup group, final SSLContext sslContext,
final SocketAddress address, final int readTimeoutSecond, final int writeTimeoutSecond) {
return createTCP(group, sslContext, address, false, readTimeoutSecond, writeTimeoutSecond);
final SocketAddress address, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
return createTCP(group, sslContext, address, false, readTimeoutSeconds, writeTimeoutSeconds);
}
/**
@@ -177,13 +177,13 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
* @param sslContext SSLContext
* @param group 连接AsynchronousChannelGroup
* @param noDelay TcpNoDelay
* @param readTimeoutSecond 读取超时秒数
* @param writeTimeoutSecond 写入超时秒数
* @param readTimeoutSeconds 读取超时秒数
* @param writeTimeoutSeconds 写入超时秒数
*
* @return 连接CompletableFuture
*/
public static CompletableFuture<AsyncConnection> createTCP(final AsynchronousChannelGroup group, final SSLContext sslContext,
final SocketAddress address, final boolean noDelay, final int readTimeoutSecond, final int writeTimeoutSecond) {
final SocketAddress address, final boolean noDelay, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
final CompletableFuture<AsyncConnection> future = new CompletableFuture<>();
try {
final AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(group);
@@ -196,7 +196,7 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
} catch (IOException e) {
}
}
future.complete(create(channel, sslContext, address, readTimeoutSecond, writeTimeoutSecond));
future.complete(create(channel, sslContext, address, readTimeoutSeconds, writeTimeoutSeconds));
}
@Override
@@ -212,9 +212,9 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
private static class BIOUDPAsyncConnection extends AsyncConnection {
private int readTimeoutSecond;
private int readTimeoutSeconds;
private int writeTimeoutSecond;
private int writeTimeoutSeconds;
private final DatagramChannel channel;
@@ -223,32 +223,32 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
private final boolean client;
public BIOUDPAsyncConnection(final DatagramChannel ch, SocketAddress addr,
final boolean client0, final int readTimeoutSecond0, final int writeTimeoutSecond0) {
final boolean client0, final int readTimeoutSeconds0, final int writeTimeoutSeconds0) {
this.channel = ch;
this.client = client0;
this.readTimeoutSecond = readTimeoutSecond0;
this.writeTimeoutSecond = writeTimeoutSecond0;
this.readTimeoutSeconds = readTimeoutSeconds0;
this.writeTimeoutSeconds = writeTimeoutSeconds0;
this.remoteAddress = addr;
}
@Override
public void setReadTimeoutSecond(int readTimeoutSecond) {
this.readTimeoutSecond = readTimeoutSecond;
public void setReadTimeoutSeconds(int readTimeoutSeconds) {
this.readTimeoutSeconds = readTimeoutSeconds;
}
@Override
public void setWriteTimeoutSecond(int writeTimeoutSecond) {
this.writeTimeoutSecond = writeTimeoutSecond;
public void setWriteTimeoutSeconds(int writeTimeoutSeconds) {
this.writeTimeoutSeconds = writeTimeoutSeconds;
}
@Override
public int getReadTimeoutSecond() {
return this.readTimeoutSecond;
public int getReadTimeoutSeconds() {
return this.readTimeoutSeconds;
}
@Override
public int getWriteTimeoutSecond() {
return this.writeTimeoutSecond;
public int getWriteTimeoutSeconds() {
return this.writeTimeoutSeconds;
}
@Override
@@ -347,15 +347,15 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
}
public static AsyncConnection create(final DatagramChannel ch, SocketAddress addr,
final boolean client0, final int readTimeoutSecond0, final int writeTimeoutSecond0) {
return new BIOUDPAsyncConnection(ch, addr, client0, readTimeoutSecond0, writeTimeoutSecond0);
final boolean client0, final int readTimeoutSeconds0, final int writeTimeoutSeconds0) {
return new BIOUDPAsyncConnection(ch, addr, client0, readTimeoutSeconds0, writeTimeoutSeconds0);
}
private static class BIOTCPAsyncConnection extends AsyncConnection {
private int readTimeoutSecond;
private int readTimeoutSeconds;
private int writeTimeoutSecond;
private int writeTimeoutSeconds;
private final Socket socket;
@@ -365,12 +365,12 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
private final SocketAddress remoteAddress;
public BIOTCPAsyncConnection(final Socket socket, final SocketAddress addr0, final int readTimeoutSecond0, final int writeTimeoutSecond0) {
public BIOTCPAsyncConnection(final Socket socket, final SocketAddress addr0, final int readTimeoutSeconds0, final int writeTimeoutSeconds0) {
this.socket = socket;
ReadableByteChannel rc = null;
WritableByteChannel wc = null;
try {
socket.setSoTimeout(Math.max(readTimeoutSecond0, writeTimeoutSecond0));
socket.setSoTimeout(Math.max(readTimeoutSeconds0, writeTimeoutSeconds0));
rc = Channels.newChannel(socket.getInputStream());
wc = Channels.newChannel(socket.getOutputStream());
} catch (IOException e) {
@@ -378,8 +378,8 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
}
this.readChannel = rc;
this.writeChannel = wc;
this.readTimeoutSecond = readTimeoutSecond0;
this.writeTimeoutSecond = writeTimeoutSecond0;
this.readTimeoutSeconds = readTimeoutSeconds0;
this.writeTimeoutSeconds = writeTimeoutSeconds0;
SocketAddress addr = addr0;
if (addr == null) {
try {
@@ -407,23 +407,23 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
}
@Override
public int getReadTimeoutSecond() {
return readTimeoutSecond;
public int getReadTimeoutSeconds() {
return readTimeoutSeconds;
}
@Override
public int getWriteTimeoutSecond() {
return writeTimeoutSecond;
public int getWriteTimeoutSeconds() {
return writeTimeoutSeconds;
}
@Override
public void setReadTimeoutSecond(int readTimeoutSecond) {
this.readTimeoutSecond = readTimeoutSecond;
public void setReadTimeoutSeconds(int readTimeoutSeconds) {
this.readTimeoutSeconds = readTimeoutSeconds;
}
@Override
public void setWriteTimeoutSecond(int writeTimeoutSecond) {
this.writeTimeoutSecond = writeTimeoutSecond;
public void setWriteTimeoutSeconds(int writeTimeoutSeconds) {
this.writeTimeoutSeconds = writeTimeoutSeconds;
}
@Override
@@ -518,20 +518,20 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
private static class AIOTCPAsyncConnection extends AsyncConnection {
private int readTimeoutSecond;
private int readTimeoutSeconds;
private int writeTimeoutSecond;
private int writeTimeoutSeconds;
private final AsynchronousSocketChannel channel;
private final SocketAddress remoteAddress;
public AIOTCPAsyncConnection(final AsynchronousSocketChannel ch, SSLContext sslContext,
final SocketAddress addr0, final int readTimeoutSecond0, final int writeTimeoutSecond0) {
final SocketAddress addr0, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
this.channel = ch;
this.sslContext = sslContext;
this.readTimeoutSecond = readTimeoutSecond0;
this.writeTimeoutSecond = writeTimeoutSecond0;
this.readTimeoutSeconds = readTimeoutSeconds;
this.writeTimeoutSeconds = writeTimeoutSeconds;
SocketAddress addr = addr0;
if (addr == null) {
try {
@@ -546,8 +546,8 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
@Override
public <A> void read(ByteBuffer dst, A attachment, CompletionHandler<Integer, ? super A> handler) {
this.readtime = System.currentTimeMillis();
if (readTimeoutSecond > 0) {
channel.read(dst, readTimeoutSecond, TimeUnit.SECONDS, attachment, handler);
if (readTimeoutSeconds > 0) {
channel.read(dst, readTimeoutSeconds, TimeUnit.SECONDS, attachment, handler);
} else {
channel.read(dst, attachment, handler);
}
@@ -562,8 +562,8 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
@Override
public <A> void write(ByteBuffer src, A attachment, CompletionHandler<Integer, ? super A> handler) {
this.writetime = System.currentTimeMillis();
if (writeTimeoutSecond > 0) {
channel.write(src, writeTimeoutSecond, TimeUnit.SECONDS, attachment, handler);
if (writeTimeoutSeconds > 0) {
channel.write(src, writeTimeoutSeconds, TimeUnit.SECONDS, attachment, handler);
} else {
channel.write(src, attachment, handler);
}
@@ -572,7 +572,7 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
@Override
public <A> void write(ByteBuffer[] srcs, int offset, int length, A attachment, final CompletionHandler<Integer, ? super A> handler) {
this.writetime = System.currentTimeMillis();
channel.write(srcs, offset, length, writeTimeoutSecond > 0 ? writeTimeoutSecond : 60, TimeUnit.SECONDS,
channel.write(srcs, offset, length, writeTimeoutSeconds > 0 ? writeTimeoutSeconds : 60, TimeUnit.SECONDS,
attachment, new CompletionHandler<Long, A>() {
@Override
@@ -589,23 +589,23 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
}
@Override
public void setReadTimeoutSecond(int readTimeoutSecond) {
this.readTimeoutSecond = readTimeoutSecond;
public void setReadTimeoutSeconds(int readTimeoutSeconds) {
this.readTimeoutSeconds = readTimeoutSeconds;
}
@Override
public void setWriteTimeoutSecond(int writeTimeoutSecond) {
this.writeTimeoutSecond = writeTimeoutSecond;
public void setWriteTimeoutSeconds(int writeTimeoutSeconds) {
this.writeTimeoutSeconds = writeTimeoutSeconds;
}
@Override
public int getReadTimeoutSecond() {
return this.readTimeoutSecond;
public int getReadTimeoutSeconds() {
return this.readTimeoutSeconds;
}
@Override
public int getWriteTimeoutSecond() {
return this.writeTimeoutSecond;
public int getWriteTimeoutSeconds() {
return this.writeTimeoutSeconds;
}
@Override
@@ -654,15 +654,15 @@ public abstract class AsyncConnection implements AsynchronousByteChannel, AutoCl
return create(ch, null, 0, 0);
}
public static AsyncConnection create(final AsynchronousSocketChannel ch, final SocketAddress addr0, final int readTimeoutSecond, final int writeTimeoutSecond) {
return new AIOTCPAsyncConnection(ch, null, addr0, readTimeoutSecond, writeTimeoutSecond);
public static AsyncConnection create(final AsynchronousSocketChannel ch, final SocketAddress addr0, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
return new AIOTCPAsyncConnection(ch, null, addr0, readTimeoutSeconds, writeTimeoutSeconds);
}
public static AsyncConnection create(final AsynchronousSocketChannel ch, SSLContext sslContext, final SocketAddress addr0, final int readTimeoutSecond, final int writeTimeoutSecond) {
return new AIOTCPAsyncConnection(ch, sslContext, addr0, readTimeoutSecond, writeTimeoutSecond);
public static AsyncConnection create(final AsynchronousSocketChannel ch, SSLContext sslContext, final SocketAddress addr0, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
return new AIOTCPAsyncConnection(ch, sslContext, addr0, readTimeoutSeconds, writeTimeoutSeconds);
}
public static AsyncConnection create(final AsynchronousSocketChannel ch, final SocketAddress addr0, final Context context) {
return new AIOTCPAsyncConnection(ch, context.sslContext, addr0, context.readTimeoutSecond, context.writeTimeoutSecond);
return new AIOTCPAsyncConnection(ch, context.sslContext, addr0, context.readTimeoutSeconds, context.writeTimeoutSeconds);
}
}

26
src/org/redkale/net/Context.java
View File

@@ -59,13 +59,13 @@ public class Context {
protected final int maxbody;
//keep alive IO读取的超时时间
protected final int aliveTimeoutSecond;
protected final int aliveTimeoutSeconds;
//IO读取的超时时间
protected final int readTimeoutSecond;
protected final int readTimeoutSeconds;
//IO写入的超时时间
protected final int writeTimeoutSecond;
protected final int writeTimeoutSeconds;
//日志Logger
protected final Logger logger;
@@ -82,7 +82,7 @@ public class Context {
public Context(long serverStartTime, Logger logger, ThreadPoolExecutor executor, SSLContext sslContext,
int bufferCapacity, ObjectPool<ByteBuffer> bufferPool, ObjectPool<Response> responsePool,
final int maxbody, Charset charset, InetSocketAddress address, ResourceFactory resourceFactory,
final PrepareServlet prepare, final int aliveTimeoutSecond, final int readTimeoutSecond, final int writeTimeoutSecond) {
final PrepareServlet prepare, final int aliveTimeoutSeconds, final int readTimeoutSeconds, final int writeTimeoutSeconds) {
this.serverStartTime = serverStartTime;
this.logger = logger;
this.executor = executor;
@@ -95,9 +95,9 @@ public class Context {
this.address = address;
this.prepare = prepare;
this.resourceFactory = resourceFactory;
this.aliveTimeoutSecond = aliveTimeoutSecond;
this.readTimeoutSecond = readTimeoutSecond;
this.writeTimeoutSecond = writeTimeoutSecond;
this.aliveTimeoutSeconds = aliveTimeoutSeconds;
this.readTimeoutSeconds = readTimeoutSeconds;
this.writeTimeoutSeconds = writeTimeoutSeconds;
this.jsonFactory = JsonFactory.root();
this.bsonFactory = BsonFactory.root();
}
@@ -169,16 +169,16 @@ public class Context {
return logger;
}
public int getAliveTimeoutSecond() {
return aliveTimeoutSecond;
public int getAliveTimeoutSeconds() {
return aliveTimeoutSeconds;
}
public int getReadTimeoutSecond() {
return readTimeoutSecond;
public int getReadTimeoutSeconds() {
return readTimeoutSeconds;
}
public int getWriteTimeoutSecond() {
return writeTimeoutSecond;
public int getWriteTimeoutSeconds() {
return writeTimeoutSeconds;
}
public JsonConvert getJsonConvert() {

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

@@ -56,7 +56,7 @@ public final class PrepareRunner implements Runnable {
if (response == null) response = responsePool.get();
final ByteBuffer buffer = response.request.pollReadBuffer();
try {
channel.read(buffer, keepalive ? context.getAliveTimeoutSecond() : 0, TimeUnit.SECONDS, null,
channel.read(buffer, keepalive ? context.getAliveTimeoutSeconds() : 0, TimeUnit.SECONDS, null,
new CompletionHandler<Integer, Void>() {
@Override
public void completed(Integer count, Void attachment1) {

6
src/org/redkale/net/ProtocolServer.java
View File

@@ -109,8 +109,8 @@ public abstract class ProtocolServer {
@Override
public void accept() {
final DatagramChannel serchannel = this.serverChannel;
final int readTimeoutSecond = this.context.readTimeoutSecond;
final int writeTimeoutSecond = this.context.writeTimeoutSecond;
final int readTimeoutSeconds = this.context.readTimeoutSeconds;
final int writeTimeoutSeconds = this.context.writeTimeoutSeconds;
final CountDownLatch cdl = new CountDownLatch(1);
this.running = true;
new Thread() {
@@ -122,7 +122,7 @@ public abstract class ProtocolServer {
try {
SocketAddress address = serchannel.receive(buffer);
buffer.flip();
AsyncConnection conn = AsyncConnection.create(serchannel, address, false, readTimeoutSecond, writeTimeoutSecond);
AsyncConnection conn = AsyncConnection.create(serchannel, address, false, readTimeoutSeconds, writeTimeoutSeconds);
context.runAsync(new PrepareRunner(context, conn, buffer, null));
} catch (Exception e) {
context.offerBuffer(buffer);

12
src/org/redkale/net/Server.java
View File

@@ -92,13 +92,13 @@ public abstract class Server<K extends Serializable, C extends Context, R extend
protected int maxbody;
//Keep-Alive IO读取的超时秒数，小于1视为不设置
protected int aliveTimeoutSecond;
protected int aliveTimeoutSeconds;
//IO读取的超时秒数，小于1视为不设置
protected int readTimeoutSecond;
protected int readTimeoutSeconds;
//IO写入 的超时秒数，小于1视为不设置
protected int writeTimeoutSecond;
protected int writeTimeoutSeconds;
//最大连接数
protected int maxconns;
@@ -116,9 +116,9 @@ public abstract class Server<K extends Serializable, C extends Context, R extend
this.address = new InetSocketAddress(config.getValue("host", "0.0.0.0"), config.getIntValue("port", 80));
this.charset = Charset.forName(config.getValue("charset", "UTF-8"));
this.maxconns = config.getIntValue("maxconns", 0);
this.aliveTimeoutSecond = config.getIntValue("aliveTimeoutSecond", 0);
this.readTimeoutSecond = config.getIntValue("readTimeoutSecond", 0);
this.writeTimeoutSecond = config.getIntValue("writeTimeoutSecond", 0);
this.aliveTimeoutSeconds = config.getIntValue("aliveTimeoutSeconds", 0);
this.readTimeoutSeconds = config.getIntValue("readTimeoutSeconds", 0);
this.writeTimeoutSeconds = config.getIntValue("writeTimeoutSeconds", 0);
this.backlog = parseLenth(config.getValue("backlog"), 16 * 1024);
this.maxbody = parseLenth(config.getValue("maxbody"), 64 * 1024);
int bufCapacity = parseLenth(config.getValue("bufferCapacity"), 32 * 1024);

10
src/org/redkale/net/Transport.java
View File

@@ -228,12 +228,12 @@ public final class Transport {
DatagramChannel channel = DatagramChannel.open();
channel.configureBlocking(true);
channel.connect(udpaddr);
return CompletableFuture.completedFuture(AsyncConnection.create(channel, udpaddr, true, factory.readTimeoutSecond, factory.writeTimeoutSecond));
return CompletableFuture.completedFuture(AsyncConnection.create(channel, udpaddr, true, factory.readTimeoutSeconds, factory.writeTimeoutSeconds));
}
if (!rand) { //指定地址
TransportNode node = findTransportNode(addr);
if (node == null) {
return AsyncConnection.createTCP(group, sslContext, addr, supportTcpNoDelay, factory.readTimeoutSecond, factory.writeTimeoutSecond);
return AsyncConnection.createTCP(group, sslContext, addr, supportTcpNoDelay, factory.readTimeoutSeconds, factory.writeTimeoutSeconds);
}
final BlockingQueue<AsyncConnection> queue = node.conns;
if (!queue.isEmpty()) {
@@ -242,7 +242,7 @@ public final class Transport {
if (conn.isOpen()) return CompletableFuture.completedFuture(conn);
}
}
return AsyncConnection.createTCP(group, sslContext, addr, supportTcpNoDelay, factory.readTimeoutSecond, factory.writeTimeoutSecond);
return AsyncConnection.createTCP(group, sslContext, addr, supportTcpNoDelay, factory.readTimeoutSeconds, factory.writeTimeoutSeconds);
}
//---------------------随机取地址------------------------
@@ -269,7 +269,7 @@ public final class Transport {
@Override
public void completed(Void result, TransportNode attachment) {
attachment.disabletime = 0;
AsyncConnection asyncConn = AsyncConnection.create(channel, attachment.address, factory.readTimeoutSecond, factory.writeTimeoutSecond);
AsyncConnection asyncConn = AsyncConnection.create(channel, attachment.address, factory.readTimeoutSeconds, factory.writeTimeoutSeconds);
if (future.isDone()) {
if (!attachment.conns.offer(asyncConn)) asyncConn.dispose();
} else {
@@ -319,7 +319,7 @@ public final class Transport {
@Override
public void completed(Void result, TransportNode attachment) {
attachment.disabletime = 0;
AsyncConnection asyncConn = AsyncConnection.create(channel, attachment.address, factory.readTimeoutSecond, factory.writeTimeoutSecond);
AsyncConnection asyncConn = AsyncConnection.create(channel, attachment.address, factory.readTimeoutSeconds, factory.writeTimeoutSeconds);
if (future.isDone()) {
if (!attachment.conns.offer(asyncConn)) asyncConn.dispose();
} else {

46
src/org/redkale/net/TransportFactory.java
View File

@@ -32,10 +32,10 @@ import org.redkale.util.*;
public class TransportFactory {
@Comment("默认TCP读取超时秒数")
public static int DEFAULT_READTIMEOUTSECOND = 6;
public static int DEFAULT_READTIMEOUTSECONDS = 6;
@Comment("默认TCP写入超时秒数")
public static int DEFAULT_WRITETIMEOUTSECOND = 6;
public static int DEFAULT_WRITETIMEOUTSECONDS = 6;
public static final String NAME_POOLMAXCONNS = "poolmaxconns";
@@ -74,10 +74,10 @@ public class TransportFactory {
protected int pinginterval;
//TCP读取超时秒数
protected int readTimeoutSecond;
protected int readTimeoutSeconds;
//TCP写入超时秒数
protected int writeTimeoutSecond;
protected int writeTimeoutSeconds;
//ping和检查的定时器
private ScheduledThreadPoolExecutor scheduler;
@@ -94,19 +94,19 @@ public class TransportFactory {
protected final TransportStrategy strategy;
protected TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
SSLContext sslContext, int readTimeoutSecond, int writeTimeoutSecond, final TransportStrategy strategy) {
SSLContext sslContext, int readTimeoutSeconds, int writeTimeoutSeconds, final TransportStrategy strategy) {
this.executor = executor;
this.bufferPool = bufferPool;
this.channelGroup = channelGroup;
this.sslContext = sslContext;
this.readTimeoutSecond = readTimeoutSecond;
this.writeTimeoutSecond = writeTimeoutSecond;
this.readTimeoutSeconds = readTimeoutSeconds;
this.writeTimeoutSeconds = writeTimeoutSeconds;
this.strategy = strategy;
}
protected TransportFactory(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
SSLContext sslContext, int readTimeoutSecond, int writeTimeoutSecond) {
this(executor, bufferPool, channelGroup, sslContext, readTimeoutSecond, writeTimeoutSecond, null);
SSLContext sslContext, int readTimeoutSeconds, int writeTimeoutSeconds) {
this(executor, bufferPool, channelGroup, sslContext, readTimeoutSeconds, writeTimeoutSeconds, null);
}
public void init(AnyValue conf, ByteBuffer pingBuffer, int pongLength) {
@@ -140,14 +140,14 @@ public class TransportFactory {
}
public static TransportFactory create(int threads) {
return create(threads, threads * 2, 8 * 1024, DEFAULT_READTIMEOUTSECOND, DEFAULT_WRITETIMEOUTSECOND);
return create(threads, threads * 2, 8 * 1024, DEFAULT_READTIMEOUTSECONDS, DEFAULT_WRITETIMEOUTSECONDS);
}
public static TransportFactory create(int threads, int bufferPoolSize, int bufferCapacity) {
return create(threads, bufferPoolSize, bufferCapacity, DEFAULT_READTIMEOUTSECOND, DEFAULT_WRITETIMEOUTSECOND);
return create(threads, bufferPoolSize, bufferCapacity, DEFAULT_READTIMEOUTSECONDS, DEFAULT_WRITETIMEOUTSECONDS);
}
public static TransportFactory create(int threads, int bufferPoolSize, int bufferCapacity, int readTimeoutSecond, int writeTimeoutSecond) {
public static TransportFactory create(int threads, int bufferPoolSize, int bufferCapacity, int readTimeoutSeconds, int writeTimeoutSeconds) {
final ObjectPool<ByteBuffer> transportPool = new ObjectPool<>(new AtomicLong(), new AtomicLong(), bufferPoolSize,
(Object... params) -> ByteBuffer.allocateDirect(bufferCapacity), null, (e) -> {
if (e == null || e.isReadOnly() || e.capacity() != bufferCapacity) return false;
@@ -167,35 +167,35 @@ public class TransportFactory {
} catch (IOException e) {
throw new RuntimeException(e);
}
return create(transportExec, transportPool, transportGroup, readTimeoutSecond, writeTimeoutSecond);
return create(transportExec, transportPool, transportGroup, readTimeoutSeconds, writeTimeoutSeconds);
}
public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup) {
return new TransportFactory(executor, bufferPool, channelGroup, null, DEFAULT_READTIMEOUTSECOND, DEFAULT_WRITETIMEOUTSECOND, null);
return new TransportFactory(executor, bufferPool, channelGroup, null, DEFAULT_READTIMEOUTSECONDS, DEFAULT_WRITETIMEOUTSECONDS, null);
}
public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
int readTimeoutSecond, int writeTimeoutSecond) {
return new TransportFactory(executor, bufferPool, channelGroup, null, readTimeoutSecond, writeTimeoutSecond, null);
int readTimeoutSeconds, int writeTimeoutSeconds) {
return new TransportFactory(executor, bufferPool, channelGroup, null, readTimeoutSeconds, writeTimeoutSeconds, null);
}
public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
int readTimeoutSecond, int writeTimeoutSecond, final TransportStrategy strategy) {
return new TransportFactory(executor, bufferPool, channelGroup, null, readTimeoutSecond, writeTimeoutSecond, strategy);
int readTimeoutSeconds, int writeTimeoutSeconds, final TransportStrategy strategy) {
return new TransportFactory(executor, bufferPool, channelGroup, null, readTimeoutSeconds, writeTimeoutSeconds, strategy);
}
public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup, SSLContext sslContext) {
return new TransportFactory(executor, bufferPool, channelGroup, sslContext, DEFAULT_READTIMEOUTSECOND, DEFAULT_WRITETIMEOUTSECOND, null);
return new TransportFactory(executor, bufferPool, channelGroup, sslContext, DEFAULT_READTIMEOUTSECONDS, DEFAULT_WRITETIMEOUTSECONDS, null);
}
public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
SSLContext sslContext, int readTimeoutSecond, int writeTimeoutSecond) {
return new TransportFactory(executor, bufferPool, channelGroup, sslContext, readTimeoutSecond, writeTimeoutSecond, null);
SSLContext sslContext, int readTimeoutSeconds, int writeTimeoutSeconds) {
return new TransportFactory(executor, bufferPool, channelGroup, sslContext, readTimeoutSeconds, writeTimeoutSeconds, null);
}
public static TransportFactory create(ExecutorService executor, ObjectPool<ByteBuffer> bufferPool, AsynchronousChannelGroup channelGroup,
SSLContext sslContext, int readTimeoutSecond, int writeTimeoutSecond, final TransportStrategy strategy) {
return new TransportFactory(executor, bufferPool, channelGroup, sslContext, readTimeoutSecond, writeTimeoutSecond, strategy);
SSLContext sslContext, int readTimeoutSeconds, int writeTimeoutSeconds, final TransportStrategy strategy) {
return new TransportFactory(executor, bufferPool, channelGroup, sslContext, readTimeoutSeconds, writeTimeoutSeconds, strategy);
}
public Transport createTransportTCP(String name, final InetSocketAddress clientAddress, final Collection<InetSocketAddress> addresses) {

4
src/org/redkale/net/http/HttpContext.java
View File

@@ -36,9 +36,9 @@ public class HttpContext extends Context {
public HttpContext(long serverStartTime, Logger logger, ThreadPoolExecutor executor, SSLContext sslContext,
final int bufferCapacity, final ObjectPool<ByteBuffer> bufferPool, ObjectPool<Response> responsePool,
int maxbody, Charset charset, InetSocketAddress address, ResourceFactory resourceFactory,
PrepareServlet prepare, int aliveTimeoutSecond, int readTimeoutSecond, int writeTimeoutSecond) {
PrepareServlet prepare, int aliveTimeoutSeconds, int readTimeoutSeconds, int writeTimeoutSeconds) {
super(serverStartTime, logger, executor, sslContext, bufferCapacity, bufferPool, responsePool,
maxbody, charset, address, resourceFactory, prepare, aliveTimeoutSecond, readTimeoutSecond, writeTimeoutSecond);
maxbody, charset, address, resourceFactory, prepare, aliveTimeoutSeconds, readTimeoutSeconds, writeTimeoutSeconds);
random.setSeed(Math.abs(System.nanoTime()));
}

2
src/org/redkale/net/http/HttpRequest.java
View File

@@ -154,7 +154,7 @@ public class HttpRequest extends Request<HttpContext> {
case "Connection":
case "connection":
this.connection = value;
if (context.getAliveTimeoutSecond() >= 0) {
if (context.getAliveTimeoutSeconds() >= 0) {
this.setKeepAlive(!"close".equalsIgnoreCase(value));
}
break;

2
src/org/redkale/net/http/HttpServer.java
View File

@@ -382,7 +382,7 @@ public class HttpServer extends Server<String, HttpContext, HttpRequest, HttpRes
ObjectPool<Response> responsePool = HttpResponse.createPool(createResponseCounter, cycleResponseCounter, this.responsePoolSize, null);
HttpContext httpcontext = new HttpContext(this.serverStartTime, this.logger, executor, this.sslContext,
rcapacity, bufferPool, responsePool, this.maxbody, this.charset, this.address, this.resourceFactory,
this.prepare, this.aliveTimeoutSecond, this.readTimeoutSecond, this.writeTimeoutSecond);
this.prepare, this.aliveTimeoutSeconds, this.readTimeoutSeconds, this.writeTimeoutSeconds);
responsePool.setCreator((Object... params) -> new HttpResponse(httpcontext, new HttpRequest(httpcontext, addrHeader),
plainType, jsonType, addHeaders, setHeaders, defCookie, options, ((HttpPrepareServlet) prepare).renders));
return httpcontext;

2
src/org/redkale/net/http/WebSocketRunner.java
View File

@@ -64,7 +64,7 @@ class WebSocketRunner implements Runnable {
final boolean debug = context.getLogger().isLoggable(Level.FINEST);
try {
webSocket.onConnected();
channel.setReadTimeoutSecond(300); //读取超时5分钟
channel.setReadTimeoutSeconds(300); //读取超时5分钟
if (channel.isOpen()) {
final int wsmaxbody = webSocket._engine.wsmaxbody;
channel.read(readBuffer, null, new CompletionHandler<Integer, Void>() {

4
src/org/redkale/net/sncp/SncpContext.java
View File

@@ -25,8 +25,8 @@ public class SncpContext extends Context {
public SncpContext(long serverStartTime, Logger logger, ThreadPoolExecutor executor, SSLContext sslContext,
int bufferCapacity, ObjectPool<ByteBuffer> bufferPool, ObjectPool<Response> responsePool,
int maxbody, Charset charset, InetSocketAddress address, ResourceFactory resourceFactory,
PrepareServlet prepare, int aliveTimeoutSecond, int readTimeoutSecond, int writeTimeoutSecond) {
PrepareServlet prepare, int aliveTimeoutSeconds, int readTimeoutSeconds, int writeTimeoutSeconds) {
super(serverStartTime, logger, executor, sslContext, bufferCapacity, bufferPool, responsePool,
maxbody, charset, address, resourceFactory, prepare, aliveTimeoutSecond, readTimeoutSecond, writeTimeoutSecond);
maxbody, charset, address, resourceFactory, prepare, aliveTimeoutSeconds, readTimeoutSeconds, writeTimeoutSeconds);
}
}

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

@@ -108,7 +108,7 @@ public class SncpServer extends Server<DLong, SncpContext, SncpRequest, SncpResp
ObjectPool<Response> responsePool = SncpResponse.createPool(createResponseCounter, cycleResponseCounter, this.responsePoolSize, null);
SncpContext sncpcontext = new SncpContext(this.serverStartTime, this.logger, executor, this.sslContext,
rcapacity, bufferPool, responsePool, this.maxbody, this.charset, this.address, this.resourceFactory,
this.prepare, this.aliveTimeoutSecond, this.readTimeoutSecond, this.writeTimeoutSecond);
this.prepare, this.aliveTimeoutSeconds, this.readTimeoutSeconds, this.writeTimeoutSeconds);
responsePool.setCreator((Object... params) -> new SncpResponse(sncpcontext, new SncpRequest(sncpcontext)));
return sncpcontext;
}