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

Browse Source
This commit is contained in:
wentch
2015-12-16 15:41:49 +08:00
parent 4677fc6fa3
commit 6e396fa2e0
9 changed files with 88 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
src/org/redkale/net/sncp/Sncp.java
View File

@@ -9,6 +9,7 @@ import org.redkale.net.sncp.SncpClient.SncpAction;
import java.lang.annotation.*;
import java.lang.reflect.*;
import java.net.*;
import java.security.*;
import java.util.*;
import java.util.function.*;
import javax.annotation.*;
@@ -46,22 +47,16 @@ public abstract class Sncp {
static final String REMOTEPREFIX = "_DynRemote";
private static final byte[] hashes = new byte[255];
private static final MessageDigest md5;
static { //64进制
//0-9:48-57 A-Z:65-90 a-z:97-122 $:36 _:95
byte index = 0;
hashes['_'] = index++;
hashes['$'] = index++;
for (int i = '0'; i <= '9'; i++) {
hashes[i] = index++;
}
for (int i = 'A'; i <= 'Z'; i++) {
hashes[i] = index++;
}
for (int i = 'a'; i <= 'z'; i++) {
hashes[i] = index++;
MessageDigest d = null;
try {
d = MessageDigest.getInstance("MD5");
} catch (NoSuchAlgorithmException ex) {
ex.printStackTrace();
}
md5 = d;
}
private Sncp() {
@@ -72,37 +67,19 @@ public abstract class Sncp {
return ((0L + ip.getPort()) << 32) | ((0xffffffff & bytes[0]) << 24) | ((0xffffff & bytes[1]) << 16) | ((0xffff & bytes[2]) << 8) | (0xff & bytes[3]);
}
public static long hash(final Class clazz) {
if (clazz == null) return Long.MIN_VALUE;
long rs = hash(clazz.getSimpleName());
return (rs < Integer.MAX_VALUE) ? rs | 0xF00000000L : rs;
public static DLong hash(final Class clazz) {
if (clazz == null) return DLong.ZERO;
return hash(clazz.getName());
}
public static long hashClass(final String clazzName) {
if (clazzName == null || clazzName.isEmpty()) return Long.MIN_VALUE;
long rs = hash(clazzName.substring(clazzName.lastIndexOf('.') + 1));
return (rs < Integer.MAX_VALUE) ? rs | 0xF00000000L : rs;
public static DLong hashClass(final String clazzName) {
if (clazzName == null || clazzName.isEmpty()) return DLong.ZERO;
return hash(clazzName);
}
public static DLong hash(final java.lang.reflect.Method method) {
if (method == null) return new DLong(-1, -1);
long rs1 = hash(method.getName());
rs1 = (rs1 < Integer.MAX_VALUE) ? rs1 | 0xF00000000L : rs1;
final Class[] params = method.getParameterTypes();
final StringBuilder sb = new StringBuilder();
if (params.length < 1) {
sb.append("00");
} else {
sb.append(params.length);
for (Class clzz : params) {
String s = clzz.getSimpleName();
sb.append(s.substring(0, s.length() > 1 ? 2 : 1)).append(s.substring(s.length() - 1));
}
}
long rs2 = hash(sb.toString());
rs2 = (rs2 < Integer.MAX_VALUE) ? rs2 | 0xF00000000L : rs2;
return new DLong(rs1, rs2);
if (method == null) return DLong.ZERO;
return hash(method.toString());
}
/**
@@ -111,25 +88,13 @@ public abstract class Sncp {
* @param name
* @return
*/
public static long hash(final String name) {
if (name == null) return Long.MIN_VALUE;
if (name.isEmpty()) return 0;
final char[] chars = Utility.charArray(name);
if (chars.length <= 11) {
long rs = 0;
for (int i = 0; i < chars.length; i++) {
rs = (rs << 6) | hashes[0xff & chars[i]];
}
return rs;
public static DLong hash(final String name) {
if (name == null || name.isEmpty()) return DLong.ZERO;
byte[] bytes = name.trim().getBytes();
synchronized (md5) {
bytes = md5.digest(bytes);
}
String len = Integer.toString(chars.length, 36);
long rs = len.length() > 1 ? hashes[0xff & len.charAt(0)] : hashes[0xff & '0'];
rs = (rs << 6) | hashes[0xff & len.charAt(len.length() - 1)]; //前2位用于存放长度
final int step = (chars.length - 1) / 9 + 1;
for (int i = 0; i < chars.length; i += step) {
rs = (rs << 6) | hashes[0xff & chars[i]];
}
return rs;
return new DLong(bytes);
}
public static boolean isRemote(Service service) {
@@ -748,7 +713,7 @@ public abstract class Sncp {
}
{
StringBuilder sb = new StringBuilder();
sb.append(newClazz.getName()).append("{name = ").append(name);
sb.append(newClazz.getName()).append("{name = '").append(name).append("'");
if (client != null) {
sb.append(", nameid = ").append(client.getNameid()).append(", serviceid = ").append(client.getServiceid());
sb.append(", action.size = ").append(client.getActionCount());

45
src/org/redkale/net/sncp/SncpClient.java
View File

@@ -109,17 +109,16 @@ public final class SncpClient {
private final int addrPort;
protected final long nameid;
protected final DLong serviceid;
protected final long serviceid;
protected final DLong nameid;
protected final SncpAction[] actions;
protected final Consumer<Runnable> executor;
public SncpClient(final String serviceName, final Consumer<Runnable> executor, final long serviceid, boolean remote, final Class serviceClass,
boolean onlySncpDyn, final InetSocketAddress clientAddress, final HashSet<String> groups) { // 以下划线_开头的serviceName只能是被系统分配, 且长度可以超过11位
if (serviceName.length() > 10 && serviceName.charAt(0) != '_') throw new RuntimeException(serviceClass + " @Resource name(" + serviceName + ") too long , must less 11");
public SncpClient(final String serviceName, final Consumer<Runnable> executor, final DLong serviceid, boolean remote, final Class serviceClass,
boolean onlySncpDyn, final InetSocketAddress clientAddress, final HashSet<String> groups) {
this.remote = remote;
this.executor = executor;
this.serviceClass = serviceClass;
@@ -140,11 +139,11 @@ public final class SncpClient {
this.addrPort = clientAddress == null ? 0 : clientAddress.getPort();
}
public long getNameid() {
public DLong getNameid() {
return nameid;
}
public long getServiceid() {
public DLong getServiceid() {
return serviceid;
}
@@ -259,7 +258,7 @@ public final class SncpClient {
throw new RuntimeException("sncp " + (conn == null ? addr : conn.getRemoteAddress()) + " cannot connect");
}
final ByteBuffer[] sendBuffers = writer.toBuffers();
fillHeader(sendBuffers[0], seqid, actionid, reqBodyLength, 0, reqBodyLength);
fillHeader(sendBuffers[0], seqid, actionid, reqBodyLength);
final ByteBuffer buffer = transport.pollBuffer();
final SncpFuture<byte[]> future = new SncpFuture();
@@ -321,8 +320,6 @@ public final class SncpClient {
checkResult(seqid, action, buffer);
final int respBodyLength = buffer.getInt();
buffer.getInt(); // bodyOffset
buffer.getInt(); // frameLength
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) + ")");
@@ -372,34 +369,28 @@ public final class SncpClient {
long rseqid = buffer.getLong();
if (rseqid != seqid) throw new RuntimeException("sncp(" + action.method + ") response.seqid = " + seqid + ", but request.seqid =" + rseqid);
if (buffer.getChar() != HEADER_SIZE) throw new RuntimeException("sncp(" + action.method + ") buffer receive header.length not " + HEADER_SIZE);
long rserviceid = buffer.getLong();
if (rserviceid != serviceid) throw new RuntimeException("sncp(" + action.method + ") response.serviceid = " + serviceid + ", but request.serviceid =" + rserviceid);
long rnameid = buffer.getLong();
if (rnameid != nameid) throw new RuntimeException("sncp(" + action.method + ") response.nameid = " + nameid + ", but receive nameid =" + rnameid);
byte[] bs = new byte[16];
buffer.get(bs);
if (!action.actionid.equals(bs)) throw new RuntimeException("sncp(" + action.method + ") response.actionid = " + action.actionid + ", but request.actionid =(" + Utility.binToHexString(bs) + ")");
DLong rserviceid = DLong.read(buffer);
if (!rserviceid.equals(serviceid)) throw new RuntimeException("sncp(" + action.method + ") response.serviceid = " + serviceid + ", but request.serviceid =" + rserviceid);
DLong rnameid = DLong.read(buffer);
if (!rnameid.equals(nameid)) throw new RuntimeException("sncp(" + action.method + ") response.nameid = " + nameid + ", but receive nameid =" + rnameid);
DLong raction = DLong.read(buffer);
if (!action.actionid.equals(raction)) throw new RuntimeException("sncp(" + action.method + ") response.actionid = " + action.actionid + ", but request.actionid =(" + raction + ")");
buffer.getInt(); //地址
buffer.getChar(); //端口
}
private void fillHeader(ByteBuffer buffer, long seqid, DLong actionid, int bodyLength, int bodyOffset, int frameLength) {
private void fillHeader(ByteBuffer buffer, long seqid, DLong actionid, int bodyLength) {
//---------------------head----------------------------------
final int currentpos = buffer.position();
buffer.position(0);
buffer.putLong(seqid); //序列号
buffer.putChar((char) HEADER_SIZE); //header长度
buffer.putLong(this.serviceid);
buffer.putLong(this.nameid);
actionid.putTo(buffer);
buffer.put(addrBytes[0]);
buffer.put(addrBytes[1]);
buffer.put(addrBytes[2]);
buffer.put(addrBytes[3]);
DLong.write(buffer, this.serviceid);
DLong.write(buffer, this.nameid);
DLong.write(buffer, actionid);
buffer.put(addrBytes);
buffer.putChar((char) this.addrPort);
buffer.putInt(bodyLength); //body长度
buffer.putInt(bodyOffset);
buffer.putInt(frameLength); //一帧数据的长度
buffer.putInt(0); //结果码， 请求方固定传0
buffer.position(currentpos);
}

8
src/org/redkale/net/sncp/SncpDynServlet.java
View File

@@ -36,9 +36,9 @@ public final class SncpDynServlet extends SncpServlet {
private final String serviceName;
private final long nameid;
private final DLong serviceid;
private final long serviceid;
private final DLong nameid;
private final HashMap<DLong, SncpServletAction> actions = new HashMap<>();
@@ -76,12 +76,12 @@ public final class SncpDynServlet extends SncpServlet {
}
@Override
public long getNameid() {
public DLong getNameid() {
return nameid;
}
@Override
public long getServiceid() {
public DLong getServiceid() {
return serviceid;
}

17
src/org/redkale/net/sncp/SncpPrepareServlet.java
View File

@@ -11,6 +11,7 @@ import org.redkale.util.AnyValue;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import org.redkale.util.*;
/**
*
@@ -21,18 +22,18 @@ public class SncpPrepareServlet extends PrepareServlet<SncpRequest, SncpResponse
private static final ByteBuffer pongBuffer = ByteBuffer.wrap("PONG".getBytes()).asReadOnlyBuffer();
private final Map<Long, Map<Long, SncpServlet>> maps = new HashMap<>();
private final Map<DLong, Map<DLong, SncpServlet>> maps = new HashMap<>();
private final Map<Long, SncpServlet> singlemaps = new HashMap<>();
private final Map<DLong, SncpServlet> singlemaps = new HashMap<>();
public void addSncpServlet(SncpServlet servlet) {
if (servlet.getNameid() == 0) {
if (servlet.getNameid() == DLong.ZERO) {
synchronized (singlemaps) {
singlemaps.put(servlet.getServiceid(), servlet);
}
} else {
synchronized (maps) {
Map<Long, SncpServlet> m = maps.get(servlet.getServiceid());
Map<DLong, SncpServlet> m = maps.get(servlet.getServiceid());
if (m == null) {
m = new HashMap<>();
maps.put(servlet.getServiceid(), m);
@@ -50,7 +51,7 @@ public class SncpPrepareServlet extends PrepareServlet<SncpRequest, SncpResponse
@Override
public void init(Context context, AnyValue config) {
Collection<Map<Long, SncpServlet>> values = this.maps.values();
Collection<Map<DLong, SncpServlet>> values = this.maps.values();
values.stream().forEach((en) -> {
en.values().stream().forEach(s -> s.init(context, s.conf));
});
@@ -58,7 +59,7 @@ public class SncpPrepareServlet extends PrepareServlet<SncpRequest, SncpResponse
@Override
public void destroy(Context context, AnyValue config) {
Collection<Map<Long, SncpServlet>> values = this.maps.values();
Collection<Map<DLong, SncpServlet>> values = this.maps.values();
values.stream().forEach((en) -> {
en.values().stream().forEach(s -> s.destroy(context, s.conf));
});
@@ -71,14 +72,14 @@ public class SncpPrepareServlet extends PrepareServlet<SncpRequest, SncpResponse
return;
}
SncpServlet servlet;
if (request.getNameid() == 0) {
if (request.getNameid() == DLong.ZERO) {
servlet = singlemaps.get(request.getServiceid());
if (servlet == null) {
response.finish(SncpResponse.RETCODE_ILLSERVICEID, null); //无效serviceid
return;
}
} else {
Map<Long, SncpServlet> m = maps.get(request.getServiceid());
Map<DLong, SncpServlet> m = maps.get(request.getServiceid());
if (m == null) {
response.finish(SncpResponse.RETCODE_ILLSERVICEID, null); //无效serviceid
return;

24
src/org/redkale/net/sncp/SncpRequest.java
View File

@@ -18,7 +18,7 @@ import org.redkale.util.*;
*/
public final class SncpRequest extends Request {
public static final int HEADER_SIZE = 64;
public static final int HEADER_SIZE = 72;
public static final byte[] DEFAULT_HEADER = new byte[HEADER_SIZE];
@@ -26,9 +26,9 @@ public final class SncpRequest extends Request {
private long seqid;
private long nameid;
private DLong serviceid;
private long serviceid;
private DLong nameid;
private DLong actionid;
@@ -36,8 +36,6 @@ public final class SncpRequest extends Request {
private int bodyoffset;
private int framelength;
private boolean ping;
private byte[] body;
@@ -61,15 +59,13 @@ public final class SncpRequest extends Request {
context.getLogger().finest("sncp buffer header.length not " + HEADER_SIZE);
return -1;
}
this.serviceid = buffer.getLong();
this.nameid = buffer.getLong();
this.serviceid = DLong.read(buffer);
this.nameid = DLong.read(buffer);
byte[] bs = new byte[16];
buffer.get(bs);
this.actionid = new DLong(bs);
buffer.get(bufferbytes);
this.bodylength = buffer.getInt();
this.bodyoffset = buffer.getInt();
this.framelength = buffer.getInt();
if (buffer.getInt() != 0) {
context.getLogger().finest("sncp buffer header.retcode not 0");
@@ -101,14 +97,14 @@ public final class SncpRequest extends Request {
return SncpRequest.class.getSimpleName() + "{seqid=" + this.seqid
+ ",serviceid=" + this.serviceid + ",actionid=" + this.actionid
+ ",bodylength=" + this.bodylength + ",bodyoffset=" + this.bodyoffset
+ ",framelength=" + this.framelength + ",remoteAddress=" + getRemoteAddress() + "}";
+ ",remoteAddress=" + getRemoteAddress() + "}";
}
@Override
protected void recycle() {
this.seqid = 0;
this.framelength = 0;
this.serviceid = 0;
this.serviceid = null;
this.nameid = null;
this.actionid = null;
this.bodylength = 0;
this.bodyoffset = 0;
@@ -130,11 +126,11 @@ public final class SncpRequest extends Request {
return seqid;
}
public long getServiceid() {
public DLong getServiceid() {
return serviceid;
}
public long getNameid() {
public DLong getNameid() {
return nameid;
}

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

@@ -52,31 +52,28 @@ public final class SncpResponse extends Response<SncpRequest> {
public void finish(final int retcode, final BsonWriter out) {
if (out == null) {
final ByteBuffer buffer = context.pollBuffer();
fillHeader(buffer, 0, 0, 0, retcode);
fillHeader(buffer, 0, retcode);
finish(buffer);
return;
}
final int respBodyLength = out.count(); //body总长度
final ByteBuffer[] buffers = out.toBuffers();
fillHeader(buffers[0], respBodyLength - HEADER_SIZE, 0, respBodyLength - HEADER_SIZE, retcode);
fillHeader(buffers[0], respBodyLength - HEADER_SIZE, retcode);
finish(buffers);
}
private void fillHeader(ByteBuffer buffer, int bodyLength, int bodyOffset, int framelength, int retcode) {
private void fillHeader(ByteBuffer buffer, int bodyLength, int retcode) {
//---------------------head----------------------------------
final int currentpos = buffer.position();
buffer.position(0);
buffer.putLong(request.getSeqid());
buffer.putChar((char) SncpRequest.HEADER_SIZE);
buffer.putLong(request.getServiceid());
buffer.putLong(request.getNameid());
DLong actionid = request.getActionid();
actionid.putTo(buffer);
DLong.write(buffer, request.getServiceid());
DLong.write(buffer, request.getNameid());
DLong.write(buffer, request.getActionid());
buffer.put(addrBytes);
buffer.putChar((char) this.addrPort);
buffer.putInt(bodyLength);
buffer.putInt(bodyOffset);
buffer.putInt(framelength);
buffer.putInt(retcode);
buffer.position(currentpos);
}

6
src/org/redkale/net/sncp/SncpServlet.java
View File

@@ -6,7 +6,7 @@
package org.redkale.net.sncp;
import org.redkale.net.Servlet;
import org.redkale.util.AnyValue;
import org.redkale.util.*;
/**
*
@@ -17,9 +17,9 @@ public abstract class SncpServlet implements Servlet<SncpRequest, SncpResponse>
AnyValue conf;
public abstract long getNameid();
public abstract DLong getNameid();
public abstract long getServiceid();
public abstract DLong getServiceid();
@Override
public final boolean equals(Object obj) {

6
src/org/redkale/service/Service.java
View File

@@ -9,10 +9,8 @@ import org.redkale.util.*;
/**
* 所有Service的实现类不得声明为final， 允许远程模式的public方法和public String name()方法都不能声明为final。
* 注意:
* 1、"$"是一个很特殊的Service.name值 。 被标记为@Resource(name = "$") 的Service的资源名与所属父Service的资源名一致。
* 2、以下划线_开头的Service.name只能是被系统分配, 且长度可以超过11位。使用者定义name不得以_开头。
*
* 注意: "$"是一个很特殊的Service.name值 。 被标记为@Resource(name = "$") 的Service的资源名与所属父Service的资源名一致。
*
* <p>
* @Resource(name = ".*")
* private HashMap<String, XXXService> nodemap;

17
src/org/redkale/util/DLong.java
View File

@@ -17,9 +17,11 @@ import java.util.*;
*/
public final class DLong extends Number implements Comparable<DLong> {
private final byte[] bytes;
public static final DLong ZERO = new DLong(new byte[16]);
public DLong(long v1, long v2) {
protected final byte[] bytes;
protected DLong(long v1, long v2) { //暂时不用
this.bytes = new byte[]{(byte) (v1 >> 56), (byte) (v1 >> 48), (byte) (v1 >> 40), (byte) (v1 >> 32),
(byte) (v1 >> 24), (byte) (v1 >> 16), (byte) (v1 >> 8), (byte) v1, (byte) (v2 >> 56), (byte) (v2 >> 48), (byte) (v2 >> 40), (byte) (v2 >> 32),
(byte) (v2 >> 24), (byte) (v2 >> 16), (byte) (v2 >> 8), (byte) v2};
@@ -38,8 +40,15 @@ public final class DLong extends Number implements Comparable<DLong> {
return bytes;
}
public ByteBuffer putTo(ByteBuffer buffer) {
buffer.put(bytes);
public static DLong read(ByteBuffer buffer) {
byte[] bs = new byte[16];
buffer.get(bs);
if (ZERO.equals(bs)) return ZERO;
return new DLong(bs);
}
public static ByteBuffer write(ByteBuffer buffer, DLong value) {
buffer.put(value.bytes);
return buffer;
}