Skip to content

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

解决多网卡获取IP问题 解决补批BUG

parent 9cf28fea
Showing with 171 additions and 22 deletions
byit-myth-admin/src/main/java/com/byit/api/ApiFlowController.java
......@@ -172,6 +172,18 @@ public class ApiFlowController {
return ResponseResult.ok("SUCCESS");
}
/**
* 补批工作流
*
* @param repairFlow
* @return
*/
@PostMapping("/repairFlow")
@ApiOperation("补批工作流")
public ResponseResult repairFlow(@RequestBody RepairFlow repairFlow) {
apiFlowService.repairFlow(repairFlow);
return ResponseResult.ok("SUCCESS");
}
/**
* 获取整体运行的统计数据
......
byit-myth-admin/src/main/java/com/byit/service/ApiFlowService.java
......@@ -74,6 +74,7 @@ public interface ApiFlowService {
*/
void repairJob(String param);
void repairFlow(RepairFlow repairFlow);
/**
* 加载运行日志
......
byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
......@@ -10,6 +10,7 @@ import com.byit.dto.api.DeleteDto;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.plugin.*;
import com.byit.dto.specials.RepairFlow;
import com.byit.dto.specials.RepairTimeParam;
import com.byit.enums.*;
import com.byit.enums.plugin.PluginNodeTypeEnum;
import com.byit.job.utils.CronExpression;
......@@ -1282,6 +1283,87 @@ public class ApiFlowServiceImpl implements ApiFlowService {
}
}
@Override
public void repairFlow(RepairFlow repairFlow) {
ValidationUtil.dataNotNull(repairFlow, "请求参数不允许为空！");
//获取工作空间名称
String workspaceName = repairFlow.getWorkspaceName();
ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空！");
//获取工作流名称
String flowName = repairFlow.getFlowName();
ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
List<String> nodeNameList = repairFlow.getNodeNames();
ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(nodeNameList), "补批节点不允许为空！");
//获取补批的日期
List<RepairTimeParam> repairTimes = repairFlow.getRepairTimes();
List<String> repairTimeList = repairTimes.stream().map(RepairTimeParam::getRepairTime).collect(Collectors.toList());
ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(repairTimeList), "补批日期不允许为空！");
String timeFormatName = repairFlow.getTimeFormatName();
ValidationUtil.dataNotBank(timeFormatName, "补批时间格式不允许为空！");
String nowDateTimeFormatName = repairFlow.getNowDateTimeFormatName();
ValidationUtil.dataNotBank(nowDateTimeFormatName, "nowDate时间格式不允许为空！");
//开始校验
Workspace workspace = workspaceMapper.getByName(workspaceName);
ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
List<Node> nodeList = nodeMapper.findByFlowIdAndName(flow.getFlowId(), nodeNameList);
try {
RedissLockUtil.trlock(flow.getFlowId().toString(), 5);
//设置触发时间
Long triggerTime = System.currentTimeMillis();
Map<String, List<WaitingTask>> waitTaskMap = buildTask(nodeList, nodeNameList, repairTimeList, triggerTime, flowName,
timeFormatName, nowDateTimeFormatName);
//将时间排序
Collections.sort(repairTimeList);
//查看当前排队的工作流最大排队序号
Integer order = waitingRecordMapper.findOrderByFlowId(flow.getFlowId());
if (order == null) {
order = 0;
}
String userName = currentUserUtils.account();
for (String repairTime : repairTimeList) {
WaitingRecord waitingRecord = new WaitingRecord();
RunRecording runRecording = new RunRecording();
String runId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
//设置排期
BeanUtils.copyProperties(flow, waitingRecord);
waitingRecord.setFlowVersionName(flow.getVersionName());
waitingRecord.setRunId(runId);
waitingRecord.setFlowNodeCount(nodeList.size());
waitingRecord.setOperator(userName);
waitingRecord.setScheduleType(ScheduleTypeEnum.REPAIR.getCode());
waitingRecord.setRepeatTime(repairTime);
waitingRecord.setWaitOrder(++order);
//需要按着时间先后来设置时间
waitingRecord.setTriggerTime(System.currentTimeMillis());
//设置实例
BeanUtils.copyProperties(waitingRecord, runRecording);
runRecording.setFlowStatus("1");
runRecording.setTriggerTime(System.currentTimeMillis());
runRecording.setWorkspaceId(flow.getWorkspaceId());
Integer waitId = waitingRecordMapper.insertSelective(waitingRecord);
runRecordingMapper.saveRunRecording(runRecording);
List<WaitingTask> repairTaskList = waitTaskMap.get(repairTime);
repairTaskList.forEach(waitingTask -> {
waitingTask.setWaitId(waitingRecord.getWaitId());
waitingTaskMapper.insertSelective(waitingTask);
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
RedissLockUtil.unlock(flow.getFlowId().toString());
}
}
private Map<String, List<WaitingTask>> buildTask(List<Node> nodeList, List<String> nodeNameList, List<String> repairTimeList,
Long triggerTime, String flowName, String timeFormatName, String nowDateTimeFormatName) {
......
byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java
......@@ -143,6 +143,8 @@ public class JobUtils {
* 补批工作流
*/
public static final String REQUEST_REPAIRFLOW = "/api/operating/node/routineSupplementBatch";
public static final String REQUEST_REPAIRFLOW_COPY = "/api/flow/repairFlow";
/**
* 特殊的补批接口
*/
......@@ -666,7 +668,7 @@ public class JobUtils {
* @return 结果
*/
public static ResponseResult repairFlow(RepairFlow repairFlow) {
String response = createHttpRequest(REQUEST_REPAIRFLOW, JSON.toJSONString(repairFlow, WriteClassName));
String response = createHttpRequest(REQUEST_REPAIRFLOW_COPY, JSON.toJSONString(repairFlow, WriteClassName));
log.debug("--------------------补批工作流接口调用成功，结果为：{}------------------------", response);
return JSON.parseObject(response, ResponseResult.class);
}
......
byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/utils/IpUtil.java
package com.byit.utils;
import org.apache.commons.lang3.StringUtils;
import java.io.IOException;
import java.net.Inet6Address;
import java.net.InetAddress;
......@@ -22,21 +24,34 @@ public class IpUtil {
// ---------------------- valid ----------------------
/**
* valid Inet4Address
*
* @param address
* @return
*/
private static boolean isValidAddress(InetAddress address) {
private static boolean isValidAddress(InetAddress address,String myIpPatternStr) {
if (address == null || address.isLoopbackAddress()) {
return false;
}
boolean flag = true;
String name = address.getHostAddress();
if(StringUtils.isNoneBlank(myIpPatternStr)) {
Pattern myIpPattern = Pattern.compile(myIpPatternStr);
flag = myIpPattern.matcher(name).matches();
}
return (name != null
&& !ANYHOST.equals(name)
&& !LOCALHOST.equals(name)
&& IP_PATTERN.matcher(name).matches());
&& IP_PATTERN.matcher(name).matches())
&& flag;
}
/**
* valid Inet4Address
*
* @param address
* @return
*/
private static boolean isValidAddress(InetAddress address) {
return isValidAddress(address, null);
}
/**
......@@ -90,6 +105,10 @@ public class IpUtil {
private static InetAddress getLocalAddress0() {
return getLocalAddress0(null);
}
private static InetAddress getLocalAddress0(String preferredNetworks) {
InetAddress localAddress = null;
try {
localAddress = InetAddress.getLocalHost();
......@@ -98,7 +117,7 @@ public class IpUtil {
if (isValidV6Address(address)){
return normalizeV6Address(address);
}
} else if (isValidAddress(localAddress)) {
} else if (isValidAddress(localAddress,preferredNetworks)) {
return localAddress;
}
} catch (Throwable e) {
......@@ -121,7 +140,7 @@ public class IpUtil {
if (isValidV6Address(v6Address)){
return normalizeV6Address(v6Address);
}
} else if (isValidAddress(address)) {
} else if (isValidAddress(address, preferredNetworks)) {
return address;
}
} catch (Throwable e) {
......@@ -139,18 +158,32 @@ public class IpUtil {
}
// ---------------------- tool ----------------------
public static void main(String[] args) {
InetAddress localAddress0 = getLocalAddress0();
String hostAddress = localAddress0.getHostAddress();
System.out.println(hostAddress);
}
// ---------------------- tool ----------------------
public static InetAddress getLocalAddress(){
return getLocalAddress(null);
}
/**
* Find first valid IP from local network card
* 从本地网卡查找第一个有效IP
*
* @return first valid local IP
* @return 第一个有效的本地IP
*/
public static InetAddress getLocalAddress() {
public static InetAddress getLocalAddress(String preferredNetworks) {
if (LOCAL_ADDRESS != null) {
return LOCAL_ADDRESS;
}
InetAddress localAddress = getLocalAddress0();
InetAddress localAddress = null;
if(StringUtils.isNoneBlank(preferredNetworks)){
localAddress = getLocalAddress0();
}else{
localAddress = getLocalAddress0(preferredNetworks);
}
LOCAL_ADDRESS = localAddress;
return localAddress;
}
......@@ -164,6 +197,10 @@ public class IpUtil {
return getLocalAddress().getHostAddress();
}
public static String getIp(String preferredNetworks){
return getLocalAddress(preferredNetworks).getHostAddress();
}
/**
* get ip:port
*
......
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/factory/PluginServerFactory.java
......@@ -55,7 +55,7 @@ public abstract class PluginServerFactory {
public PluginServerFactory() {
}
public void init(int corePoolSize, int maxPoolSize,Integer port, String registryUrl, String env, String biz,
public void init(int corePoolSize, int maxPoolSize,Integer port, String registryUrl, String env, String biz,String preferredNetworks,
Class<? extends PluginServiceRegistry> serviceRegistryClass,
Class<? extends PluginServer> pluginServerClass,
Class<? extends RemainingOperationsCallBack> remainingOperationsCallBack){
......@@ -74,7 +74,7 @@ public abstract class PluginServerFactory {
}else{
this.env = DEFAULT_ENV_NAME;
}
this.ip = IpUtil.getIp();
this.ip = IpUtil.getIp(preferredNetworks);
if(port <= 0){
this.port = 7060;
......
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/factory/RpcMainPluginServerFactory.java
......@@ -44,7 +44,7 @@ public class RpcMainPluginServerFactory extends PluginServerFactory {
List<String> classNames = serviceConfigModel.getClassNames();
super.init(serverConfigurationModel.getCoreSize(),serverConfigurationModel.getMaxSize(),
serverConfigurationModel.getPort(), serverConfigurationModel.getRegistryUrl(),
serverConfigurationModel.getServerEnv(), serverConfigurationModel.getServerBiz(),
serverConfigurationModel.getServerEnv(), serverConfigurationModel.getServerBiz(),null,
pluginServiceRegistryClass,pluginServerClass,remainingOperationsClass);
if (CollectionUtil.isNotEmpty(classNames)) {
......
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/factory/RpcSpringPluginServerFactory.java
......@@ -25,14 +25,20 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
private String biz;
private String env;
private int port;
private String preferredNetworks;
private Map<String,String> taskNameExpand = new ConcurrentHashMap<>(8);
public RpcSpringPluginServerFactory(String address, String biz, String env, int port) {
this(address, biz, env, port, null);
}
public RpcSpringPluginServerFactory(String address, String biz, String env, int port, String preferredNetworks) {
this.address = address;
this.biz = biz;
this.env = env;
this.port = port;
this.preferredNetworks = preferredNetworks;
}
@Override
......@@ -42,7 +48,7 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
@Override
public void afterPropertiesSet() throws Exception {
super.init(-1,-1,port,address,env,biz,null,null,null);
super.init(-1,-1,port,address,env,biz,preferredNetworks,null,null,null);
super.start();
}
......
byit-plugin-core/plugin-spring-boot-starter/src/main/java/com/byit/configuration/PluginAutoConfigure.java
......@@ -34,8 +34,7 @@ public class PluginAutoConfigure {
public RpcSpringPluginServerFactory pluginServerFactory(){
PluginConfigurationModel plugin = mythConfigModel.getPlugin();
RegisteredConfigurationModel register = mythConfigModel.getRegister();
RpcSpringPluginServerFactory rpcSpringPluginServerFactory =
new RpcSpringPluginServerFactory(register.getUrl(),plugin.getBiz(),plugin.getEnv(),plugin.getPort());
RpcSpringPluginServerFactory rpcSpringPluginServerFactory = new RpcSpringPluginServerFactory(register.getUrl(),plugin.getBiz(),plugin.getEnv(),plugin.getPort());
rpcSpringPluginServerFactory.setPluginServerClass(NettyPluginServer.class);
return rpcSpringPluginServerFactory;
}
......
byit-plugin-core/plugin-spring-boot-starter/src/main/java/com/byit/model/PluginConfigurationModel.java
......@@ -8,6 +8,8 @@ public class PluginConfigurationModel {
private String env;
private String biz;
private Integer port;
//网卡正则表达式
private String preferredNetworks;
public String getEnv() {
return env;
......@@ -32,4 +34,12 @@ public class PluginConfigurationModel {
public void setPort(Integer port) {
this.port = port;
}
public String getPreferredNetworks() {
return preferredNetworks;
}
public void setPreferredNetworks(String preferredNetworks) {
this.preferredNetworks = preferredNetworks;
}
}
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