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

Browse Source
This commit is contained in:
RedKale
2016-04-15 11:06:03 +08:00
parent 120334aaee
commit f184376f95
3 changed files with 65 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -73,7 +73,8 @@
</properties>
</resources>
<!--
protocol: required server所启动的协议，有HTTP、SNCP， 目前只支持HTTP、SNCP。SNCP使用TCP实现;
protocol: required server所启动的协议，Redkale内置的有HTTP、SNCP，SNCP使用TCP实现;
name: 服务的名称，用于监控识别，一个配置文件中的server.name不能重复,命名规则: 字母、数字、下划线、减号
host: 服务所占address ， 默认: 0.0.0.0
port: required 服务所占端口
root: 如果是web类型服务，则包含页面 默认:{APP_HOME}/root

31
src/org/redkale/boot/NodeServer.java
View File

@@ -28,6 +28,7 @@ import org.redkale.util.AnyValue.DefaultAnyValue;
import org.redkale.util.*;
/**
* Server节点的初始化配置类
*
* <p>
* 详情见: http://redkale.org
@@ -58,18 +59,25 @@ public abstract class NodeServer {
//当前Server对象
protected final Server server;
private String sncpGroup = null; //当前Server的SNCP协议的组
private InetSocketAddress sncpAddress; //SNCP服务的地址， 非SNCP为null
//当前Server的SNCP协议的组
private String sncpGroup = null;
//SNCP服务的地址， 非SNCP为null
private InetSocketAddress sncpAddress;
//加载Service时的处理函数
protected Consumer<ServiceWrapper> consumer;
//server节点的配置
protected AnyValue serverConf;
//加载server节点后的拦截器
protected NodeInterceptor interceptor;
//本地模式的Service对象集合
protected final Set<ServiceWrapper> localServiceWrappers = new LinkedHashSet<>();
//远程模式的Service对象集合
protected final Set<ServiceWrapper> remoteServiceWrappers = new LinkedHashSet<>();
public NodeServer(Application application, Server server) {
@@ -121,8 +129,8 @@ public abstract class NodeServer {
//单向SNCP服务不需要对等group
//if (this.sncpGroup == null) throw new RuntimeException("Server (" + String.valueOf(config).replaceAll("\\s+", " ") + ") not found <group> info");
}
if (this.sncpAddress != null) this.resourceFactory.register(RESNAME_SERVER_ADDR, this.sncpAddress); //单点服务不会有 sncpAddress、sncpGroup
//单点服务不会有 sncpAddress、sncpGroup
if (this.sncpAddress != null) this.resourceFactory.register(RESNAME_SERVER_ADDR, this.sncpAddress);
if (this.sncpGroup != null) this.resourceFactory.register(RESNAME_SERVER_GROUP, this.sncpGroup);
{
//设置root文件夹
@@ -137,9 +145,11 @@ public abstract class NodeServer {
resourceFactory.register(Server.RESNAME_SERVER_ROOT, Path.class, myroot.toPath());
final String homepath = myroot.getCanonicalPath();
//加入指定的classpath
Server.loadLib(logger, config.getValue("lib", "").replace("${APP_HOME}", homepath) + ";" + homepath + "/lib/*;" + homepath + "/classes");
if (server != null) server.init(config);
}
//必须要进行初始化， 构建Service时需要使用Context中的ExecutorService
server.init(config);
initResource(); //给 DataSource、CacheSource 注册依赖注入时的监听回调事件。
String interceptorClass = config.getValue("nodeInterceptor", "");
@@ -147,6 +157,7 @@ public abstract class NodeServer {
Class clazz = forName(interceptorClass);
this.interceptor = (NodeInterceptor) clazz.newInstance();
}
ClassFilter<Servlet> servletFilter = createServletClassFilter();
ClassFilter<Service> serviceFilter = createServiceClassFilter();
long s = System.currentTimeMillis();

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

@@ -6,16 +6,16 @@
package org.redkale.net;
import java.io.*;
import java.lang.reflect.*;
import java.lang.reflect.Method;
import java.net.*;
import java.nio.charset.*;
import java.nio.charset.Charset;
import java.text.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.logging.*;
import org.redkale.util.*;
import org.redkale.watch.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;
import org.redkale.util.AnyValue;
import org.redkale.watch.WatchFactory;
/**
*
@@ -23,6 +23,11 @@ import org.redkale.watch.*;
* 详情见: http://redkale.org
*
* @author zhangjx
* @param <K> 请求ID的数据类型， 例如HTTP协议请求标识为url，请求ID的数据类型就是String
* @param <C> Context
* @param <R> Request
* @param <P> Response
* @param <S> Servlet
*/
public abstract class Server<K extends Serializable, C extends Context, R extends Request<C>, P extends Response<C, R>, S extends Servlet<C, R, P>> {
@@ -31,44 +36,63 @@ public abstract class Server<K extends Serializable, C extends Context, R extend
protected final Logger logger = Logger.getLogger(this.getClass().getSimpleName());
//-------------------------------------------------------------
//服务的启动时间
protected final long serverStartTime;
//监控对象
protected final WatchFactory watch;
//服务的名称
protected String name;
//应用层协议名
protected final String protocol;
//服务的根Servlet
protected final PrepareServlet<K, C, R, P, S> prepare;
//服务的上下文对象
protected C context;
//服务的配置信息
protected AnyValue config;
//服务数据的编解码，null视为UTF-8
protected Charset charset;
//服务的监听端口
protected InetSocketAddress address;
//连接队列大小
protected int backlog;
//传输层协议的服务
protected ProtocolServer serverChannel;
//ByteBuffer的容量大小
protected int bufferCapacity;
//线程数
protected int threads;
//线程池
protected ExecutorService executor;
//ByteBuffer池大小
protected int bufferPoolSize;
//Response池大小
protected int responsePoolSize;
//请求包大小的上限，单位:字节
protected int maxbody;
//IO读取的超时秒数，小于1视为不设置
protected int readTimeoutSecond;
//IO写入 的超时秒数，小于1视为不设置
protected int writeTimeoutSecond;
private ScheduledThreadPoolExecutor scheduler;
protected Server(long serverStartTime, String protocol, PrepareServlet<K, C, R, P, S> servlet, final WatchFactory watch) {
this.serverStartTime = serverStartTime;
this.protocol = protocol;
@@ -89,25 +113,30 @@ public abstract class Server<K extends Serializable, C extends Context, R extend
this.threads = config.getIntValue("threads", Runtime.getRuntime().availableProcessors() * 16);
this.bufferPoolSize = config.getIntValue("bufferPoolSize", Runtime.getRuntime().availableProcessors() * 512);
this.responsePoolSize = config.getIntValue("responsePoolSize", Runtime.getRuntime().availableProcessors() * 256);
final int port = this.address.getPort();
this.name = config.getValue("name", "Server-" + protocol + "-" + this.address.getPort());
if (!this.name.matches("^[a-zA-Z][\\w_-]{1,64}$")) throw new RuntimeException("server.name (" + this.name + ") is illegal");
final AtomicInteger counter = new AtomicInteger();
final Format f = createFormat();
final String n = name;
this.executor = Executors.newFixedThreadPool(threads, (Runnable r) -> {
Thread t = new WorkThread(executor, r);
t.setName("Servlet-" + protocol + "-" + port + "-Thread-" + f.format(counter.incrementAndGet()));
t.setName(n + "-ServletThread-" + f.format(counter.incrementAndGet()));
return t;
});
}
public void destroy(final AnyValue config) throws Exception {
this.prepare.destroy(context, config);
if (scheduler != null) scheduler.shutdownNow();
}
public InetSocketAddress getSocketAddress() {
return address;
}
public String getName() {
return name;
}
public String getProtocol() {
return protocol;
}
@@ -134,8 +163,8 @@ public abstract class Server<K extends Serializable, C extends Context, R extend
serverChannel.accept();
final String threadName = "[" + Thread.currentThread().getName() + "] ";
logger.info(threadName + this.getClass().getSimpleName() + ("TCP".equalsIgnoreCase(protocol) ? "" : ("." + protocol)) + " listen: " + address
+ ", threads: " + threads + ", bufferCapacity: " + bufferCapacity + ", bufferPoolSize: " + bufferPoolSize + ", responsePoolSize: " + responsePoolSize
+ ", started in " + (System.currentTimeMillis() - context.getServerStartTime()) + " ms");
+ ", threads: " + threads + ", bufferCapacity: " + bufferCapacity + ", bufferPoolSize: " + bufferPoolSize + ", responsePoolSize: " + responsePoolSize
+ ", started in " + (System.currentTimeMillis() - context.getServerStartTime()) + " ms");
}
protected abstract C createContext();