Skip to content

B
byit-myth-job
Commit 0cd5c593 by huangfusuper
Options

rpc探针以及plugin探针增加

parent eb11d56a
Showing with 194 additions and 21 deletions
byit-myth-executor/myth-executor-server/src/main/java/com/byit/utils/RequestUtil.java
package com.byit.utils;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
/**
* 发送链接的工具类
*
* @author huangfu
*/
@Slf4j
......@@ -17,25 +17,26 @@ public class RequestUtil {
/**
* 带有重试的post请求
*
* @param requestUrl 请求的url
* @param body 请求的数据结果
* @param retryTotalCount 请求的总次数
* @param thisRetryCount 当前是第几次请求
* @throws InterruptedException 睡眠异常
*/
public static void retryPost(String requestUrl,String body,int retryTotalCount, int thisRetryCount) throws InterruptedException {
log.info("------当前的请求的地址为:{}-------",requestUrl);
public static void retryPost(String requestUrl, String body, int retryTotalCount, int thisRetryCount) throws InterruptedException {
log.info("------当前的请求的地址为:{}-------", requestUrl);
try {
HttpUtil.post(requestUrl, body, TIMEOUT_DEFAULT);
log.info("------{}请求成功，重试了{}次------",requestUrl,thisRetryCount-1);
}catch (Exception e) {
log.info("------{}请求成功，重试了{}次------", requestUrl, thisRetryCount - 1);
} catch (Exception e) {
//当前重试次数 小于等于总共的重试次数时
if(thisRetryCount <= retryTotalCount){
log.error("{},出现异常，异常信息为：{}，开始第{}次重试！",body,ExecutorLogUtil.getMessage(e), thisRetryCount);
if (thisRetryCount <= retryTotalCount) {
log.error("{},出现异常，异常信息为：{}，开始第{}次重试！", body, ExecutorLogUtil.getMessage(e), thisRetryCount);
Thread.sleep(1000 * thisRetryCount);
retryPost(requestUrl,body,retryTotalCount,++thisRetryCount);
}else{
throw new RuntimeException(String.format("执行机回调调度中心失败，地址为%s，失败原因是：%s,重试了%s次", requestUrl, ExecutorLogUtil.getMessage(e),retryTotalCount));
retryPost(requestUrl, body, retryTotalCount, ++thisRetryCount);
} else {
throw new RuntimeException(String.format("执行机回调调度中心失败，地址为%s，失败原因是：%s,重试了%s次", requestUrl, ExecutorLogUtil.getMessage(e), retryTotalCount));
}
}
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/reference/RpcReferenceBean.java
......@@ -14,6 +14,7 @@ import com.byit.rpc.remoting.net.params.RpcResponse;
import com.byit.rpc.remoting.provider.RpcProviderFactory;
import com.byit.rpc.serialize.Serializer;
import com.byit.rpc.util.ClassUtil;
import com.byit.rpc.util.RequestUtil;
import com.byit.rpc.util.RpcException;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
......@@ -29,6 +30,16 @@ import java.util.concurrent.TimeUnit;
*
*/
public class RpcReferenceBean {
public static void main(String[] args) {
LoadBalance loadBalance = LoadBalance.ROUND;
TreeSet<String> adds = new TreeSet<>();
adds.add("192.168.1.1");
System.out.println(loadBalance.rpcInvokerRouter.route("name", adds));
System.out.println(loadBalance.rpcInvokerRouter.route("name", adds));
System.out.println(loadBalance.rpcInvokerRouter.route("name", adds));
System.out.println(loadBalance.rpcInvokerRouter.route("name", adds));
}
private static final Logger logger = LoggerFactory.getLogger(RpcReferenceBean.class);
// [tips01: save 30ms/100invoke. why why why??? with this logger, it can save lots of time.]
......@@ -188,13 +199,17 @@ public class RpcReferenceBean {
// 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);
//finalAddress = loadBalance.rpcInvokerRouter.route(serviceKey, addressSet);
finalAddress = RequestUtil.selectRPCServer(client,loadBalance,addressSet,serverName);
logger.debug("-----选取的健壮主机为【{}】-----",finalAddress);
}
// else if (addressSet.size()==1) {
// //TODO 这里就是根据路由规则选出的路由地址
// //finalAddress = addressSet.first();
// }
}
}
if (finalAddress==null || finalAddress.trim().length()==0) {
......@@ -216,7 +231,8 @@ public class RpcReferenceBean {
// future-response set
RpcFutureResponse futureResponse = new RpcFutureResponse(invokerFactory, rpcRequest, null);
try {
//TODO 改方法需要添加ip测试功能项
// do invoke
client.asyncSend(finalAddress, rpcRequest);
......
byit-myth-rpc/src/main/java/com/byit/rpc/util/RPCLogUtil.java 0 → 100644
package com.byit.rpc.util;
import java.io.PrintWriter;
import java.io.StringWriter;
/**
* @author huangfu
*/
public class RPCLogUtil {
/**
* 读取异常信息
* @param obj 异常对象
* @return 返回堆栈异常信息
*/
public static String getMessage(Object obj) {
if (obj == null) {
return "";
}
if (obj instanceof Throwable) {
StringWriter str = new StringWriter();
PrintWriter pw = new PrintWriter(str);
((Throwable) obj).printStackTrace(pw);
return str.toString();
} else {
return obj.toString();
}
}
}
\ No newline at end of file
byit-myth-rpc/src/main/java/com/byit/rpc/util/RequestUtil.java 0 → 100644
package com.byit.rpc.util;
import com.byit.rpc.remoting.invoker.reference.RpcReferenceBean;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.rpc.remoting.net.Client;
import com.byit.rpc.remoting.net.params.Beat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.TreeSet;
public class RequestUtil {
private static final Logger log = LoggerFactory.getLogger(RequestUtil.class);
/**
* 基于负载均衡方案选择合适的主机进行选取
* @param client 客户端链接帮助其
* @param loadBalance 负载均衡器
* @param address 地址
* @param serverKey 服务名称
* @return 健全的服务地址
* @throws InterruptedException 线程异常
*/
public static String selectRPCServer(Client client, LoadBalance loadBalance, TreeSet<String> address, String serverKey) throws InterruptedException {
for (String ignored : address) {
String routeHost = loadBalance.rpcInvokerRouter.route(serverKey, address);
boolean retryRpcHost = retryRpcHost(client, routeHost, 3, 1);
if(retryRpcHost) {
return routeHost;
}
}
throw new RpcException(String.format("rpc建立通道全部失败，全部地址为%s，失败原因是：【不能与目标主机建立通道】,全部主机重试了【3】次", address));
}
/**
* rpc节点重试
* @param client 客户端
* @param host 主机
* @param retryTotalCount 重试次数
* @param thisRetryCount 当前重试次数
* @return 是否成功
*/
public static boolean retryRpcHost(Client client, String host ,int retryTotalCount, int thisRetryCount) throws InterruptedException {
log.info("------当前rpc的请求的地址为:{}-------",host);
try {
client.asyncSend(host, Beat.BEAT_PING);
log.info("------{}rpc通道建立成功，重试了{}次------",host,thisRetryCount-1);
return true;
}catch (Exception e) {
//当前重试次数 小于等于总共的重试次数时
if(thisRetryCount <= retryTotalCount){
log.error("{},rpc通道建立时出现异常，异常信息为：{}，开始第{}次重试！",host,RPCLogUtil.getMessage(e), thisRetryCount);
Thread.sleep(1000 * thisRetryCount);
retryRpcHost(client,host,retryTotalCount,++thisRetryCount);
}
log.error("----与主机【{}】建立通道，总共【{}】次，全部失败，开始挑选下一个负载均衡方案重试，请稍后------",host, retryTotalCount);
return false;
}
}
}
byit-plugin-core/myth-plugin-rpc-client/pom.xml
......@@ -25,6 +25,10 @@
<version>1.0-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
</dependencies>
<distributionManagement>
......
byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/init/PluginClientInitialization.java
......@@ -8,6 +8,7 @@ import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.registry.PluginServiceRegistry;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.utils.ClientRpcUtil;
import org.apache.commons.lang3.StringUtils;
import java.lang.reflect.Proxy;
......@@ -70,12 +71,12 @@ public class PluginClientInitialization {
String address = null;
TreeSet<String> discovery = pluginServiceRegistry.discovery(pluginRpcRequestPacket.getJobName());
if(CollectionUtil.isNotEmpty(discovery)){
if(discovery.size() ==1){
address = discovery.first();
}else{
address = loadBalance.rpcInvokerRouter.route(pluginRpcRequestPacket.getJobName(), discovery);
}
address = ClientRpcUtil.selectRPCServer(pluginClient,loadBalance,discovery,pluginRpcRequestPacket.getJobName());
// if(discovery.size() ==1){
// address = discovery.first();
// }else{
// address = loadBalance.rpcInvokerRouter.route(pluginRpcRequestPacket.getJobName(), discovery);
// }
}
if(StringUtils.isBlank(address)){
......
byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/utils/ClientRpcUtil.java 0 → 100644
package com.byit.utils;
import com.byit.client.PluginClient;
import com.byit.param.PluginBeat;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.rpc.util.RPCLogUtil;
import com.byit.rpc.util.RpcException;
import lombok.extern.slf4j.Slf4j;
import java.util.TreeSet;
@Slf4j
public class ClientRpcUtil {
/**
* 基于负载均衡方案选择合适的主机进行选取
*
* @param client 客户端链接帮助其
* @param loadBalance 负载均衡器
* @param address 地址
* @param serverKey 服务名称
* @return 健全的服务地址
* @throws InterruptedException 线程异常
*/
public static String selectRPCServer(PluginClient client, LoadBalance loadBalance, TreeSet<String> address, String serverKey) throws InterruptedException {
for (String ignored : address) {
String routeHost = loadBalance.rpcInvokerRouter.route(serverKey, address);
boolean retryRpcHost = retryRpcHost(client, routeHost, 3, 1);
if (retryRpcHost) {
return routeHost;
}
}
throw new RpcException(String.format("plugin-rpc建立通道全部失败，全部地址为%s，失败原因是：【不能与目标主机建立通道】,全部主机重试了【3】次", address));
}
/**
* rpc节点重试
*
* @param client 客户端
* @param host 主机
* @param retryTotalCount 重试次数
* @param thisRetryCount 当前重试次数
* @return 是否成功
*/
public static boolean retryRpcHost(PluginClient client, String host, int retryTotalCount, int thisRetryCount) throws InterruptedException {
log.info("------当前plugin-rpc的请求的地址为:{}-------", host);
try {
client.send(host, PluginBeat.PLUGIN_RPC_REQUEST_PACKET);
log.info("------{}plugin-rpc通道建立成功，重试了{}次------", host, thisRetryCount - 1);
return true;
} catch (Exception e) {
//当前重试次数 小于等于总共的重试次数时
if (thisRetryCount <= retryTotalCount) {
log.error("{},plugin-rpc通道建立时出现异常，异常信息为：{}，开始第{}次重试！", host, RPCLogUtil.getMessage(e), thisRetryCount);
Thread.sleep(1000 * thisRetryCount);
retryRpcHost(client, host, retryTotalCount, ++thisRetryCount);
}
log.error("----与主机【{}】建立通道，总共【{}】次，全部失败，开始挑选下一个负载均衡方案重试，请稍后------", host, retryTotalCount);
return false;
}
}
}
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