Skip to content

B
byit-myth-job
Commit ed07e73f by huangfusuper
Options

增加生命周期回调

parent 1e7791f6
Showing with 360 additions and 29 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/dto/RunRecordingWrapped.java 0 → 100644
package com.byit.dto;
import com.byit.model.JobTask;
import com.byit.model.RunRecording;
import lombok.Data;
import java.io.Serializable;
import java.util.List;
/**
* 实例包装
*
* @author huangfu
* @date 2020年11月25日18:31:07
*/
@Data
public class RunRecordingWrapped implements Serializable {
private static final long serialVersionUID = 6309696511670011734L;
private List<JobTask> nodeList;
private RunRecording runRecording;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/InstanceRunTheLifeCycleCallback.java
package com.byit.strategy;
import com.byit.dto.RunRecordingWrapped;
import com.byit.model.JobTask;
import com.byit.model.RunRecording;
......@@ -13,7 +14,17 @@ import java.util.List;
*/
public interface InstanceRunTheLifeCycleCallback {
void postProcessAfterInitialization(RunRecording runRecording, List<JobTask> jobTaskList);
/**
* 实例执行前
* @param runRecording 运行实例
* @param jobTaskList 节点
* @return 两者的包装对象
*/
RunRecordingWrapped postProcessAfterInitialization(RunRecording runRecording, List<JobTask> jobTaskList);
void postProcessBeforeInitialization(RunRecording runRecording, List<JobTask> jobTaskList);
/**
* 后置处理器
* @param runRecording 实例
*/
void postProcessBeforeInitialization(RunRecording runRecording);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/strategy/impl/RpcCallbackInstanceRunTheLifeCycleCallback.java 0 → 100644
package com.byit.strategy.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.byit.call.TaskServer;
import com.byit.dto.RunRecordingWrapped;
import com.byit.dto.common.CallbackDto;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.dto.plugin.FlowExtendedConfiguration;
import com.byit.enums.CallMethodEnum;
import com.byit.model.JobTask;
import com.byit.model.RunRecording;
import com.byit.model.Workspace;
import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.param.CommunicationParam;
import com.byit.service.WorkspaceService;
import com.byit.strategy.InstanceRunTheLifeCycleCallback;
import com.byit.task.annotations.TaskClient;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/**
* rpc回调平台服务
*
* @author huangfu
*/
@Component
public class RpcCallbackInstanceRunTheLifeCycleCallback implements InstanceRunTheLifeCycleCallback {
@TaskClient(timeout = 30000, callMethodEnum = CallMethodEnum.SYNCHRONIZE)
private TaskServer taskServer;
private final WorkspaceService workspaceService;
public RpcCallbackInstanceRunTheLifeCycleCallback(WorkspaceService workspaceService) {
this.workspaceService = workspaceService;
}
/**
* 实例执行前
*
* @param runRecording 运行实例
* @param jobTaskList 节点
* @return 两者的包装对象
*/
@Override
public RunRecordingWrapped postProcessAfterInitialization(RunRecording runRecording, List<JobTask> jobTaskList) {
PluginRpcRequestPacket param = buildPluginRpcRequestPacket(runRecording);
PluginRpcResponsePacket call = taskServer.call(param);
Object result = call.getResult();
ScriptParamAndPlaceholderDto rpcScriptParamAndPlaceholderDto = (ScriptParamAndPlaceholderDto) result;
if (rpcScriptParamAndPlaceholderDto != null) {
List<JobTask> jobTasks = jobTaskList.stream().map(jobTask -> {
//新增参数
Map<String, String> placeholder = rpcScriptParamAndPlaceholderDto.getPlaceholder();
//新增参数
Map<String, String> scriptParam = rpcScriptParamAndPlaceholderDto.getParam();
//获取原本的数据
String runParam = jobTask.getRunParam();
//如果原本数据为空 直接重新设置然后返回
if (StringUtils.isBlank(runParam)) {
RunParamWrapped runParamWrapped = new RunParamWrapped();
runParamWrapped.setPrivateParam(JSON.toJSONString(rpcScriptParamAndPlaceholderDto));
jobTask.setRunParam(JSON.toJSONString(runParamWrapped));
return jobTask;
}
//转换节点参数为包装对象
RunParamWrapped runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
String privateParam = runParamWrapped.getPrivateParam();
//如果对应参数为空直接设置
if (StringUtils.isBlank(privateParam)) {
runParamWrapped.setPrivateParam(JSON.toJSONString(rpcScriptParamAndPlaceholderDto));
jobTask.setRunParam(JSON.toJSONString(runParamWrapped));
return jobTask;
}
//获取任务里面的数据
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(privateParam, ScriptParamAndPlaceholderDto.class);
//替换参数
if (CollectionUtil.isNotEmpty(placeholder)) {
//获取任务里面的替换参数
Map<String, String> jobTaskPlaceholder = scriptParamAndPlaceholderDto.getPlaceholder();
//如果任务里面的替换参数为空 就直接设置
if (CollectionUtil.isEmpty(jobTaskPlaceholder)) {
scriptParamAndPlaceholderDto.setPlaceholder(placeholder);
} else {
placeholder.forEach(jobTaskPlaceholder::put);
//追加数据
scriptParamAndPlaceholderDto.setPlaceholder(jobTaskPlaceholder);
}
}
//脚本参数
if (CollectionUtil.isNotEmpty(scriptParam)) {
Map<String, String> jobTaskParam = scriptParamAndPlaceholderDto.getParam();
//如果任务里面的替换参数为空 就直接设置
if (!CollectionUtil.isEmpty(jobTaskParam)) {
scriptParam.forEach(jobTaskParam::put);
//追加数据
}
scriptParamAndPlaceholderDto.setParam(jobTaskParam);
}
runParamWrapped.setPrivateParam(JSON.toJSONString(scriptParamAndPlaceholderDto));
jobTask.setRunParam(JSON.toJSONString(runParamWrapped));
return jobTask;
}).collect(Collectors.toList());
RunRecordingWrapped runRecordingWrapped = new RunRecordingWrapped();
runRecordingWrapped.setNodeList(jobTasks);
runRecordingWrapped.setRunRecording(runRecording);
return runRecordingWrapped;
}
RunRecordingWrapped runRecordingWrapped = new RunRecordingWrapped();
runRecordingWrapped.setNodeList(jobTaskList);
runRecordingWrapped.setRunRecording(runRecording);
return runRecordingWrapped;
}
/**
* 构建 PluginRpcRequestPacket
*
* @param runRecording 执行实例
* @return {@link PluginRpcRequestPacket}
*/
private PluginRpcRequestPacket buildPluginRpcRequestPacket(RunRecording runRecording) {
PluginRpcRequestPacket rpcRequestPacket = new PluginRpcRequestPacket();
String extendedConfiguration = runRecording.getExtendedConfiguration();
FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
String rpcServerKey = flowExtendedConfiguration.getRpcServerKey();
rpcRequestPacket.setCallMethodEnum(CallMethodEnum.SYNCHRONIZE);
rpcRequestPacket.setJobName(rpcServerKey);
CommunicationParam param = new CommunicationParam();
param.setRunId(runRecording.getRunId());
param.setHasMakeUp(String.valueOf(runRecording.getScheduleType()));
CallbackDto callbackDto = new CallbackDto();
callbackDto.setFlowName(runRecording.getFlowName());
callbackDto.setVersionName(runRecording.getFlowVersionName());
Workspace workspaceServiceOneById = workspaceService.findOneById(runRecording.getWorkspaceId());
callbackDto.setWorkspaceName(workspaceServiceOneById.getWorkspaceName());
param.setBody(JSON.toJSONString(callbackDto));
rpcRequestPacket.setParam(param);
return rpcRequestPacket;
}
/**
* 后置处理器
*
* @param runRecording 实例
*/
@Override
public void postProcessBeforeInitialization(RunRecording runRecording) {
}
}
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/common/CallbackDto.java 0 → 100644
package com.byit.dto.common;
import lombok.Data;
import java.io.Serializable;
/**
* 回调dto
*
* @author huangfu
* @date 2020年11月26日17:53:00
*/
@Data
public class CallbackDto implements Serializable {
private static final long serialVersionUID = 3452189099483453750L;
private String flowName;
private String workspaceName;
private String versionName;
}
byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/enums/CallMethodEnum.java 0 → 100644
package com.byit.enums;
/**
* 调用的方式 异步 同步
*
* @author huangfu
* @date 2020年11月26日11:12:26
*/
public enum CallMethodEnum {
/**
* 同步
*/
SYNCHRONIZE,
/**
* 异步
*/
ASYNCHRONOUS
}
byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/packet/request/PluginRpcRequestPacket.java
package com.byit.packet.request;
import com.byit.enums.CallMethodEnum;
import com.byit.enums.Command;
import com.byit.enums.SerializerAlgorithm;
import com.byit.packet.BasePacketModel;
......@@ -32,6 +33,8 @@ public class PluginRpcRequestPacket extends BasePacketModel {
private String callbackUrl;
private CallMethodEnum callMethodEnum = CallMethodEnum.ASYNCHRONOUS;
@Override
public Command getCommand() {
return Command.RUN_REMOTELY_JOB_NODE_REQUEST;
......
byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/task/annotations/TaskClient.java
package com.byit.task.annotations;
import com.byit.enums.CallMethodEnum;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import java.lang.annotation.*;
......@@ -29,4 +30,6 @@ public @interface TaskClient {
* @return
*/
long timeout() default 1000;
CallMethodEnum callMethodEnum() default CallMethodEnum.ASYNCHRONOUS;
}
byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/factory/PluginSpringClientFactory.java
......@@ -2,7 +2,7 @@ package com.byit.factory;
import com.byit.client.NettyPluginClient;
import com.byit.client.PluginClient;
import com.byit.executor.handler.interfaces.IJobHandler;
import com.byit.enums.CallMethodEnum;
import com.byit.init.PluginClientInitialization;
import com.byit.registry.DataSourceServiceRegistry;
import com.byit.registry.PluginServiceRegistry;
......@@ -13,12 +13,12 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.InstantiationAwareBeanPostProcessorAdapter;
import org.springframework.util.ReflectionUtils;
/**
* spring 方式实现的插件端 客户端 实现工厂
*
* @author huangfu
*/
public class PluginSpringClientFactory extends InstantiationAwareBeanPostProcessorAdapter implements InitializingBean, DisposableBean, BeanFactoryAware {
......@@ -32,14 +32,13 @@ public class PluginSpringClientFactory extends InstantiationAwareBeanPostProcess
private String biz;
public PluginSpringClientFactory(String registryUrl, String env, String biz) {
this(null,null);
this(null, null);
this.registryUrl = registryUrl;
this.env = env;
this.biz = biz;
}
public PluginSpringClientFactory(Class<? extends PluginServiceRegistry> pluginServiceRegistryClass,
Class<? extends PluginClient> pluginClientClass) {
this.pluginServiceRegistryClass = pluginServiceRegistryClass;
......@@ -56,16 +55,18 @@ public class PluginSpringClientFactory extends InstantiationAwareBeanPostProcess
/**
* 属性设置之后调用
*
* @throws Exception
*/
@Override
public void afterPropertiesSet() throws Exception {
pluginClientFactory = new PluginClientFactory(pluginServiceRegistryClass,registryUrl,biz,env);
pluginClientFactory = new PluginClientFactory(pluginServiceRegistryClass, registryUrl, biz, env);
pluginClientFactory.start();
}
/**
* bean进行实例化后的后期处理 能够拿到每一个bean的实例 插手bean的实例化后的属性注入
*
* @param bean
* @param beanName
* @return
......@@ -75,7 +76,7 @@ public class PluginSpringClientFactory extends InstantiationAwareBeanPostProcess
public boolean postProcessAfterInstantiation(final Object bean, final String beanName) throws BeansException {
Class<?> beanClass = bean.getClass();
//寻找bean中每一个带有注解的操作 对其进行代理
ReflectionUtils.doWithFields(beanClass,field -> {
ReflectionUtils.doWithFields(beanClass, field -> {
if (field.isAnnotationPresent(TaskClient.class)) {
//获取改属性的接口
Class<?> iFace = field.getType();
......@@ -87,19 +88,21 @@ public class PluginSpringClientFactory extends InstantiationAwareBeanPostProcess
String callbackUrl = taskClientAnn.callbackUrl();
LoadBalance loadBalance = taskClientAnn.loadBalance();
long timeout = taskClientAnn.timeout();
CallMethodEnum callMethodEnum = taskClientAnn.callMethodEnum();
PluginClientInitialization pluginClientInitialization = new PluginClientInitialization(callbackUrl, timeout, loadBalance,
pluginClientFactory, iFace,pluginClientClass);
pluginClientFactory, iFace, pluginClientClass, callMethodEnum);
Object proxyObject = pluginClientInitialization.getObject();
field.setAccessible(true);
field.set(bean,proxyObject);
field.set(bean, proxyObject);
}
});
return super.postProcessAfterInstantiation(bean,beanName);
return super.postProcessAfterInstantiation(bean, beanName);
}
/**
* 能够获取bean工厂
*
* @param beanFactory
* @throws BeansException
*/
......@@ -110,6 +113,7 @@ public class PluginSpringClientFactory extends InstantiationAwareBeanPostProcess
/**
* bean被销毁时 自动调用
*
* @throws Exception
*/
@Override
......
byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/init/PluginClientInitialization.java
......@@ -2,6 +2,7 @@ package com.byit.init;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.client.PluginClient;
import com.byit.enums.CallMethodEnum;
import com.byit.factory.PluginClientFactory;
import com.byit.future.PluginFutureResponse;
import com.byit.packet.request.PluginRpcRequestPacket;
......@@ -18,6 +19,7 @@ import java.util.concurrent.TimeUnit;
/**
* 插件端 客户端的数据初始化
*
* @author huangfu
*/
public class PluginClientInitialization {
......@@ -30,20 +32,24 @@ public class PluginClientInitialization {
private PluginClient pluginClient;
private CallMethodEnum callMethodEnum;
public PluginClientInitialization(String callbackUrl, long timeout, LoadBalance loadBalance, PluginClientFactory pluginClientFactory, Class<?> iFace,Class<? extends PluginClient> pluginClientClass) {
public PluginClientInitialization(String callbackUrl, long timeout, LoadBalance loadBalance,
PluginClientFactory pluginClientFactory, Class<?> iFace, Class<? extends PluginClient> pluginClientClass,
CallMethodEnum callMethodEnum) {
this.callbackUrl = callbackUrl;
this.timeout = timeout;
this.loadBalance = loadBalance;
this.pluginClientFactory = pluginClientFactory;
this.iFace = iFace;
this.pluginClientClass = pluginClientClass;
this.callMethodEnum = callMethodEnum;
initClient();
}
private void initClient(){
private void initClient() {
try {
pluginClient = this.pluginClientClass.newInstance();
pluginClient.init(this);
......@@ -55,47 +61,44 @@ public class PluginClientInitialization {
/**
* 动态代理 生成动态代理的方法实现 ，并将动态代理生成的实现返还回调用方！
*
* @return 动态代理生成的实现类
*/
public Object getObject(){
public Object getObject() {
return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
new Class[]{this.iFace},
(proxy, method, args) ->{
(proxy, method, args) -> {
PluginRpcRequestPacket pluginRpcRequestPacket = (PluginRpcRequestPacket)args[0];
if(pluginRpcRequestPacket == null){
PluginRpcRequestPacket pluginRpcRequestPacket = (PluginRpcRequestPacket) args[0];
if (pluginRpcRequestPacket == null) {
throw new RuntimeException("服务参数为null:【com.byit.packet.request.PluginRpcRequestPacket】!");
}
PluginServiceRegistry pluginServiceRegistry = pluginClientFactory.getPluginServiceRegistry();
//服务器的注册ip:port
String address = null;
TreeSet<String> discovery = pluginServiceRegistry.discovery(pluginRpcRequestPacket.getJobName());
if(CollectionUtil.isNotEmpty(discovery)){
address = ClientRpcUtil.selectRPCServer(pluginClient,loadBalance,discovery,pluginRpcRequestPacket);
// if(discovery.size() ==1){
// address = discovery.first();
// }else{
// address = loadBalance.rpcInvokerRouter.route(pluginRpcRequestPacket.getJobName(), discovery);
// }
if (CollectionUtil.isNotEmpty(discovery)) {
address = ClientRpcUtil.selectRPCServer(pluginClient, loadBalance, discovery, pluginRpcRequestPacket);
}
if(StringUtils.isBlank(address)){
if (StringUtils.isBlank(address)) {
throw new RuntimeException("该服务在注册表中不存在！");
}
pluginRpcRequestPacket.setRequestId(UUID.randomUUID().toString());
pluginRpcRequestPacket.setCallMethodEnum(callMethodEnum);
PluginFutureResponse pluginFutureResponse = new PluginFutureResponse(pluginClientFactory,pluginRpcRequestPacket,null);
PluginFutureResponse pluginFutureResponse = new PluginFutureResponse(pluginClientFactory, pluginRpcRequestPacket, null);
try {
//发送请求
pluginClient.send(address,pluginRpcRequestPacket);
pluginClient.send(address, pluginRpcRequestPacket);
//获取结果
PluginRpcResponsePacket pluginRpcResponsePacket = pluginFutureResponse.get(timeout, TimeUnit.MILLISECONDS);
pluginRpcResponsePacket.setRunIp(address);
return pluginRpcResponsePacket;
}catch (Exception e){
} catch (Exception e) {
throw new RuntimeException(e);
}finally {
} finally {
pluginFutureResponse.remove(pluginRpcRequestPacket.getRequestId());
}
......
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/netty/handler/NettyPluginServerHandler.java
......@@ -3,6 +3,7 @@ package com.byit.server.netty.handler;
import cn.hutool.http.HttpRequest;
import com.alibaba.fastjson.JSON;
import com.byit.dto.web.ReturnResult;
import com.byit.enums.CallMethodEnum;
import com.byit.enums.JobResultEnum;
import com.byit.enums.ResponseTyEnum;
import com.byit.factory.PluginServerFactory;
......@@ -45,6 +46,76 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
if(PluginBeat.BEAT_ID.equals(msg.getRequestId())){
return;
}
CallMethodEnum callMethodEnum = msg.getCallMethodEnum();
CallMethod callMethod = null;
//同步调用
if(CallMethodEnum.SYNCHRONIZE.equals(callMethodEnum)){
callMethod = new Synchronize();
callMethod.run(msg,ctx);
}
//异步调用
if(CallMethodEnum.ASYNCHRONOUS.equals(callMethodEnum)){
callMethod = new Asynchronous();
callMethod.run(msg,ctx);
}
}
/**
* 同步调用
* @author huangfu
* @date 2020年11月26日11:22:11
*/
private class Synchronize implements CallMethod{
/**
* 开始执行
*/
@Override
public void run(PluginRpcRequestPacket msg,ChannelHandlerContext ctx) {
PluginRpcResponsePacket rpcResponsePacket = new PluginRpcResponsePacket();
try {
Map<String, Object> serverPoll = pluginServerFactory.getServerPoll();
String jobName = msg.getJobName();
Object bean = serverPoll.get(jobName);
IJobHandler iJobHandler = (IJobHandler)bean;
long startTime = System.currentTimeMillis();
ReturnResult<String> execute = iJobHandler.execute(msg.getParam());
long endTime = System.currentTimeMillis();
rpcResponsePacket.setResult(execute);
rpcResponsePacket.setCode(JobResultEnum.SUCCESS.getCode());
rpcResponsePacket.setMsg(JobResultEnum.SUCCESS.getMsg());
rpcResponsePacket.setStatus(true);
rpcResponsePacket.setRunTime(endTime-startTime);
rpcResponsePacket.setExtension(msg.getExtension());
}catch (Throwable e){
rpcResponsePacket.setCode(JobResultEnum.FAIL.getCode());
rpcResponsePacket.setMsg(PluginLogUtils.getMessage(e));
rpcResponsePacket.setStatus(false);
rpcResponsePacket.setExtension(msg.getExtension());
ReturnResult<String> execute = new ReturnResult<>();
execute.setMsg(PluginLogUtils.getMessage(e));
execute.setCode(JobResultEnum.FAIL.getCode());
rpcResponsePacket.setResult(execute);
throw new RuntimeException(e);
}
rpcResponsePacket.setType(ResponseTyEnum.RESPONSE.getType());
ctx.channel().writeAndFlush(rpcResponsePacket);
}
}
/**
* 异步调用
* @author huangfu
* @date 2020年11月26日11:22:11
*/
private class Asynchronous implements CallMethod{
/**
* 开始执行
*/
@Override
public void run(PluginRpcRequestPacket msg, ChannelHandlerContext ctx) {
PluginRpcResponsePacket transferPluginRpcResponse = new PluginRpcResponsePacket();
threadPoolExecutor.execute(()->{
PluginRpcResponsePacket rpcResponsePacket = new PluginRpcResponsePacket();
......@@ -86,6 +157,14 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
transferPluginRpcResponse.setType(ResponseTyEnum.TRANSFER.getType());
ctx.channel().writeAndFlush(transferPluginRpcResponse);
}
}
private interface CallMethod{
/**
* 开始执行
*/
void run(PluginRpcRequestPacket pluginRpcRequestPacket,ChannelHandlerContext ctx);
}
/**
* 发送信息
......
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