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

Browse Source
This commit is contained in:
kamhung
2015-11-23 16:58:00 +08:00
parent 0cbb4ed0e2
commit 2c0ad21ef3
12 changed files with 116 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -62,7 +62,7 @@
</properties>
</resources>
<!--
protocol: required server所启动的协议，有HTTP、SNCP， 目前只支持HTTP、SNCP。SNCP分TCP、UDP实现，默认使用TCP实现，UDP实现则使用SNCP.UDP值;
protocol: required server所启动的协议，有HTTP、SNCP， 目前只支持HTTP、SNCP。SNCP使用TCP实现;
host: 服务所占address ， 默认: 0.0.0.0
port: required 服务所占端口
root: 如果是web类型服务，则包含页面 默认:{APP_HOME}/root

2
src/com/wentch/redkale/boot/Application.java
View File

@@ -269,7 +269,7 @@ public final class Application {
for (AnyValue conf : resources.getAnyValues("group")) {
final String group = conf.getValue("name", "");
String protocol = conf.getValue("protocol", Sncp.DEFAULT_PROTOCOL).toUpperCase();
String protocol = conf.getValue("protocol", Transport.DEFAULT_PROTOCOL).toUpperCase();
if (!"TCP".equalsIgnoreCase(protocol) && !"UDP".equalsIgnoreCase(protocol)) {
throw new RuntimeException("Not supported Transport Protocol " + conf.getValue("protocol"));
}

2
src/com/wentch/redkale/boot/NodeServer.java
View File

@@ -47,7 +47,7 @@ public abstract class NodeServer {
private String sncpGroup = null; //当前Server的SNCP协议的组
private String nodeProtocol = Sncp.DEFAULT_PROTOCOL;
private String nodeProtocol = Transport.DEFAULT_PROTOCOL;
private InetSocketAddress sncpAddress; //HttpServer中的sncpAddress 为所属group对应的SncpServer, 为null表示只是单节点，没有分布式结构

8
src/com/wentch/redkale/boot/NodeSncpServer.java
View File

@@ -27,13 +27,7 @@ public final class NodeSncpServer extends NodeServer {
}
private static Server createServer(Application application, AnyValue serconf) {
String proto = serconf.getValue("protocol", "");
String subprotocol = Sncp.DEFAULT_PROTOCOL;
int pos = proto.indexOf('.');
if (pos > 0) {
subprotocol = proto.substring(pos + 1);
}
return new SncpServer(application.getStartTime(), subprotocol, application.getWatchFactory());
return new SncpServer(application.getStartTime(), application.getWatchFactory());
}
@Override

6
src/com/wentch/redkale/net/Transport.java
View File

@@ -22,6 +22,8 @@ import java.util.function.*;
*/
public final class Transport {
public static final String DEFAULT_PROTOCOL = "TCP";
protected static final int MAX_POOL_LIMIT = Runtime.getRuntime().availableProcessors() * 16;
protected final String name;
@@ -46,6 +48,10 @@ public final class Transport {
this(transport.name, transport.protocol, null, transport.bufferPoolSize, parse(localAddress, transports));
}
public Transport(String name, WatchFactory watch, int bufferPoolSize, Collection<InetSocketAddress> addresses) {
this(name, DEFAULT_PROTOCOL, watch, bufferPoolSize, addresses);
}
public Transport(String name, String protocol, WatchFactory watch, int bufferPoolSize, Collection<InetSocketAddress> addresses) {
this.name = name;
this.protocol = protocol;

153
src/com/wentch/redkale/net/sncp/Sncp.java
View File

@@ -38,8 +38,6 @@ public abstract class Sncp {
private static final java.lang.reflect.Type GROUPS_TYPE2 = new TypeToken<String[]>() {
}.getType();
public static final String DEFAULT_PROTOCOL = "TCP";
static final String LOCALPREFIX = "_DynLocal";
static final String REMOTEPREFIX = "_DynRemote";
@@ -138,75 +136,75 @@ public abstract class Sncp {
/**
* public class TestService implements Service{
*
* public String queryNode(){
* return "hello";
* }
* public String queryNode(){
* return "hello";
* }
*
* @MultiRun
* public String updateSomeThing(String id){
* return "hello" + id;
* }
* @MultiRun
* public String updateSomeThing(String id){
* return "hello" + id;
* }
*
* @MultiRun(selfrun = false)
* public void createSomeThing(TestBean bean){
* "xxxxx" + bean;
* }
* @MultiRun(selfrun = false)
* public void createSomeThing(TestBean bean){
* "xxxxx" + bean;
* }
* }
*
* public final class _DynLocalTestService extends TestService{
*
* @Resource
* private BsonConvert _convert;
* @Resource
* private BsonConvert _convert;
*
* private Transport[] _sameGroupTransports;
* private Transport[] _sameGroupTransports;
*
* private Transport[] _diffGroupTransports;
* private Transport[] _diffGroupTransports;
*
* private SncpClient _client;
*
* private String _selfstring;
* private SncpClient _client;
*
* @Override
* public final String name() {
* return "";
* }
* private String _selfstring;
*
* @Override
* public String toString() {
* return _selfstring == null ? super.toString() : _selfstring;
* }
* @Override
* public final String name() {
* return "";
* }
*
* @Override
* public String updateSomeThing(String id){
* return _updateSomeThing(true, true, true, id);
* }
* @Override
* public String toString() {
* return _selfstring == null ? super.toString() : _selfstring;
* }
*
* public String _updateSomeThing(boolean canselfrun, boolean cansamerun, boolean candiffrun, String id){
* String rs = super.updateSomeThing(id);
* if (_client== null) return;
* _client.remote(_convert, _sameGroupTransports, cansamerun, 0, true, false, false, id);
* _client.remote(_convert, _diffGroupTransports, candiffrun, 0, true, true, false, id);
* return rs;
* }
* @Override
* public String updateSomeThing(String id){
* return _updateSomeThing(true, true, true, id);
* }
*
* @Override
* public void createSomeThing(TestBean bean){
* _createSomeThing(false, true, true, bean);
* }
* public String _updateSomeThing(boolean canselfrun, boolean cansamerun, boolean candiffrun, String id){
* String rs = super.updateSomeThing(id);
* if (_client== null) return;
* _client.remote(_convert, _sameGroupTransports, cansamerun, 0, true, false, false, id);
* _client.remote(_convert, _diffGroupTransports, candiffrun, 0, true, true, false, id);
* return rs;
* }
*
* public void _createSomeThing(boolean canselfrun, boolean cansamerun, boolean candiffrun, TestBean bean){
* if(canselfrun) super.createSomeThing(bean);
* if (_client== null) return;
* _client.remote(_convert, _sameGroupTransports, cansamerun, 1, true, false, false, bean);
* _client.remote(_convert, _diffGroupTransports, candiffrun, 1, true, true, false, bean);
* }
* @Override
* public void createSomeThing(TestBean bean){
* _createSomeThing(false, true, true, bean);
* }
*
* public void _createSomeThing(boolean canselfrun, boolean cansamerun, boolean candiffrun, TestBean bean){
* if(canselfrun) super.createSomeThing(bean);
* if (_client== null) return;
* _client.remote(_convert, _sameGroupTransports, cansamerun, 1, true, false, false, bean);
* _client.remote(_convert, _diffGroupTransports, candiffrun, 1, true, true, false, bean);
* }
* }
*
* 创建Service的本地模式Class
* @param <T>
* @param name
* @param serviceClass
* @return
* @return
*/
@SuppressWarnings("unchecked")
public static <T extends Service> Class<? extends T> createLocalServiceClass(final String name, final Class<T> serviceClass) {
@@ -611,6 +609,7 @@ public abstract class Sncp {
/**
*
* 创建本地模式Service实例
*
* @param <T>
* @param name
* @param executor
@@ -619,7 +618,7 @@ public abstract class Sncp {
* @param groups
* @param sameGroupTransports
* @param diffGroupTransports
* @return
* @return
*/
@SuppressWarnings("unchecked")
public static <T extends Service> T createLocalService(final String name, final Consumer<Runnable> executor, final Class<T> serviceClass,
@@ -732,39 +731,39 @@ public abstract class Sncp {
/**
* public final class _DynRemoteTestService extends TestService{
*
* @Resource
* private BsonConvert _convert;
* @Resource
* private BsonConvert _convert;
*
* private Transport _transport;
* private Transport _transport;
*
* private SncpClient _client;
* private SncpClient _client;
*
* private String _selfstring;
* private String _selfstring;
*
* @Override
* public final String name() {
* return "";
* }
* @Override
* public final String name() {
* return "";
* }
*
* @Override
* public String toString() {
* return _selfstring == null ? super.toString() : _selfstring;
* }
* @Override
* public String toString() {
* return _selfstring == null ? super.toString() : _selfstring;
* }
*
* @Override
* public boolean testChange(TestBean bean) {
* return _client.remote(_convert, _transport, 0, bean);
* }
* @Override
* public boolean testChange(TestBean bean) {
* return _client.remote(_convert, _transport, 0, bean);
* }
*
* @Override
* public TestBean findTestBean(long id) {
* return _client.remote(_convert, _transport, 1, id);
* }
* @Override
* public TestBean findTestBean(long id) {
* return _client.remote(_convert, _transport, 1, id);
* }
*
* @Override
* public void runTestBean(long id, TestBean bean) {
* _client.remote(_convert, _transport, 2, id, bean);
* }
* @Override
* public void runTestBean(long id, TestBean bean) {
* _client.remote(_convert, _transport, 2, id, bean);
* }
* }
*
* 创建远程模式的Service实例
@@ -773,7 +772,7 @@ public abstract class Sncp {
* @param name
* @param executor
* @param serviceClass
* @param clientAddress
* @param clientAddress
* @param groups
* @param transport
* @return

91
src/com/wentch/redkale/net/sncp/SncpClient.java
View File

@@ -181,7 +181,7 @@ public final class SncpClient {
}
public <T> T remote(final BsonConvert convert, Transport transport, final int index, final Object... params) {
Future<byte[]> future = transport.isTCP() ? remoteTCP(convert, transport, actions[index], params) : remoteUDP(convert, transport, actions[index], params);
Future<byte[]> future = remote(convert, transport, actions[index], params);
try {
return convert.convertFrom(actions[index].resultTypes, future.get(5, TimeUnit.SECONDS));
} catch (InterruptedException | ExecutionException | TimeoutException e) {
@@ -193,11 +193,7 @@ public final class SncpClient {
public <T> void remote(final BsonConvert convert, Transport[] transports, boolean run, final int index, final Object... params) {
if (!run) return;
for (Transport transport : transports) {
if (transport.isTCP()) {
remoteTCP(convert, transport, actions[index], params);
} else {
remoteUDP(convert, transport, actions[index], params);
}
remote(convert, transport, actions[index], params);
}
}
@@ -206,94 +202,17 @@ public final class SncpClient {
if (executor != null) {
executor.accept(() -> {
for (Transport transport : transports) {
if (transport.isTCP()) {
remoteTCP(convert, transport, actions[index], params);
} else {
remoteUDP(convert, transport, actions[index], params);
}
remote(convert, transport, actions[index], params);
}
});
} else {
for (Transport transport : transports) {
if (transport.isTCP()) {
remoteTCP(convert, transport, actions[index], params);
} else {
remoteUDP(convert, transport, actions[index], params);
}
remote(convert, transport, actions[index], params);
}
}
}
private Future<byte[]> remoteUDP(final BsonConvert convert, final Transport transport, final SncpAction action, final Object... params) {
Type[] myparamtypes = action.paramTypes;
final Supplier<ByteBuffer> supplier = transport.getBufferSupplier();
final BsonWriter bw = convert.pollBsonWriter(() -> supplier.get().put(DEFAULT_HEADER)); // 将head写入
for (int i = 0; i < params.length; i++) {
convert.convertTo(bw, myparamtypes[i], params[i]);
}
final SocketAddress addr = action.addressParamIndex >= 0 ? (SocketAddress) params[action.addressParamIndex] : null;
final AsyncConnection conn = transport.pollConnection(addr);
if (conn == null || !conn.isOpen()) {
logger.log(Level.SEVERE, action.method + " sncp (params: " + jsonConvert.convertTo(params) + ") cannot connect " + (conn == null ? addr : conn.getRemoteAddress()));
throw new RuntimeException("sncp " + (conn == null ? addr : conn.getRemoteAddress()) + " cannot connect");
}
final int reqBodyLength = bw.count(); //body总长度
final long seqid = System.nanoTime();
final DLong actionid = action.actionid;
final int readto = conn.getReadTimeoutSecond();
final int writeto = conn.getWriteTimeoutSecond();
final ByteBuffer buffer = transport.pollBuffer();
try {
//------------------------------ 发送请求 ---------------------------------------------------
int pos = 0;
for (ByteBuffer buf : bw.toBuffers()) {
int len = buf.remaining() - HEADER_SIZE;
fillHeader(buf, seqid, actionid, reqBodyLength, pos, len);
pos += len;
Thread.sleep(20);
conn.write(buf).get(writeto > 0 ? writeto : 3, TimeUnit.SECONDS);
transport.offerBuffer(buf);
}
//------------------------------ 接收响应 ---------------------------------------------------
int received = 0;
int respBodyLength = 1;
byte[] respBody = null;
while (received < respBodyLength) {
buffer.clear();
conn.read(buffer).get(readto > 0 ? readto : 3, TimeUnit.SECONDS);
buffer.flip();
checkResult(seqid, action, buffer);
int respbodylen = buffer.getInt();
if (respBody == null) {
respBodyLength = respbodylen;
respBody = new byte[respBodyLength];
}
int bodyOffset = buffer.getInt(); //
int frameLength = buffer.getInt(); //
final int retcode = buffer.getInt();
if (retcode != 0) {
logger.log(Level.SEVERE, action.method + " sncp (params: " + jsonConvert.convertTo(params) + ") deal error (retcode=" + retcode + ", retinfo=" + SncpResponse.getRetCodeInfo(retcode) + ")");
throw new RuntimeException("remote service(" + action.method + ") deal error (retcode=" + retcode + ", retinfo=" + SncpResponse.getRetCodeInfo(retcode) + ")");
}
int len = Math.min(buffer.remaining(), frameLength);
buffer.get(respBody, bodyOffset, len);
received += len;
}
return new SncpFuture<>(respBody);
} catch (RuntimeException e) {
throw e;
} catch (Exception ex) {
logger.log(Level.SEVERE, action.method + " sncp (params: " + jsonConvert.convertTo(params) + ") udp remote error", ex);
throw new RuntimeException(ex);
} finally {
transport.offerBuffer(buffer);
transport.offerConnection(true, conn);
}
}
private Future<byte[]> remoteTCP(final BsonConvert convert, final Transport transport, final SncpAction action, final Object... params) {
private Future<byte[]> remote(final BsonConvert convert, final Transport transport, final SncpAction action, final Object... params) {
Type[] myparamtypes = action.paramTypes;
final BsonWriter writer = convert.pollBsonWriter(transport.getBufferSupplier()); // 将head写入
writer.writeTo(DEFAULT_HEADER);

56
src/com/wentch/redkale/net/sncp/SncpContext.java
View File

@@ -12,7 +12,6 @@ import com.wentch.redkale.watch.*;
import java.net.*;
import java.nio.*;
import java.nio.charset.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.*;
@@ -22,35 +21,6 @@ import java.util.logging.*;
*/
public final class SncpContext extends Context {
protected static class RequestEntry {
protected final long seqid;
protected final byte[] body;
protected final long time = System.currentTimeMillis();
private int received;
public RequestEntry(long seqid, byte[] body) {
this.seqid = seqid;
this.body = body;
}
public void add(ByteBuffer buffer, int pos) {
int len = Math.min(buffer.remaining(), this.body.length - this.received);
this.received += len;
buffer.get(body, pos, len);
}
public boolean isCompleted() {
return this.body.length <= this.received;
}
}
private final ConcurrentHashMap<Long, RequestEntry> requests = new ConcurrentHashMap<>();
protected final BsonFactory bsonFactory;
public SncpContext(long serverStartTime, Logger logger, ExecutorService executor, int bufferCapacity, ObjectPool<ByteBuffer> bufferPool,
@@ -61,32 +31,6 @@ public final class SncpContext extends Context {
this.bsonFactory = BsonFactory.root();
}
protected RequestEntry addRequestEntity(long seqid, byte[] bodys) {
RequestEntry entry = new RequestEntry(seqid, bodys);
requests.put(seqid, entry);
return entry;
}
protected void expireRequestEntry(long milliSecond) {
if (requests.size() < 32) return;
List<Long> seqids = new ArrayList<>();
long t = System.currentTimeMillis() - milliSecond;
requests.forEach((x, y) -> {
if (y.time < t) seqids.add(x);
});
for (long seqid : seqids) {
requests.remove(seqid);
}
}
protected void removeRequestEntity(long seqid) {
requests.remove(seqid);
}
protected RequestEntry getRequestEntity(long seqid) {
return requests.get(seqid);
}
protected WatchFactory getWatchFactory() {
return watch;
}

32
src/com/wentch/redkale/net/sncp/SncpDynServlet.java
View File

@@ -85,14 +85,8 @@ public final class SncpDynServlet extends SncpServlet {
@Override
public void execute(SncpRequest request, SncpResponse response) throws IOException {
final boolean tcp = request.isTCP();
if (bufferSupplier == null) {
if (tcp) {
bufferSupplier = request.getContext().getBufferSupplier();
} else { //UDP 需要分包
final Supplier<ByteBuffer> supplier = request.getContext().getBufferSupplier();
bufferSupplier = () -> supplier.get().put(DEFAULT_HEADER);
}
bufferSupplier = request.getContext().getBufferSupplier();
}
SncpServletAction action = actions.get(request.getActionid());
//if (finest) logger.log(Level.FINEST, "sncpdyn.execute: " + request + ", " + (action == null ? "null" : action.method));
@@ -100,7 +94,7 @@ public final class SncpDynServlet extends SncpServlet {
response.finish(SncpResponse.RETCODE_ILLACTIONID, null); //无效actionid
} else {
BsonWriter out = action.convert.pollBsonWriter(bufferSupplier);
if (tcp) out.writeTo(DEFAULT_HEADER);
out.writeTo(DEFAULT_HEADER);
BsonReader in = action.convert.pollBsonReader();
try {
in.setBytes(request.getBody());
@@ -130,23 +124,23 @@ public final class SncpDynServlet extends SncpServlet {
/*
*
* public class TestService implements Service {
* public boolean change(TestBean bean, String name, int id) {
* public boolean change(TestBean bean, String name, int id) {
*
* }
* }
* }
*
* public class DynActionTestService_change extends SncpServletAction {
*
* public TestService service;
* public TestService service;
*
* @Override
* public void action(final BsonReader in, final BsonWriter out) throws Throwable {
* TestBean arg1 = convert.convertFrom(in, paramTypes[1]);
* String arg2 = convert.convertFrom(in, paramTypes[2]);
* int arg3 = convert.convertFrom(in, paramTypes[3]);
* Object rs = service.change(arg1, arg2, arg3);
* convert.convertTo(out, paramTypes[0], rs);
* }
* @Override
* public void action(final BsonReader in, final BsonWriter out) throws Throwable {
* TestBean arg1 = convert.convertFrom(in, paramTypes[1]);
* String arg2 = convert.convertFrom(in, paramTypes[2]);
* int arg3 = convert.convertFrom(in, paramTypes[3]);
* Object rs = service.change(arg1, arg2, arg3);
* convert.convertTo(out, paramTypes[0], rs);
* }
* }
*/
/**

45
src/com/wentch/redkale/net/sncp/SncpRequest.java
View File

@@ -7,7 +7,6 @@ package com.wentch.redkale.net.sncp;
import com.wentch.redkale.convert.bson.*;
import com.wentch.redkale.net.*;
import com.wentch.redkale.net.sncp.SncpContext.RequestEntry;
import com.wentch.redkale.util.*;
import java.net.*;
import java.nio.*;
@@ -74,41 +73,15 @@ public final class SncpRequest extends Request {
return -1;
}
//---------------------body----------------------------------
if (this.channel.isTCP()) { // TCP模式， 不管数据包大小 只传一帧数据
this.body = new byte[this.bodylength];
int len = Math.min(this.bodylength, buffer.remaining());
buffer.get(body, 0, len);
this.bodyoffset = len;
return bodylength - len;
}
//--------------------- UDP 模式 ----------------------------
if (this.bodylength == this.framelength) { //只有一帧的数据
if (this.framelength > buffer.remaining()) { //缺失一部分数据
throw new RuntimeException(SncpRequest.class.getSimpleName() + " data need " + this.framelength + " bytes, but only " + buffer.remaining() + " bytes");
}
this.body = new byte[this.framelength];
buffer.get(body);
return 0;
}
//多帧数据
final SncpContext scontext = (SncpContext) this.context;
RequestEntry entry = scontext.getRequestEntity(this.seqid);
if (entry == null) entry = scontext.addRequestEntity(this.seqid, new byte[this.bodylength]);
entry.add(buffer, this.bodyoffset);
if (entry.isCompleted()) { //数据读取完毕
this.body = entry.body;
scontext.removeRequestEntity(this.seqid);
return 0;
} else {
scontext.expireRequestEntry(10 * 1000); //10秒过期
}
if (this.channel.isTCP()) return this.bodylength - this.framelength;
return Integer.MIN_VALUE; //多帧数据返回 Integer.MIN_VALUE
this.body = new byte[this.bodylength];
int len = Math.min(this.bodylength, buffer.remaining());
buffer.get(body, 0, len);
this.bodyoffset = len;
return bodylength - len;
}
@Override
protected int readBody(ByteBuffer buffer) { // 只有 TCP 模式会调用此方法
protected int readBody(ByteBuffer buffer) {
final int framelen = buffer.remaining();
buffer.get(this.body, this.bodyoffset, framelen);
this.bodyoffset += framelen;
@@ -117,11 +90,7 @@ public final class SncpRequest extends Request {
@Override
protected void prepare() {
this.keepAlive = this.channel.isTCP();
}
protected boolean isTCP() {
return this.channel.isTCP();
this.keepAlive = true;
}
@Override

11
src/com/wentch/redkale/net/sncp/SncpResponse.java
View File

@@ -57,16 +57,7 @@ public final class SncpResponse extends Response<SncpRequest> {
}
final int respBodyLength = out.count(); //body总长度
final ByteBuffer[] buffers = out.toBuffers();
if (this.channel.isTCP()) { //TCP模式 TCP的总长度需要减去第一个buffer的header长度
fillHeader(buffers[0], respBodyLength - HEADER_SIZE, 0, respBodyLength - HEADER_SIZE, retcode);
} else {
int pos = 0;
for (ByteBuffer buffer : buffers) {
int len = buffer.remaining() - HEADER_SIZE;
fillHeader(buffer, respBodyLength, pos, len, retcode);
pos += len;
}
}
fillHeader(buffers[0], respBodyLength - HEADER_SIZE, 0, respBodyLength - HEADER_SIZE, retcode);
finish(buffers);
}

8
src/com/wentch/redkale/net/sncp/SncpServer.java
View File

@@ -20,12 +20,12 @@ import java.util.concurrent.atomic.*;
*/
public final class SncpServer extends Server {
public SncpServer(String protocol) {
this(System.currentTimeMillis(), protocol, null);
public SncpServer() {
this(System.currentTimeMillis(), null);
}
public SncpServer(long serverStartTime, String protocol, final WatchFactory watch) {
super(serverStartTime, protocol, new SncpPrepareServlet(), watch);
public SncpServer(long serverStartTime, final WatchFactory watch) {
super(serverStartTime, "TCP", new SncpPrepareServlet(), watch);
}
@Override