Skip to content

B
byit-myth-job
Commit 9739275a by huangfusuper
Options

修改调度中心rpc的执行规则

parent a8cdccb0
Showing with 106 additions and 19 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/EndAndNotWarningThreadRunHelper.java
......@@ -65,6 +65,8 @@ public class EndAndNotWarningThreadRunHelper extends BaseThreadRunHelper {
log.debug("------工作流{},被设置为完成时告警-----",runRecording);
if (RunRecordingEnum.FLOW_STATUS_IS_END.getCode().equals(runRecording.getFlowStatus())) {
runRecordingAndEmailService.saveEmailAndRunRecording(runRecording);
//修改为已告警
runRecording.setIsAlarm("0");
}
break;
//失败时告警
......@@ -73,6 +75,8 @@ public class EndAndNotWarningThreadRunHelper extends BaseThreadRunHelper {
if (RunRecordingEnum.RUN_FLOW_FAILURE.getCode().equals(runRecording.getFlowRunResult())
|| RunRecordingEnum.RUN_FLOW_RE_FAILURE.getCode().equals(runRecording.getFlowRunResult())) {
runRecordingAndEmailService.saveEmailAndRunRecording(runRecording);
//修改为已告警
runRecording.setIsAlarm("0");
}
break;
//成功时告警
......@@ -81,14 +85,15 @@ public class EndAndNotWarningThreadRunHelper extends BaseThreadRunHelper {
if (RunRecordingEnum.RUN_FLOW_SUCCESS.getCode().equals(runRecording.getFlowRunResult())
|| RunRecordingEnum.RUN_FLOW_RE_SUCCESS.getCode().equals(runRecording.getFlowRunResult())) {
runRecordingAndEmailService.saveEmailAndRunRecording(runRecording);
//修改为已告警
runRecording.setIsAlarm("0");
}
break;
default:
log.debug("------工作流{},告警类别不告警-----",runRecording);
break;
}
//修改为已告警
runRecording.setIsAlarm("0");
runRecordingService.updateRunRecordingById(runRecording);
});
}else{
......
byit-myth-gateway/src/main/java/com/byit/gateway/GatewayApplication.java
......@@ -20,7 +20,7 @@ import org.springframework.web.filter.CorsFilter;
* modified By
**/
@EnableZuulProxy
@RpcService
@RpcService(serverName = "[System]-myth-job-gateway")
@SpringBootApplication
@Slf4j
public class GatewayApplication {
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/annotation/RpcReference.java
......@@ -29,6 +29,7 @@ public @interface RpcReference {
String address() default "";
String accessToken() default "";
String serverName() default "";
//RpcInvokeCallback invokeCallback() ;
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/impl/RpcSpringInvokerFactory.java
......@@ -86,7 +86,8 @@ public class RpcSpringInvokerFactory extends InstantiationAwareBeanPostProcessor
rpcReference.address(),
rpcReference.accessToken(),
null,
rpcInvokerFactory
rpcInvokerFactory,
rpcReference.serverName()
);
//获取一个代理对象 这个代理对象传入了 rpcInvokerFactory
Object serviceProxy = referenceBean.getObject();
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/reference/RpcReferenceBean.java
......@@ -15,6 +15,7 @@ import com.byit.rpc.remoting.provider.RpcProviderFactory;
import com.byit.rpc.serialize.Serializer;
import com.byit.rpc.util.ClassUtil;
import com.byit.rpc.util.RpcException;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -48,6 +49,7 @@ public class RpcReferenceBean {
private String address;
private String accessToken;
private String serverName;
private RpcInvokeCallback invokeCallback;
......@@ -63,7 +65,8 @@ public class RpcReferenceBean {
String address,
String accessToken,
RpcInvokeCallback invokeCallback,
RpcInvokerFactory invokerFactory
RpcInvokerFactory invokerFactory,
String serverName
) {
this.netType = netType;
......@@ -77,6 +80,7 @@ public class RpcReferenceBean {
this.accessToken = accessToken;
this.invokeCallback = invokeCallback;
this.invokerFactory = invokerFactory;
this.serverName = serverName;
// valid
if (this.netType==null) {
......@@ -178,7 +182,10 @@ public class RpcReferenceBean {
if (finalAddress==null || finalAddress.trim().length()==0) {
if (invokerFactory!=null && invokerFactory.getServiceRegistry()!=null) {
// discovery
String serviceKey = RpcProviderFactory.makeServiceKey(className, varsion_);
if(StringUtil.isNullOrEmpty(serverName)){
this.serverName = className;
}
String serviceKey = RpcProviderFactory.makeServiceKey(this.serverName, varsion_);
TreeSet<String> addressSet = invokerFactory.getServiceRegistry().discovery(serviceKey);
// load balance
if (addressSet==null || addressSet.size()==0) {
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/reference/impl/RpcSpringReferenceBean.java
......@@ -31,6 +31,15 @@ public class RpcSpringReferenceBean implements FactoryBean<Object>, Initializing
private String address;
private String accessToken;
private String serverName;
public String getServerName() {
return serverName;
}
public void setServerName(String serverName) {
this.serverName = serverName;
}
private RpcInvokeCallback invokeCallback;
......@@ -105,7 +114,8 @@ public class RpcSpringReferenceBean implements FactoryBean<Object>, Initializing
address,
accessToken,
invokeCallback,
xxlRpcInvokerFactory);
xxlRpcInvokerFactory,
serverName);
}
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/provider/annotation/RpcService.java
......@@ -11,6 +11,8 @@ import java.lang.annotation.*;
@Inherited
public @interface RpcService {
String serverName() default "";
String version() default "";
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/provider/impl/RpcSpringProviderFactory.java
......@@ -6,6 +6,7 @@ import com.byit.rpc.remoting.provider.RpcProviderFactory;
import com.byit.rpc.remoting.provider.annotation.RpcService;
import com.byit.rpc.serialize.Serializer;
import com.byit.rpc.util.RpcException;
import io.netty.util.internal.StringUtil;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
......@@ -122,14 +123,18 @@ public class RpcSpringProviderFactory extends RpcProviderFactory implements Appl
// add service
RpcService rpcService = serviceBean.getClass().getAnnotation(RpcService.class);
String iface = serviceBean.getClass().getInterfaces()[0].getName();
//String iface = serviceBean.getClass().getInterfaces()[0].getName();
String serverName = rpcService.serverName();
if(StringUtil.isNullOrEmpty(serverName)){
serverName = serviceBean.getClass().getInterfaces()[0].getName();
}
String version = rpcService.version();
//如果提供的是htp服务
if (rpcService.http_type()) {
this.isHttp = true;
super.addService(serverName, version, serviceBean);
super.addService(this.serverName, version, serviceBean);
} else {
super.addService(iface, version, serviceBean);
super.addService(serverName, version, serviceBean);
}
}
......
byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/utils/PluginRequestUtil.java 0 → 100644
package com.byit.utils;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpUtil;
import lombok.extern.slf4j.Slf4j;
/**
* 插件请求HTTP工具类
* @author huangfu
*/
@Slf4j
public class PluginRequestUtil {
/**
* 默认的超时时间
*/
final static Integer TIMEOUT_DEFAULT = 30000;
public static final String TOKEN_NAME = "token";
public static final String HEADER_CONTENT_TYPE = "contentType";
/**
* 带有重试的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);
try {
HttpRequest post = HttpRequest.post(requestUrl);
post.header(TOKEN_NAME, TOKEN_NAME);
post.header(HEADER_CONTENT_TYPE,"application/json");
post.body(body).execute();
log.info("------{}请求成功，重试了{}次------",requestUrl,thisRetryCount);
}catch (Exception e) {
//当前重试次数 小于等于总共的重试次数时
if(thisRetryCount <= retryTotalCount){
log.error("{},出现异常，异常信息为：{}，开始第{}次重试！",body,PluginLogUtils.getMessage(e), thisRetryCount);
Thread.sleep(1000 * thisRetryCount);
retryPost(requestUrl,body,retryTotalCount,++thisRetryCount);
}else{
throw new RuntimeException(String.format("执行机回调调度中心失败，失败原因是：%s,重试了%s次", PluginLogUtils.getMessage(e),retryTotalCount));
}
}
}
}
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/netty/handler/NettyPluginServerHandler.java
......@@ -11,11 +11,13 @@ import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.param.PluginBeat;
import com.byit.task.handler.interfaces.IJobHandler;
import com.byit.utils.PluginLogUtils;
import com.byit.utils.PluginRequestUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadPoolExecutor;
/**
......@@ -24,9 +26,6 @@ import java.util.concurrent.ThreadPoolExecutor;
*/
public class NettyPluginServerHandler extends SimpleChannelInboundHandler<PluginRpcRequestPacket> {
public static final String TOKEN_NAME = "token";
public static final String HEADER_TOKEN_NAME = TOKEN_NAME;
public static final String HEADER_CONTENT_TYPE = "contentType";
private PluginServerFactory pluginServerFactory;
private ThreadPoolExecutor threadPoolExecutor;
......@@ -87,16 +86,24 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
ctx.channel().writeAndFlush(transferPluginRpcResponse);
}
/**
* 发送信息
* @param rpcResponsePacket 结果集
* @param msg 消息
*/
private void sendMsg(PluginRpcResponsePacket rpcResponsePacket, PluginRpcRequestPacket msg ){
rpcResponsePacket.setType(ResponseTyEnum.RESPONSE.getType());
rpcResponsePacket.setRequestId(msg.getRequestId());
String responseStr = JSON.toJSONString(rpcResponsePacket);
HttpRequest post = HttpRequest.post(msg.getCallbackUrl());
post.header(HEADER_TOKEN_NAME, "Token");
post.header(HEADER_CONTENT_TYPE,"application/json");
post.body(responseStr).execute();
System.out.println("-------消息回复成功-----------");
String callbackUrl = msg.getCallbackUrl();
try {
PluginRequestUtil.retryPost(callbackUrl,responseStr,3,1);
} catch (InterruptedException e) {
e.printStackTrace();
System.err.println(String.format("执行机异常:%s", PluginLogUtils.getMessage(e)));
}
}
/**
......
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