Skip to content

B
byit-myth-job
Commit fa323143 by guo_minglei@163.com
Options

Merge remote-tracking branch 'origin/developer' into developer

parents 4d4ebfde dd74a46b
Showing with 140 additions and 135 deletions
byit-myth-admin/src/main/java/com/byit/config/AdminRegisterConfig.java
......@@ -25,6 +25,10 @@ public class AdminRegisterConfig {
// @Value("${myth-rpc.registry.port}")
// private int port;
/**
* 初始化 rpc客户端 属于消费者
* @return
*/
@Bean
public RpcSpringInvokerFactory invokerFactory(){
RpcSpringInvokerFactory invokerFactory = new RpcSpringInvokerFactory();
......@@ -39,6 +43,10 @@ public class AdminRegisterConfig {
return invokerFactory;
}
/**
* 设置对外提供服务的接口 这个其实是配合网关调用的方式 属于生产者
* @return
*/
@Bean
public RpcSpringProviderFactory rpcSpringProviderFactory() {
RpcSpringProviderFactory providerFactory = new RpcSpringProviderFactory();
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/impl/RpcSpringInvokerFactory.java
......@@ -87,7 +87,7 @@ public class RpcSpringInvokerFactory extends InstantiationAwareBeanPostProcessor
null,
rpcInvokerFactory
);
//获取一个代理对象
Object serviceProxy = referenceBean.getObject();
// set bean
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/reference/RpcReferenceBean.java
......@@ -136,154 +136,151 @@ public class RpcReferenceBean {
public Object getObject() {
return Proxy.newProxyInstance(Thread.currentThread()
.getContextClassLoader(), new Class[] { iface },
new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// method param
String className = method.getDeclaringClass().getName(); // iface.getName()
String varsion_ = version;
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
Object[] parameters = args;
// filters for generic
if (className.equals(RpcGenericService.class.getName()) && methodName.equals("invoke")) {
Class<?>[] paramTypes = null;
if (args[3]!=null) {
String[] paramTypes_str = (String[]) args[3];
if (paramTypes_str.length > 0) {
paramTypes = new Class[paramTypes_str.length];
for (int i = 0; i < paramTypes_str.length; i++) {
paramTypes[i] = ClassUtil.resolveClass(paramTypes_str[i]);
}
(proxy, method, args) -> {
// method param
String className = method.getDeclaringClass().getName(); // iface.getName()
String varsion_ = version;
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
Object[] parameters = args;
// filters for generic
if (className.equals(RpcGenericService.class.getName()) && methodName.equals("invoke")) {
Class<?>[] paramTypes = null;
if (args[3]!=null) {
String[] paramTypes_str = (String[]) args[3];
if (paramTypes_str.length > 0) {
paramTypes = new Class[paramTypes_str.length];
for (int i = 0; i < paramTypes_str.length; i++) {
paramTypes[i] = ClassUtil.resolveClass(paramTypes_str[i]);
}
}
className = (String) args[0];
varsion_ = (String) args[1];
methodName = (String) args[2];
parameterTypes = paramTypes;
parameters = (Object[]) args[4];
}
// filters method like "Object.toString()"
if (className.equals(Object.class.getName())) {
logger.info(">>>>>>>>>>> myth-rpc proxy class-method not support [{}#{}]", className, methodName);
throw new RpcException("myth-rpc proxy class-method not support");
}
className = (String) args[0];
varsion_ = (String) args[1];
methodName = (String) args[2];
parameterTypes = paramTypes;
parameters = (Object[]) args[4];
}
// address
String finalAddress = address;
if (finalAddress==null || finalAddress.trim().length()==0) {
if (invokerFactory!=null && invokerFactory.getServiceRegistry()!=null) {
// discovery
String serviceKey = RpcProviderFactory.makeServiceKey(className, varsion_);
TreeSet<String> addressSet = invokerFactory.getServiceRegistry().discovery(serviceKey);
// load balance
if (addressSet==null || addressSet.size()==0) {
// pass
} else if (addressSet.size()==1) {
//TODO 这里就是根据路由规则选出的路由地址
finalAddress = addressSet.first();
} else {
finalAddress = loadBalance.rpcInvokerRouter.route(serviceKey, addressSet);
}
// filters method like "Object.toString()"
if (className.equals(Object.class.getName())) {
logger.info(">>>>>>>>>>> myth-rpc proxy class-method not support [{}#{}]", className, methodName);
throw new RpcException("myth-rpc proxy class-method not support");
}
// address
String finalAddress = address;
if (finalAddress==null || finalAddress.trim().length()==0) {
if (invokerFactory!=null && invokerFactory.getServiceRegistry()!=null) {
// discovery
String serviceKey = RpcProviderFactory.makeServiceKey(className, varsion_);
TreeSet<String> addressSet = invokerFactory.getServiceRegistry().discovery(serviceKey);
// load balance
if (addressSet==null || addressSet.size()==0) {
// pass
} else if (addressSet.size()==1) {
//TODO 这里就是根据路由规则选出的路由地址
finalAddress = addressSet.first();
} else {
finalAddress = loadBalance.rpcInvokerRouter.route(serviceKey, addressSet);
}
}
if (finalAddress==null || finalAddress.trim().length()==0) {
throw new RpcException("myth-rpc reference bean["+ className +"] address empty");
}
}
if (finalAddress==null || finalAddress.trim().length()==0) {
throw new RpcException("myth-rpc reference bean["+ className +"] address empty");
}
// request
RpcRequest rpcRequest = new RpcRequest();
rpcRequest.setRequestId(UUID.randomUUID().toString());
rpcRequest.setCreateMillisTime(System.currentTimeMillis());
rpcRequest.setAccessToken(accessToken);
rpcRequest.setClassName(className);
rpcRequest.setMethodName(methodName);
rpcRequest.setParameterTypes(parameterTypes);
rpcRequest.setParameters(parameters);
// send
if (CallType.SYNC == callType) {
// future-response set
RpcFutureResponse futureResponse = new RpcFutureResponse(invokerFactory, rpcRequest, null);
try {
// do invoke
client.asyncSend(finalAddress, rpcRequest);
// future get
RpcResponse rpcResponse = futureResponse.get(timeout, TimeUnit.MILLISECONDS);
if (rpcResponse.getErrorMsg() != null) {
throw new RpcException(rpcResponse.getErrorMsg());
}
return rpcResponse.getResult();
} catch (Exception e) {
logger.info(">>>>>>>>>>> myth-rpc, invoke error, address:{}, RpcRequest{}", finalAddress, rpcRequest);
throw (e instanceof RpcException)?e:new RpcException(e);
} finally{
// future-response remove
futureResponse.removeInvokerFuture();
}
} else if (CallType.FUTURE == callType) {
// future-response set
RpcFutureResponse futureResponse = new RpcFutureResponse(invokerFactory, rpcRequest, null);
try {
// invoke future set
RpcInvokeFuture invokeFuture = new RpcInvokeFuture(futureResponse);
RpcInvokeFuture.setFuture(invokeFuture);
// do invoke
client.asyncSend(finalAddress, rpcRequest);
return null;
} catch (Exception e) {
logger.info(">>>>>>>>>>> myth-rpc, invoke error, address:{}, RpcRequest{}", finalAddress, rpcRequest);
// future-response remove
futureResponse.removeInvokerFuture();
throw (e instanceof RpcException)?e:new RpcException(e);
}
} else if (CallType.CALLBACK == callType) {
// get callback
RpcInvokeCallback finalInvokeCallback = invokeCallback;
RpcInvokeCallback threadInvokeCallback = RpcInvokeCallback.getCallback();
if (threadInvokeCallback != null) {
finalInvokeCallback = threadInvokeCallback;
}
if (finalInvokeCallback == null) {
throw new RpcException("myth-rpc RpcInvokeCallback（CallType="+ CallType.CALLBACK.name() +"） cannot be null.");
// request
RpcRequest rpcRequest = new RpcRequest();
rpcRequest.setRequestId(UUID.randomUUID().toString());
rpcRequest.setCreateMillisTime(System.currentTimeMillis());
rpcRequest.setAccessToken(accessToken);
rpcRequest.setClassName(className);
rpcRequest.setMethodName(methodName);
rpcRequest.setParameterTypes(parameterTypes);
rpcRequest.setParameters(parameters);
// send
if (CallType.SYNC == callType) {
// future-response set
RpcFutureResponse futureResponse = new RpcFutureResponse(invokerFactory, rpcRequest, null);
try {
// do invoke
client.asyncSend(finalAddress, rpcRequest);
// future get
RpcResponse rpcResponse = futureResponse.get(timeout, TimeUnit.MILLISECONDS);
if (rpcResponse.getErrorMsg() != null) {
throw new RpcException(rpcResponse.getErrorMsg());
}
return rpcResponse.getResult();
} catch (Exception e) {
logger.info(">>>>>>>>>>> myth-rpc, invoke error, address:{}, RpcRequest{}", finalAddress, rpcRequest);
throw (e instanceof RpcException)?e:new RpcException(e);
} finally{
// future-response remove
futureResponse.removeInvokerFuture();
}
} else if (CallType.FUTURE == callType) {
// future-response set
RpcFutureResponse futureResponse = new RpcFutureResponse(invokerFactory, rpcRequest, null);
try {
// invoke future set
RpcInvokeFuture invokeFuture = new RpcInvokeFuture(futureResponse);
RpcInvokeFuture.setFuture(invokeFuture);
// future-response set
RpcFutureResponse futureResponse = new RpcFutureResponse(invokerFactory, rpcRequest, finalInvokeCallback);
try {
client.asyncSend(finalAddress, rpcRequest);
} catch (Exception e) {
logger.info(">>>>>>>>>>> myth-rpc, invoke error, address:{}, RpcRequest{}", finalAddress, rpcRequest);
// do invoke
client.asyncSend(finalAddress, rpcRequest);
// future-response remove
futureResponse.removeInvokerFuture();
return null;
} catch (Exception e) {
logger.info(">>>>>>>>>>> myth-rpc, invoke error, address:{}, RpcRequest{}", finalAddress, rpcRequest);
throw (e instanceof RpcException)?e:new RpcException(e);
}
// future-response remove
futureResponse.removeInvokerFuture();
throw (e instanceof RpcException)?e:new RpcException(e);
}
} else if (CallType.CALLBACK == callType) {
return null;
} else if (CallType.ONEWAY == callType) {
client.asyncSend(finalAddress, rpcRequest);
return null;
} else {
throw new RpcException("myth-rpc callType["+ callType +"] invalid");
// get callback
RpcInvokeCallback finalInvokeCallback = invokeCallback;
RpcInvokeCallback threadInvokeCallback = RpcInvokeCallback.getCallback();
if (threadInvokeCallback != null) {
finalInvokeCallback = threadInvokeCallback;
}
if (finalInvokeCallback == null) {
throw new RpcException("myth-rpc RpcInvokeCallback（CallType="+ CallType.CALLBACK.name() +"） cannot be null.");
}
// future-response set
RpcFutureResponse futureResponse = new RpcFutureResponse(invokerFactory, rpcRequest, finalInvokeCallback);
try {
client.asyncSend(finalAddress, rpcRequest);
} catch (Exception e) {
logger.info(">>>>>>>>>>> myth-rpc, invoke error, address:{}, RpcRequest{}", finalAddress, rpcRequest);
// future-response remove
futureResponse.removeInvokerFuture();
throw (e instanceof RpcException)?e:new RpcException(e);
}
return null;
} else if (CallType.ONEWAY == callType) {
client.asyncSend(finalAddress, rpcRequest);
return null;
} else {
throw new RpcException("myth-rpc callType["+ callType +"] invalid");
}
});
}
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/net/common/ConnectClient.java
......@@ -82,7 +82,7 @@ public abstract class ConnectClient {
if (connectClient!=null && connectClient.isValidate()) {
return connectClient;
}
//获取对应地址的锁对象
// lock
Object clientLock = connectClientLockMap.get(address);
if (clientLock == null) {
......
demo-client/byit-demo-client/src/main/java/com/byit/job/AddComplexPy.java
......@@ -19,7 +19,7 @@ public class AddComplexPy {
runNode.setNodeId("2");
runNode.setNodeName("lijkki");
runNode.setRunCmd("python ${biz_file}");
JobUtils.setRequestUrl("http://127.0.0.1:8081/myth-job-admin");
JobUtils.setRequestUrl("http://10.0.120.208:8081/myth-job-admin/");
JobUtils.setTOKEN("test");
JobUtils.realExectNode(runNode);
/* PluginPackage pluginPackage = new PluginPackage();
......
demo-client/byit-demo-client/src/main/java/com/byit/job/AddComplexPy1.java
......@@ -16,7 +16,7 @@ public class AddComplexPy1 {
PluginPackage pluginPackage = new PluginPackage();
pluginPackage.setWorkspaceName("test");
pluginPackage.setFlow(createFlow());
JobUtils.setRequestUrl("http://127.0.0.1:8081/myth-job-admin");
JobUtils.setRequestUrl("http://10.0.120.208:8081/myth-job-admin");
JobUtils.setTOKEN("test");
JobUtils.publish(pluginPackage);
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment