This commit is contained in:
地平线
2015-08-13 10:55:22 +08:00
parent 70448778f0
commit 6c2ddf49a7
5 changed files with 60 additions and 58 deletions

View File

@@ -38,7 +38,7 @@
一个组包含多个NODE 同一Service服务可以由多个进程提供这些进程称为一个GROUP且同一GROUP内的进程必须在同一机房或局域网内 一个组包含多个NODE 同一Service服务可以由多个进程提供这些进程称为一个GROUP且同一GROUP内的进程必须在同一机房或局域网内
name: 服务组ID长度不能超过11个字节. 默认为空字符串。 name: 服务组ID长度不能超过11个字节. 默认为空字符串。
protocol值只能是UDP TCP 默认TCP protocol值只能是UDP TCP 默认TCP
注意: 一个node只能所属一个group。 注意: 一个node只能所属一个group。只要存在protocol == SNCP的Server节点信息 就必须有group节点信息。
--> -->
<group name="" protocol="TCP"> <group name="" protocol="TCP">
<!-- <!--
@@ -87,7 +87,9 @@
protocol: required server所启动的协议有HTTP、SNCP 目前只支持HTTP、SNCP。SNCP也分TCP、UDP实现默认使用UDP实现TCP实现则使用SNCP.TCP值; protocol: required server所启动的协议有HTTP、SNCP 目前只支持HTTP、SNCP。SNCP也分TCP、UDP实现默认使用UDP实现TCP实现则使用SNCP.TCP值;
host: 服务所占address 默认: 0.0.0.0 host: 服务所占address 默认: 0.0.0.0
port: required 服务所占端口 port: required 服务所占端口
group: 所属组的节点,多个节点值用;隔开如果配置文件中存在多个SNCP协议的Server节点需要显式指定group属性 group: 所属组的节点,多个节点值用;隔开如果配置文件中存在多个SNCP协议的Server节点需要显式指定group属性.
当 protocol == SNCP 时 group表示当前Server与哪些节点组关联。
当 protocol != SNCP 时 group只能是空或者一个group的节点值不能为多个节点值。
root: 如果是web类型服务则包含页面 默认:{APP_HOME}/root root: 如果是web类型服务则包含页面 默认:{APP_HOME}/root
lib: server额外的class目录 默认为空 lib: server额外的class目录 默认为空
charset: 文本编码, 默认: UTF-8 charset: 文本编码, 默认: UTF-8
@@ -100,7 +102,7 @@
readTimeoutSecond: 读操作超时秒数, 默认0 表示永久不超时 readTimeoutSecond: 读操作超时秒数, 默认0 表示永久不超时
writeTimeoutSecond: 写操作超时秒数, 默认0 表示永久不超时 writeTimeoutSecond: 写操作超时秒数, 默认0 表示永久不超时
forwardproxy: 正向代理(支持CONNECT); 默认: false 只有当protocol=HTTP才生效 forwardproxy: 正向代理(支持CONNECT); 默认: false 只有当 protocol==HTTP 才生效
--> -->
<server protocol="HTTP" host="127.0.0.1" port="6060" root="root" lib=""> <server protocol="HTTP" host="127.0.0.1" port="6060" root="root" lib="">

View File

@@ -385,6 +385,8 @@ public final class Application {
others.add(entry); others.add(entry);
} }
} }
if (!sncps.isEmpty() && globalNodes.isEmpty()) throw new RuntimeException("found SNCP Server node bug not found <group> node info.");
factory.register(RESNAME_SNCP_NODES, new TypeToken<Map<InetSocketAddress, String>>() { factory.register(RESNAME_SNCP_NODES, new TypeToken<Map<InetSocketAddress, String>>() {
}.getType(), globalNodes); }.getType(), globalNodes);
factory.register(RESNAME_SNCP_NODES, new TypeToken<HashMap<InetSocketAddress, String>>() { factory.register(RESNAME_SNCP_NODES, new TypeToken<HashMap<InetSocketAddress, String>>() {

View File

@@ -58,7 +58,7 @@ public final class NodeHttpServer extends NodeServer {
for (NodeServer ns : application.servers) { for (NodeServer ns : application.servers) {
if (!ns.isSNCP()) continue; if (!ns.isSNCP()) continue;
if (sncpServer0 == null) sncpServer0 = (NodeSncpServer) ns; if (sncpServer0 == null) sncpServer0 = (NodeSncpServer) ns;
if (ns.getNodeGroup().equals(getNodeGroup())) { if (ns.getSncpGroup().equals(getSncpGroup())) {
sncpServer0 = (NodeSncpServer) ns; sncpServer0 = (NodeSncpServer) ns;
break; break;
} }
@@ -77,20 +77,20 @@ public final class NodeHttpServer extends NodeServer {
if (nodeService == null) { if (nodeService == null) {
Class<? extends Service> sc = (Class<? extends Service>) application.webSocketNodeClass; Class<? extends Service> sc = (Class<? extends Service>) application.webSocketNodeClass;
nodeService = Sncp.createLocalService(rcname, (Class<? extends Service>) (sc == null ? WebSocketNodeService.class : sc), nodeService = Sncp.createLocalService(rcname, (Class<? extends Service>) (sc == null ? WebSocketNodeService.class : sc),
getNodeAddress(), (sc == null ? null : nodeSameGroupTransports), (sc == null ? null : nodeDiffGroupTransports)); getSncpAddress(), (sc == null ? null : nodeSameGroupTransports), (sc == null ? null : nodeDiffGroupTransports));
regFactory.register(rcname, WebSocketNode.class, nodeService); regFactory.register(rcname, WebSocketNode.class, nodeService);
WebSocketNode wsn = (WebSocketNode) nodeService; WebSocketNode wsn = (WebSocketNode) nodeService;
wsn.setLocalSncpAddress(getNodeAddress()); wsn.setLocalSncpAddress(getSncpAddress());
final Set<InetSocketAddress> alladdrs = new HashSet<>(); final Set<InetSocketAddress> alladdrs = new HashSet<>();
application.globalNodes.forEach((k, v) -> alladdrs.add(k)); application.globalNodes.forEach((k, v) -> alladdrs.add(k));
alladdrs.remove(getNodeAddress()); alladdrs.remove(getSncpAddress());
WebSocketNode remoteNode = (WebSocketNode) Sncp.createRemoteService(rcname, (Class<? extends Service>) (sc == null ? WebSocketNodeService.class : sc), WebSocketNode remoteNode = (WebSocketNode) Sncp.createRemoteService(rcname, (Class<? extends Service>) (sc == null ? WebSocketNodeService.class : sc),
getNodeAddress(), (sc == null ? null : loadTransport(getNodeGroup(), getNodeProtocol(), alladdrs))); getSncpAddress(), (sc == null ? null : loadTransport(getSncpGroup(), getNodeProtocol(), alladdrs)));
wsn.setRemoteWebSocketNode(remoteNode); wsn.setRemoteWebSocketNode(remoteNode);
factory.inject(nodeService); factory.inject(nodeService);
factory.inject(remoteNode); factory.inject(remoteNode);
if (sncpServer != null) { if (sncpServer != null) {
ServiceWrapper wrapper = new ServiceWrapper((Class<? extends Service>) (sc == null ? WebSocketNodeService.class : sc), nodeService, getNodeGroup(), rcname, null); ServiceWrapper wrapper = new ServiceWrapper((Class<? extends Service>) (sc == null ? WebSocketNodeService.class : sc), nodeService, getSncpGroup(), rcname, null);
sncpServer.getSncpServer().addService(wrapper); sncpServer.getSncpServer().addService(wrapper);
} }
} }

View File

@@ -45,9 +45,9 @@ public abstract class NodeServer {
private final Server server; private final Server server;
private InetSocketAddress nodeAddress; //HttpServer中的nodeAddress 为所属group对应的SncpServer, 为null表示没有分布式结构 private InetSocketAddress sncpAddress; //HttpServer中的sncpAddress 为所属group对应的SncpServer, 为null表示只是单节点,没有分布式结构
private String nodeGroup = ""; //当前Server的SNCP协议的组 private String sncpGroup = ""; //当前Server的SNCP协议的组
private AnyValue nodeConf; private AnyValue nodeConf;
@@ -77,31 +77,34 @@ public abstract class NodeServer {
this.nodeConf = config == null ? AnyValue.create() : config; this.nodeConf = config == null ? AnyValue.create() : config;
if (isSNCP()) { // SNCP协议 if (isSNCP()) { // SNCP协议
String host = this.nodeConf.getValue("host", "0.0.0.0").replace("0.0.0.0", ""); String host = this.nodeConf.getValue("host", "0.0.0.0").replace("0.0.0.0", "");
this.nodeAddress = new InetSocketAddress(host.isEmpty() ? application.localAddress.getHostAddress() : host, this.nodeConf.getIntValue("port")); this.sncpAddress = new InetSocketAddress(host.isEmpty() ? application.localAddress.getHostAddress() : host, this.nodeConf.getIntValue("port"));
this.nodeGroup = application.globalNodes.getOrDefault(this.nodeAddress, ""); this.sncpGroup = application.globalNodes.getOrDefault(this.sncpAddress, "");
if (server != null) this.nodeProtocol = server.getProtocol(); if (server != null) this.nodeProtocol = server.getProtocol();
} else { // HTTP协议 } else { // HTTP协议
this.sncpAddress = null;
this.sncpGroup = "";
this.nodeProtocol = Sncp.DEFAULT_PROTOCOL;
String mbgroup = this.nodeConf.getValue("group", ""); String mbgroup = this.nodeConf.getValue("group", "");
NodeServer sncpServer = null; //有匹配的就取匹配的, 没有且SNCP只有一个则取此SNCP。 NodeServer sncpServer = null; //有匹配的就取匹配的, 没有且SNCP只有一个则取此SNCP。
for (NodeServer ns : application.servers) { for (NodeServer ns : application.servers) {
if (!ns.isSNCP()) continue; if (!ns.isSNCP()) continue;
if (sncpServer == null) sncpServer = ns; if (sncpServer == null) sncpServer = ns;
if (ns.getNodeGroup().equals(mbgroup)) { if (ns.getSncpGroup().equals(mbgroup)) {
sncpServer = ns; sncpServer = ns;
break; break;
} }
} }
if (sncpServer != null) { if (sncpServer != null) {
this.nodeAddress = sncpServer.getNodeAddress(); this.sncpAddress = sncpServer.getSncpAddress();
this.nodeGroup = sncpServer.getNodeGroup(); this.sncpGroup = sncpServer.getSncpGroup();
this.nodeProtocol = sncpServer.getNodeProtocol(); this.nodeProtocol = sncpServer.getNodeProtocol();
} }
} }
if (this.nodeAddress != null) { // 无分布式结构下 HTTP协议的nodeAddress 为 null if (this.sncpAddress != null) { // 无分布式结构下 HTTP协议的sncpAddress 为 null
this.factory.register(RESNAME_SNCP_NODE, SocketAddress.class, this.nodeAddress); this.factory.register(RESNAME_SNCP_NODE, SocketAddress.class, this.sncpAddress);
this.factory.register(RESNAME_SNCP_NODE, InetSocketAddress.class, this.nodeAddress); this.factory.register(RESNAME_SNCP_NODE, InetSocketAddress.class, this.sncpAddress);
this.factory.register(RESNAME_SNCP_NODE, String.class, this.nodeAddress.getAddress().getHostAddress()); this.factory.register(RESNAME_SNCP_NODE, String.class, this.sncpAddress.getAddress().getHostAddress());
this.factory.register(RESNAME_SNCP_GROUP, this.nodeGroup); this.factory.register(RESNAME_SNCP_GROUP, this.sncpGroup);
} }
{ {
//设置root文件夹 //设置root文件夹
@@ -135,9 +138,9 @@ public abstract class NodeServer {
regFactory.register(rs.name(), DataSource.class, source); regFactory.register(rs.name(), DataSource.class, source);
Class<? extends Service> sc = (Class<? extends Service>) application.dataCacheListenerClass; Class<? extends Service> sc = (Class<? extends Service>) application.dataCacheListenerClass;
if (sc != null) { if (sc != null) {
Service cacheListenerService = Sncp.createLocalService(rs.name(), sc, this.nodeAddress, nodeSameGroupTransports, nodeDiffGroupTransports); Service cacheListenerService = Sncp.createLocalService(rs.name(), sc, this.sncpAddress, nodeSameGroupTransports, nodeDiffGroupTransports);
regFactory.register(rs.name(), DataCacheListener.class, cacheListenerService); regFactory.register(rs.name(), DataCacheListener.class, cacheListenerService);
ServiceWrapper wrapper = new ServiceWrapper(sc, cacheListenerService, nodeGroup, rs.name(), null); ServiceWrapper wrapper = new ServiceWrapper(sc, cacheListenerService, sncpGroup, rs.name(), null);
localServices.add(wrapper); localServices.add(wrapper);
if (consumer != null) consumer.accept(wrapper); if (consumer != null) consumer.accept(wrapper);
rf.inject(cacheListenerService); rf.inject(cacheListenerService);
@@ -151,18 +154,18 @@ public abstract class NodeServer {
} }
protected List<Transport>[] parseTransport(final String[] groups) { protected List<Transport>[] parseTransport(final String[] groups) {
final Set<InetSocketAddress> sameGroupAddrs = application.findGlobalGroup(this.nodeGroup); final Set<InetSocketAddress> sameGroupAddrs = application.findGlobalGroup(this.sncpGroup);
final Map<String, Set<InetSocketAddress>> diffGroupAddrs = new HashMap<>(); final Map<String, Set<InetSocketAddress>> diffGroupAddrs = new HashMap<>();
for (String groupitem : groups) { for (String groupitem : groups) {
final Set<InetSocketAddress> addrs = application.findGlobalGroup(groupitem); final Set<InetSocketAddress> addrs = application.findGlobalGroup(groupitem);
if (addrs == null || groupitem.equals(this.nodeGroup)) continue; if (addrs == null || groupitem.equals(this.sncpGroup)) continue;
diffGroupAddrs.put(groupitem, addrs); diffGroupAddrs.put(groupitem, addrs);
} }
final List<Transport> sameGroupTransports0 = new ArrayList<>(); final List<Transport> sameGroupTransports0 = new ArrayList<>();
if (sameGroupAddrs != null) { if (sameGroupAddrs != null) {
sameGroupAddrs.remove(this.nodeAddress); sameGroupAddrs.remove(this.sncpAddress);
for (InetSocketAddress iaddr : sameGroupAddrs) { for (InetSocketAddress iaddr : sameGroupAddrs) {
sameGroupTransports0.add(loadTransport(this.nodeGroup, getNodeProtocol(), iaddr)); sameGroupTransports0.add(loadTransport(this.sncpGroup, getNodeProtocol(), iaddr));
} }
} }
final List<Transport> diffGroupTransports0 = new ArrayList<>(); final List<Transport> diffGroupTransports0 = new ArrayList<>();
@@ -174,12 +177,12 @@ public abstract class NodeServer {
public abstract boolean isSNCP(); public abstract boolean isSNCP();
public InetSocketAddress getNodeAddress() { public InetSocketAddress getSncpAddress() {
return nodeAddress; return sncpAddress;
} }
public String getNodeGroup() { public String getSncpGroup() {
return nodeGroup; return sncpGroup;
} }
public String getNodeProtocol() { public String getNodeProtocol() {
@@ -235,7 +238,7 @@ public abstract class NodeServer {
if (serviceFilter == null) return; if (serviceFilter == null) return;
final String threadName = "[" + Thread.currentThread().getName() + "] "; final String threadName = "[" + Thread.currentThread().getName() + "] ";
final Set<FilterEntry<Service>> entrys = serviceFilter.getFilterEntrys(); final Set<FilterEntry<Service>> entrys = serviceFilter.getFilterEntrys();
final String defgroup = nodeConf == null ? "" : nodeConf.getValue("group", ""); //Server节点获取group信息 final String defgroups = nodeConf == null ? "" : nodeConf.getValue("group", ""); //Server节点获取group信息
ResourceFactory regFactory = isSNCP() ? application.factory : factory; ResourceFactory regFactory = isSNCP() ? application.factory : factory;
for (FilterEntry<Service> entry : entrys) { //service实现类 for (FilterEntry<Service> entry : entrys) { //service实现类
final Class<? extends Service> type = entry.getType(); final Class<? extends Service> type = entry.getType();
@@ -245,50 +248,44 @@ public abstract class NodeServer {
if (Modifier.isAbstract(type.getModifiers())) continue; if (Modifier.isAbstract(type.getModifiers())) continue;
if (type.getAnnotation(Ignore.class) != null) continue; if (type.getAnnotation(Ignore.class) != null) continue;
if (!isSNCP() && factory.find(entry.getName(), type) != null) continue; if (!isSNCP() && factory.find(entry.getName(), type) != null) continue;
String group = entry.getGroup(); String groups = entry.getGroup();
if (group == null || group.isEmpty()) group = defgroup; if (groups == null || groups.isEmpty()) groups = defgroups;
final Set<InetSocketAddress> sameGroupAddrs = new LinkedHashSet<>(); final Set<InetSocketAddress> sameGroupAddrs = new LinkedHashSet<>();
final Map<String, Set<InetSocketAddress>> diffGroupAddrs = new HashMap<>(); final Map<String, Set<InetSocketAddress>> diffGroupAddrs = new HashMap<>();
for (String str : group.split(";")) { for (String g : groups.split(";")) {
application.globalNodes.forEach((k, v) -> { if (g.isEmpty()) continue;
if (v.equals(str)) { Set<InetSocketAddress> set = application.findGlobalGroup(g);
if (v.equals(this.nodeGroup)) { if (set == null) throw new RuntimeException(type.getName() + " has illegal group (" + groups + ")");
sameGroupAddrs.add(k); if (g.equals(this.sncpGroup)) {
} else { sameGroupAddrs.addAll(set);
Set<InetSocketAddress> set = diffGroupAddrs.get(v); } else {
if (set == null) { diffGroupAddrs.put(g, set);
set = new LinkedHashSet<>(); }
diffGroupAddrs.put(v, set);
}
set.add(k);
}
}
});
} }
final boolean localable = sameGroupAddrs.contains(this.nodeAddress); final boolean localable = this.sncpAddress == null || sameGroupAddrs.contains(this.sncpAddress);
Service service; Service service;
List<Transport> diffGroupTransports = new ArrayList<>(); List<Transport> diffGroupTransports = new ArrayList<>();
diffGroupAddrs.forEach((k, v) -> diffGroupTransports.add(loadTransport(k, server.getProtocol(), v))); diffGroupAddrs.forEach((k, v) -> diffGroupTransports.add(loadTransport(k, server.getProtocol(), v)));
if (localable || (sameGroupAddrs.isEmpty() && diffGroupTransports.isEmpty())) { if (localable || (sameGroupAddrs.isEmpty() && diffGroupTransports.isEmpty())) {
sameGroupAddrs.remove(this.nodeAddress); sameGroupAddrs.remove(this.sncpAddress);
List<Transport> sameGroupTransports = new ArrayList<>(); List<Transport> sameGroupTransports = new ArrayList<>();
for (InetSocketAddress iaddr : sameGroupAddrs) { for (InetSocketAddress iaddr : sameGroupAddrs) {
Set<InetSocketAddress> tset = new HashSet<>(); Set<InetSocketAddress> tset = new HashSet<>();
tset.add(iaddr); tset.add(iaddr);
sameGroupTransports.add(loadTransport(this.nodeGroup, server.getProtocol(), tset)); sameGroupTransports.add(loadTransport(this.sncpGroup, server.getProtocol(), tset));
} }
service = Sncp.createLocalService(entry.getName(), type, this.nodeAddress, sameGroupTransports, diffGroupTransports); service = Sncp.createLocalService(entry.getName(), type, this.sncpAddress, sameGroupTransports, diffGroupTransports);
} else { } else {
StringBuilder g = new StringBuilder(this.nodeGroup); StringBuilder g = new StringBuilder(this.sncpGroup);
diffGroupAddrs.forEach((k, v) -> { diffGroupAddrs.forEach((k, v) -> {
if (g.length() > 0) g.append(';'); if (g.length() > 0) g.append(';');
g.append(k); g.append(k);
sameGroupAddrs.addAll(v); sameGroupAddrs.addAll(v);
}); });
if (sameGroupAddrs.isEmpty()) throw new RuntimeException(type + ":" + group); if (sameGroupAddrs.isEmpty()) throw new RuntimeException(type.getName() + " has no remote address on group (" + groups + ")");
service = Sncp.createRemoteService(entry.getName(), type, this.nodeAddress, loadTransport(g.toString(), server.getProtocol(), sameGroupAddrs)); service = Sncp.createRemoteService(entry.getName(), type, this.sncpAddress, loadTransport(g.toString(), server.getProtocol(), sameGroupAddrs));
} }
ServiceWrapper wrapper = new ServiceWrapper(type, service, entry); ServiceWrapper wrapper = new ServiceWrapper(type, service, entry);
if (factory.find(wrapper.getName(), wrapper.getType()) == null) { if (factory.find(wrapper.getName(), wrapper.getType()) == null) {
@@ -300,7 +297,7 @@ public abstract class NodeServer {
if (consumer != null) consumer.accept(wrapper); if (consumer != null) consumer.accept(wrapper);
} }
} else if (isSNCP()) { } else if (isSNCP()) {
throw new RuntimeException(ServiceWrapper.class.getSimpleName() + "(class:" + type.getName() + ", name:" + entry.getName() + ", group:" + group + ") is repeat."); throw new RuntimeException(ServiceWrapper.class.getSimpleName() + "(class:" + type.getName() + ", name:" + entry.getName() + ", group:" + groups + ") is repeat.");
} }
} }
servicecdl.countDown(); servicecdl.countDown();
@@ -327,7 +324,7 @@ public abstract class NodeServer {
} }
protected ClassFilter<Service> createServiceClassFilter(final AnyValue config) { protected ClassFilter<Service> createServiceClassFilter(final AnyValue config) {
return createClassFilter(this.nodeGroup, config, null, Service.class, Annotation.class, "services", "service"); return createClassFilter(this.sncpGroup, config, null, Service.class, Annotation.class, "services", "service");
} }
protected static ClassFilter createClassFilter(final String localGroup, final AnyValue config, Class<? extends Annotation> ref, protected static ClassFilter createClassFilter(final String localGroup, final AnyValue config, Class<? extends Annotation> ref,

View File

@@ -112,6 +112,7 @@ public final class SncpRequest extends Request {
@Override @Override
protected void prepare() { protected void prepare() {
this.keepAlive = true;
if (this.body == null) return; if (this.body == null) return;
byte[] bytes = this.body; byte[] bytes = this.body;
int pos = 0; int pos = 0;