Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
B
byit-myth-job
Overview
Overview
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
liyuan
byit-myth-job
Commits
4566b6da
Commit
4566b6da
authored
Mar 16, 2020
by
guominglei@byitgroup.com
Browse files
Options
Browse Files
Download
Plain Diff
Merge remote-tracking branch 'origin/developer' into developer
parents
ff80653f
e5e1e24a
Show whitespace changes
Inline
Side-by-side
Showing
13 changed files
with
212 additions
and
57 deletions
+212
-57
DemoRegisterConfig.java
...min/src/main/java/com/byit/config/DemoRegisterConfig.java
+3
-3
application-dev.yml
byit-myth-admin/src/main/resources/application-dev.yml
+1
-1
NodeRunStatusPropertyEnum.java
...c/main/java/com/byit/enums/NodeRunStatusPropertyEnum.java
+2
-1
RunNodeServiceImpl.java
...c/main/java/com/byit/service/impl/RunNodeServiceImpl.java
+16
-7
JavaBeanJobTask.java
...min-core/src/main/java/com/byit/task/JavaBeanJobTask.java
+6
-1
ScriptExecutorJobTask.java
...re/src/main/java/com/byit/task/ScriptExecutorJobTask.java
+5
-1
ScheduleThreadRunHelper.java
.../java/com/byit/thread/helper/ScheduleThreadRunHelper.java
+9
-0
TaskThreadRunHelper.java
...main/java/com/byit/thread/helper/TaskThreadRunHelper.java
+23
-12
ServiceInfoUtil.java
...min-core/src/main/java/com/byit/util/ServiceInfoUtil.java
+41
-0
PlaceholderEnum.java
...mon/src/main/java/com/byit/job/enums/PlaceholderEnum.java
+24
-0
PlaceholderUtils.java
...on/src/main/java/com/byit/job/utils/PlaceholderUtils.java
+67
-29
ScriptExecutorServiceImpl.java
...main/java/com/byit/service/ScriptExecutorServiceImpl.java
+15
-1
ScriptAddFlow.java
...demo-client/src/main/java/com/byit/job/ScriptAddFlow.java
+0
-1
No files found.
byit-myth-admin/src/main/java/com/byit/config/DemoRegisterConfig.java
View file @
4566b6da
...
...
@@ -26,8 +26,8 @@ public class DemoRegisterConfig {
private
String
biz
;
@Value
(
"${myth-rpc.registry.env}"
)
private
String
env
;
@Value
(
"${myth-rpc.registry.port}"
)
private
int
port
;
//
@Value("${myth-rpc.registry.port}")
//
private int port;
@Bean
public
RpcSpringInvokerFactory
invokerFactory
(){
...
...
@@ -46,7 +46,7 @@ public class DemoRegisterConfig {
@Bean
public
RpcSpringProviderFactory
rpcSpringProviderFactory
()
{
RpcSpringProviderFactory
providerFactory
=
new
RpcSpringProviderFactory
();
providerFactory
.
setPort
(
port
);
//
providerFactory.setPort(port);
providerFactory
.
setServiceRegistryClass
(
RegistryServiceRegistry
.
class
);
providerFactory
.
setServiceRegistryParam
(
new
HashMap
<
String
,
String
>()
{{
put
(
RegistryServiceRegistry
.
REGISTRY_ADDRESS
,
address
);
...
...
byit-myth-admin/src/main/resources/application-dev.yml
View file @
4566b6da
...
...
@@ -22,7 +22,7 @@ myth-rpc:
address
:
http://localhost:8080/myth-register
env
:
huangfu
biz
:
byit-myth-job
port
:
6665
#
port: 6665
logging
:
path
:
/data/mythjob
file
:
myth_log_file
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/NodeRunStatusPropertyEnum.java
View file @
4566b6da
...
...
@@ -11,7 +11,8 @@ public enum NodeRunStatusPropertyEnum {
RE_RUN_SUCCESS
(
"3"
,
"补批成功"
),
RE_RUN_FAILURE
(
"4"
,
"补批失败"
),
KILL
(
"5"
,
"kill"
),
PARENT_NODE_FAILED
(
"6"
,
"上级节点执行失败"
)
PARENT_NODE_FAILED
(
"6"
,
"上级节点执行失败"
),
NODE_RELY_ERROR
(
"7"
,
"节点依赖错误"
),
;
private
String
code
;
private
String
msg
;
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunNodeServiceImpl.java
View file @
4566b6da
package
com
.
byit
.
service
.
impl
;
import
com.byit.enums.
FlowProperty
Enum
;
import
com.byit.enums.
NodeName
Enum
;
import
com.byit.event.FlowScanEndEvent
;
import
com.byit.job.utils.CronExpression
;
import
com.byit.model.Flow
;
import
com.byit.model.JobTask
;
import
com.byit.model.Node
;
import
com.byit.model.RunRecording
;
import
com.byit.service.FlowService
;
import
com.byit.service.JobTaskService
;
import
com.byit.service.RunNodeServer
;
import
com.byit.service.RunRecordingService
;
import
com.byit.service.*
;
import
lombok.extern.slf4j.Slf4j
;
import
org.apache.commons.lang3.StringUtils
;
import
org.springframework.beans.BeanUtils
;
import
org.springframework.context.ApplicationEventPublisher
;
import
org.springframework.context.ApplicationEventPublisherAware
;
...
...
@@ -36,12 +34,17 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
private
final
JobTaskService
jobTaskService
;
private
final
FlowService
flowService
;
private
ApplicationEventPublisher
applicationEventPublisher
;
/**
* 节点依赖查询操作
*/
private
final
NodeDependencyService
nodeDependencyService
;
public
RunNodeServiceImpl
(
RunRecordingService
runRecordingService
,
JobTaskService
jobTaskService
,
FlowService
flowService
)
{
FlowService
flowService
,
NodeDependencyService
nodeDependencyService
)
{
this
.
runRecordingService
=
runRecordingService
;
this
.
jobTaskService
=
jobTaskService
;
this
.
flowService
=
flowService
;
this
.
nodeDependencyService
=
nodeDependencyService
;
}
/**
...
...
@@ -58,7 +61,7 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
BeanUtils
.
copyProperties
(
flow
,
build
);
build
.
setRunId
(
runId
);
//TODO 这个不解释 不知道干嘛的 后续需要修改
build
.
setDispatchIp
(
"
127.0.0.1
"
);
build
.
setDispatchIp
(
"
0.0.0.0
"
);
build
.
setFlowVersionName
(
flow
.
getVersionName
());
build
.
setTriggerTime
(
flow
.
getTriggerNextTime
());
runRecordingService
.
saveRunRecording
(
build
);
...
...
@@ -75,6 +78,12 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
}
jobTask
.
setRunId
(
runId
);
jobTask
.
setFlowName
(
flow
.
getFlowName
());
//如果不是开始节点
if
(!
NodeNameEnum
.
START_NODE
.
getNodeName
().
equals
(
jobTask
.
getNodeName
())){
List
<
Integer
>
dependIdByNodeId
=
nodeDependencyService
.
findDependIdByNodeId
(
jobTask
.
getNodeId
());
String
parentIds
=
StringUtils
.
join
(
dependIdByNodeId
,
","
);
jobTask
.
setNodeDepend
(
parentIds
);
}
jobTasks
.
add
(
jobTask
);
});
jobTaskService
.
saveJobTasks
(
jobTasks
);
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java
View file @
4566b6da
...
...
@@ -17,6 +17,7 @@ import com.byit.rpc.remoting.invoker.route.RpcLoadBalance;
import
com.byit.service.impl.JobTaskRunLogServiceImpl
;
import
com.byit.service.impl.RunRecordingServiceImpl
;
import
com.byit.util.ServiceInfoUtil
;
import
com.byit.util.SpringUtil
;
import
io.netty.util.Timeout
;
import
io.netty.util.TimerTask
;
...
...
@@ -32,6 +33,10 @@ import java.util.Date;
**/
@Slf4j
public
class
JavaBeanJobTask
implements
TimerTask
{
private
final
String
HTTP_PRE
=
"http://"
;
private
final
String
HTTP_SUFFIX
=
"/job/callbackRes"
;
private
JobTaskSchedule
mythJobTaskSchedule
;
public
JavaBeanJobTask
(
JobTaskSchedule
mythJobTaskSchedule
)
{
this
.
mythJobTaskSchedule
=
mythJobTaskSchedule
;
...
...
@@ -79,7 +84,7 @@ public class JavaBeanJobTask implements TimerTask {
String
param
=
mythJobTaskSchedule
.
getRunParam
(
);
String
runId
=
mythJobTaskSchedule
.
getRunId
();
AdminSenPluginDto
adminSenPluginDto
=
new
AdminSenPluginDto
();
adminSenPluginDto
.
setCallbackUrl
(
"http://127.0.0.1:8998/job/callbackRes"
);
adminSenPluginDto
.
setCallbackUrl
(
HTTP_PRE
+
ServiceInfoUtil
.
getIpAndPort
()+
HTTP_SUFFIX
);
adminSenPluginDto
.
setJobHandelName
(
jobHandelName
);
adminSenPluginDto
.
setJobParam
(
param
);
adminSenPluginDto
.
setRunId
(
runId
);
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java
View file @
4566b6da
...
...
@@ -14,6 +14,7 @@ import com.byit.rpc.util.RpcException;
import
com.byit.service.RunScriptService
;
import
com.byit.service.impl.JobTaskRunLogServiceImpl
;
import
com.byit.service.impl.RunRecordingServiceImpl
;
import
com.byit.util.ServiceInfoUtil
;
import
com.byit.util.SpringUtil
;
import
io.netty.util.Timeout
;
import
io.netty.util.TimerTask
;
...
...
@@ -27,6 +28,9 @@ import java.util.Date;
@Slf4j
public
class
ScriptExecutorJobTask
implements
TimerTask
{
private
final
String
HTTP_PRE
=
"http://"
;
private
final
String
HTTP_SUFFIX
=
"/job/callbackRes"
;
private
JobTaskSchedule
mythJobTaskSchedule
;
public
ScriptExecutorJobTask
(
JobTaskSchedule
mythJobTaskSchedule
)
{
...
...
@@ -90,7 +94,7 @@ public class ScriptExecutorJobTask implements TimerTask {
scriptDto
.
setParam
(
mythJobTaskSchedule
.
getRunParam
());
scriptDto
.
setRunId
(
mythJobTaskSchedule
.
getRunId
());
scriptDto
.
setRemotePath
(
mythJobTaskSchedule
.
getScriptUrls
());
scriptDto
.
setCallbackUrl
(
"http://127.0.0.1:8998/job/callbackRes"
);
scriptDto
.
setCallbackUrl
(
HTTP_PRE
+
ServiceInfoUtil
.
getIpAndPort
()+
HTTP_SUFFIX
);
//二次执行的情况下 会有这个信息
scriptDto
.
setLogRemotePath
(
jobTaskRunLogById
.
getLogRemotelyPath
());
DispatchResponseDto
dispatchResponseDto
=
new
DispatchResponseDto
();
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ScheduleThreadRunHelper.java
View file @
4566b6da
package
com
.
byit
.
thread
.
helper
;
import
cn.hutool.core.collection.CollectionUtil
;
import
com.alibaba.fastjson.JSON
;
import
com.byit.enums.NodeTypeEnum
;
import
com.byit.job.WorkRoulette
;
import
com.byit.job.dto.ScriptParamAndPlaceholderDto
;
import
com.byit.job.enums.PlaceholderEnum
;
import
com.byit.job.utils.DateUtil
;
import
com.byit.job.utils.PlaceholderUtils
;
import
com.byit.model.JobTaskRunLogWithBLOBs
;
import
com.byit.model.JobTaskSchedule
;
import
com.byit.service.JobTaskScheduleService
;
...
...
@@ -16,7 +21,9 @@ import lombok.extern.slf4j.Slf4j;
import
org.springframework.stereotype.Component
;
import
javax.sql.DataSource
;
import
java.util.Date
;
import
java.util.List
;
import
java.util.Map
;
/**
* 排期表操作
...
...
@@ -48,6 +55,7 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
log
.
info
(
"------排期表查询到有需要存在的节点--------"
);
//循环遍历添加任务
jobTaskSchedules
.
forEach
(
mythJobTaskSchedule
->{
mythJobTaskSchedule
.
setRunParam
(
PlaceholderUtils
.
formatParam
(
mythJobTaskSchedule
.
getRunParam
()));
//如果是重跑就有logId
Integer
logId
=
mythJobTaskSchedule
.
getLogId
();
if
(
logId
==
null
){
...
...
@@ -108,4 +116,5 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
return
jobTaskRunLog
.
getLogId
();
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/TaskThreadRunHelper.java
View file @
4566b6da
...
...
@@ -5,19 +5,20 @@ import com.byit.enums.FlowPropertyEnum;
import
com.byit.enums.NodeNameEnum
;
import
com.byit.enums.NodePropertyEnum
;
import
com.byit.enums.NodeRunStatusPropertyEnum
;
import
com.byit.job.exceptions.BusinessException
;
import
com.byit.model.JobTask
;
import
com.byit.model.JobTaskRunLog
;
import
com.byit.model.JobTaskSchedule
;
import
com.byit.model.RunRecording
;
import
com.byit.service.JobTaskRunLogService
;
import
com.byit.service.JobTaskService
;
import
com.byit.service.NodeDependencyService
;
import
com.byit.service.RunRecordingService
;
import
com.byit.service.mapservice.RunRecordingAndJobTaskService
;
import
com.byit.service.mapservice.TaskAndLogServer
;
import
com.byit.service.mapservice.TaskAndScheduleService
;
import
com.byit.thread.BaseThreadRunHelper
;
import
lombok.extern.slf4j.Slf4j
;
import
org.apache.commons.lang3.StringUtils
;
import
org.springframework.beans.BeanUtils
;
import
org.springframework.stereotype.Component
;
...
...
@@ -49,10 +50,7 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
* 运行记录表信息操作
*/
private
final
RunRecordingService
runRecordingService
;
/**
* 节点依赖查询操作
*/
private
final
NodeDependencyService
nodeDependencyService
;
/**
* 日志节点操作
*/
...
...
@@ -68,26 +66,28 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
public
TaskThreadRunHelper
(
DataSource
dataSource
,
JobTaskService
jobTaskService
,
RunRecordingAndJobTaskService
runRecordingAndJobTaskService
,
RunRecordingService
runRecordingService
,
NodeDependencyService
nodeDependencyService
,
JobTaskRunLogService
jobTaskRunLogService
,
TaskAndLogServer
taskAndLogServer
,
TaskAndScheduleService
taskAndScheduleService
)
{
RunRecordingService
runRecordingService
,
JobTaskRunLogService
jobTaskRunLogService
,
TaskAndLogServer
taskAndLogServer
,
TaskAndScheduleService
taskAndScheduleService
)
{
this
.
dataSource
=
dataSource
;
this
.
jobTaskService
=
jobTaskService
;
this
.
runRecordingAndJobTaskService
=
runRecordingAndJobTaskService
;
this
.
runRecordingService
=
runRecordingService
;
this
.
nodeDependencyService
=
nodeDependencyService
;
this
.
jobTaskRunLogService
=
jobTaskRunLogService
;
this
.
taskAndLogServer
=
taskAndLogServer
;
this
.
taskAndScheduleService
=
taskAndScheduleService
;
}
/**
* 业务逻辑
* @return 睡眠时间
*/
@Override
public
Long
start
()
{
long
nowTime
=
System
.
currentTimeMillis
();
//开始寻找此时 不是暂停状态,而且七秒内即将运行的任务 而且还不是暂停的节点
List
<
JobTask
>
jobTasks
=
jobTaskService
.
findJobTaskByTriggerNextTimeLessThanEqual
(
nowTime
+
PRE_READ_MS
);
if
(
CollectionUtil
.
isNotEmpty
(
jobTasks
))
{
List
<
JobTaskSchedule
>
jobTaskSchedules
=
new
ArrayList
<
JobTaskSchedule
>(
15
);
List
<
JobTaskSchedule
>
jobTaskSchedules
=
new
ArrayList
<>(
15
);
//遍历七秒内将要运行的节点数据
for
(
JobTask
jobTask
:
jobTasks
){
log
.
debug
(
"任务:{},开始运行"
,
jobTask
);
...
...
@@ -130,10 +130,21 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
*/
private
void
nodeOperating
(
JobTask
thisJobTask
,
List
<
JobTaskSchedule
>
jobTaskSchedules
){
//查询该节点的依赖节点
List
<
Integer
>
dependIdByNodeId
=
nodeDependencyService
.
findDependIdByNodeId
(
thisJobTask
.
getNodeId
());
List
<
Integer
>
dependIdByNodeId
=
null
;
if
(
StringUtils
.
isNotBlank
(
thisJobTask
.
getNodeDepend
())){
dependIdByNodeId
=
new
ArrayList
<>();
String
[]
parentIds
=
thisJobTask
.
getNodeDepend
().
split
(
","
);
for
(
String
parentId
:
parentIds
)
{
dependIdByNodeId
.
add
(
Integer
.
parseInt
(
parentId
));
}
}
//这里返回的是上级节点的日志执行情况 把运行中的数据给过滤掉了
List
<
JobTaskRunLog
>
jobTaskRunLogList
=
jobTaskRunLogService
.
findJobTaskRunLogNotEndNodeByRunCodeCount
(
dependIdByNodeId
,
thisJobTask
.
getRunId
());
if
(
CollectionUtil
.
isNotEmpty
(
jobTaskRunLogList
))
{
if
(
CollectionUtil
.
isEmpty
(
dependIdByNodeId
)){
throw
new
BusinessException
(
NodeRunStatusPropertyEnum
.
NODE_RELY_ERROR
.
getMsg
());
}
//判断父类节点是否已经全部完成,只需要判断依赖节点的数目和查询出来的日志数据是否相同
if
(
dependIdByNodeId
.
size
()
==
jobTaskRunLogList
.
size
()){
//过滤失败的节点
...
...
@@ -209,7 +220,7 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
/**
* 判断失败节点的重试次数是不是为0
* @param errorJobLog 上级节点的全部失败节点
* @return
* @return
失败节点是否有重试次数
*/
private
boolean
parentNodeErrorCount
(
List
<
JobTaskRunLog
>
errorJobLog
){
if
(
CollectionUtil
.
isEmpty
(
errorJobLog
)){
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/ServiceInfoUtil.java
0 → 100644
View file @
4566b6da
package
com
.
byit
.
util
;
import
lombok.extern.slf4j.Slf4j
;
import
org.springframework.boot.context.embedded.EmbeddedServletContainerInitializedEvent
;
import
org.springframework.context.ApplicationListener
;
import
org.springframework.stereotype.Component
;
import
org.springframework.util.Assert
;
import
java.net.InetAddress
;
import
java.net.UnknownHostException
;
/**
* 获取当前服务器的信息数据
* @author huangfu
*/
@Component
@Slf4j
public
class
ServiceInfoUtil
implements
ApplicationListener
<
EmbeddedServletContainerInitializedEvent
>
{
private
static
EmbeddedServletContainerInitializedEvent
event
;
@Override
public
void
onApplicationEvent
(
EmbeddedServletContainerInitializedEvent
embeddedServletContainerInitializedEvent
)
{
ServiceInfoUtil
.
event
=
embeddedServletContainerInitializedEvent
;
}
public
static
EmbeddedServletContainerInitializedEvent
getEvent
()
{
return
event
;
}
public
static
String
getIpAndPort
()
{
int
port
=
getEvent
().
getEmbeddedServletContainer
().
getPort
();
Assert
.
state
(
port
!=
-
1
,
"端口号获取失败"
);
InetAddress
address
=
null
;
try
{
address
=
InetAddress
.
getLocalHost
();
}
catch
(
UnknownHostException
e
)
{
e
.
printStackTrace
();
}
assert
address
!=
null
;
return
address
.
getHostAddress
()
+
":"
+
port
;
}
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/enums/PlaceholderEnum.java
0 → 100644
View file @
4566b6da
package
com
.
byit
.
job
.
enums
;
/**
* 替换参数枚举
* @author huangfu
*/
public
enum
PlaceholderEnum
{
DATE_PLACEHOLDER
(
"biz_date"
),
BIZ_SCRIPT_FILE
(
"biz_file"
)
;
private
String
name
;
PlaceholderEnum
(
String
name
)
{
this
.
name
=
name
;
}
PlaceholderEnum
()
{
}
public
String
getName
()
{
return
name
;
}
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/utils/PlaceholderUtils.java
View file @
4566b6da
...
...
@@ -3,15 +3,10 @@ package com.byit.job.utils;
import
cn.hutool.core.collection.CollectionUtil
;
import
com.alibaba.fastjson.JSON
;
import
com.byit.job.dto.ScriptParamAndPlaceholderDto
;
import
lombok.extern.java.Log
;
import
com.byit.job.enums.PlaceholderEnum
;
import
lombok.extern.slf4j.Slf4j
;
import
org.apache.commons.lang3.StringUtils
;
import
java.io.BufferedReader
;
import
java.io.File
;
import
java.io.FileReader
;
import
java.io.IOException
;
import
java.nio.ByteBuffer
;
import
java.nio.charset.Charset
;
import
java.nio.charset.StandardCharsets
;
import
java.util.Date
;
import
java.util.HashMap
;
...
...
@@ -32,16 +27,12 @@ public class PlaceholderUtils {
* 占位符后缀
*/
private
static
final
String
PLACEHOLDER_SUFFIX
=
"}"
;
/**
* 补批时间参数占位符
*/
private
static
final
String
DATE_KEY
=
"biz_date"
;
/**
* 占位符数据替换
* @param scriptData
* @param parameter
* @return
* @param scriptData
脚本数据
* @param parameter
脚本参数
* @return
替换后的脚本参数
*/
public
static
byte
[]
resolvePlaceholders
(
byte
[]
scriptData
,
Map
<
String
,
String
>
parameter
){
//替换为null 不做操作
...
...
@@ -63,7 +54,7 @@ public class PlaceholderUtils {
//截取变量值
String
placeholder
=
sbt
.
substring
(
startIndex
+
PLACEHOLDER_PREFIX
.
length
(),
endIndex
);
//确定下一次查询的位置
int
nextIndex
=
endIndex
+
PLACEHOLDER_SUFFIX
.
length
()
;
int
nextIndex
;
//开始从 map中解析
if
(
parameter
.
containsKey
(
placeholder
)){
String
placeholderValue
=
parameter
.
get
(
placeholder
);
...
...
@@ -80,37 +71,84 @@ public class PlaceholderUtils {
return
sbt
.
toString
().
getBytes
(
StandardCharsets
.
UTF_8
);
}
/**
* 命令占位符替换
* @param command 命令
* @param param 参数
* @return 替换后的参数
*/
public
static
String
commandReplace
(
String
command
,
Map
<
String
,
String
>
param
){
if
(
StringUtils
.
isBlank
(
command
)
||
CollectionUtil
.
isEmpty
(
param
)){
log
.
info
(
"-----------命令{},不需要替换-----------"
,
command
);
return
command
;
}
byte
[]
bytes
=
resolvePlaceholders
(
command
.
getBytes
(),
param
);
return
new
String
(
bytes
,
StandardCharsets
.
UTF_8
);
}
/**
* 脚本时间参数替换
* @param runParam 运行参数
* @param thisDate 补批时间
* @return 替换后的参数
*/
public
static
String
paramPlaceholder
(
String
runParam
,
String
thisDate
){
ScriptParamAndPlaceholderDto
scriptParamAndPlaceholderDto
=
JSON
.
parseObject
(
runParam
,
ScriptParamAndPlaceholderDto
.
class
);
Map
<
String
,
String
>
placeholder
=
scriptParamAndPlaceholderDto
.
getPlaceholder
();
Map
<
String
,
String
>
param
=
scriptParamAndPlaceholderDto
.
getParam
();
//替换时间参数
if
(
CollectionUtil
.
isNotEmpty
(
placeholder
)
&&
placeholder
.
containsKey
(
PlaceholderEnum
.
DATE_PLACEHOLDER
.
getName
()))
{
placeholder
.
put
(
PlaceholderEnum
.
DATE_PLACEHOLDER
.
getName
(),
thisDate
);
}
//替换时间占位符
if
(
CollectionUtil
.
isNotEmpty
(
param
)
&&
param
.
containsKey
(
PlaceholderEnum
.
DATE_PLACEHOLDER
.
getName
())){
param
.
put
(
PlaceholderEnum
.
DATE_PLACEHOLDER
.
getName
(),
thisDate
);
}
return
JSON
.
toJSONString
(
scriptParamAndPlaceholderDto
);
}
/**
* 替换时间参数
* 初始化脚本信息
* @param command 基础命令
* @param scriptPath 脚本路径
* @return 基础命令
*/
if
(
CollectionUtil
.
isNotEmpty
(
placeholder
)
&&
placeholder
.
containsKey
(
DATE_KEY
))
{
placeholder
.
put
(
DATE_KEY
,
thisDate
);
public
static
String
initCommand
(
String
command
,
String
scriptPath
){
log
.
info
(
"----------替换路径{}------"
,
scriptPath
);
return
command
.
replace
(
PLACEHOLDER_PREFIX
+
PlaceholderEnum
.
BIZ_SCRIPT_FILE
.
getName
()+
PLACEHOLDER_SUFFIX
,
scriptPath
);
}
/**
* 替换时间占位符
* 格式化运行参数
* 如果参数内存在时间参数 就将时间参数改为
* @param runParam 运行参数
* @return 参数字符串
*/
if
(
CollectionUtil
.
isNotEmpty
(
param
)
&&
param
.
containsKey
(
DATE_KEY
)){
param
.
put
(
DATE_KEY
,
thisDate
);
public
static
String
formatParam
(
String
runParam
){
if
(
StringUtils
.
isBlank
(
runParam
))
{
return
null
;
}
ScriptParamAndPlaceholderDto
scriptParamAndPlaceholderDto
=
JSON
.
parseObject
(
runParam
,
ScriptParamAndPlaceholderDto
.
class
);
Map
<
String
,
String
>
param
=
scriptParamAndPlaceholderDto
.
getParam
();
if
(
CollectionUtil
.
isNotEmpty
(
param
)
&&
param
.
containsKey
(
PlaceholderEnum
.
DATE_PLACEHOLDER
.
getName
())){
String
calculationDate
=
DateUtil
.
dateLessDayStr
(
new
Date
(),
"yyyyMMdd"
,
1
);
param
.
put
(
PlaceholderEnum
.
DATE_PLACEHOLDER
.
getName
(),
calculationDate
);
}
return
JSON
.
toJSONString
(
scriptParamAndPlaceholderDto
);
}
public
static
void
main
(
String
[]
args
)
{
ScriptParamAndPlaceholderDto
sc
=
new
ScriptParamAndPlaceholderDto
();
/*
ScriptParamAndPlaceholderDto sc = new ScriptParamAndPlaceholderDto();
Map<String,String> m1 = new HashMap<>(2);
Map<String,String> m2 = new HashMap<>(2);
m1
.
put
(
DATE_KEY
,
"2018年12月12日"
);
m1.put(
PlaceholderEnum.DATE_PLACEHOLDER.getName()
,"2018年12月12日");
m1.put("name","狗子");
m2
.
put
(
DATE_KEY
,
"2010年10月10日"
);
m2.put(
PlaceholderEnum.DATE_PLACEHOLDER.getName()
,"2010年10月10日");
m2.put("age","1000");
...
...
@@ -118,13 +156,13 @@ public class PlaceholderUtils {
sc.setParam(m1);
String
toJSON
String
=
JSON
.
toJSONString
(
sc
);
String
json
String = JSON.toJSONString(sc);
String
s
=
PlaceholderUtils
.
paramPlaceholder
(
toJSONString
,
"2020/3/12 15:13"
);
String s = PlaceholderUtils.paramPlaceholder(
jsonString, "2020/3/12 15:13");*/
System
.
out
.
println
(
initCommand
(
"python ${biz_file}"
,
"D:/2020project/byit-myth-job/demo-client/byit-demo-client/src/main/java/com/byit/job/Mains.java"
));
System
.
out
.
println
(
s
);
}
}
byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java
View file @
4566b6da
...
...
@@ -73,8 +73,14 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
}
String
scriptPath
=
byteArrayToFile
(
remotePath
,
scriptParamAndPlaceholderDto
);
log
.
info
(
"-------------脚本所在路径为:{}-------------"
,
scriptPath
);
//初始化命令信息
command
=
PlaceholderUtils
.
initCommand
(
command
,
scriptPath
);
//开始执行脚本
List
<
String
>
cmdList
=
Arrays
.
asList
(
command
,
scriptPath
);
if
(
scriptParamAndPlaceholderDto
!=
null
){
command
=
PlaceholderUtils
.
commandReplace
(
command
,
scriptParamAndPlaceholderDto
.
getParam
());
}
List
<
String
>
cmdList
=
Arrays
.
asList
(
command
.
split
(
" "
));
MythJobProcess
mythJobProcess
=
new
MythJobProcess
(
cmdList
,
null
,
null
,
scriptDto
.
getLogId
());
//保存日志
String
logData
=
mythJobProcess
.
call
();
...
...
@@ -207,4 +213,12 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
return
date
.
format
(
DateTimeFormatter
.
ofPattern
(
"yyyyMMdd"
));
}
public
static
void
main
(
String
[]
args
)
throws
IOException
{
File
file
=
new
File
(
"D:\\2020project\\byit-myth-job\\demo-client\\byit-demo-client\\src\\main\\java\\com\\byit\\job\\Mains.java"
);
System
.
out
.
println
(
file
.
getPath
());
System
.
out
.
println
(
file
.
getAbsolutePath
());
System
.
out
.
println
(
file
.
getCanonicalPath
());
}
}
demo-client/byit-demo-client/src/main/java/com/byit/job/ScriptAddFlow.java
View file @
4566b6da
...
...
@@ -96,7 +96,6 @@ public class ScriptAddFlow {
pluginNode5
.
setAuthor
(
"皇甫"
);
pluginNode5
.
setJobType
(
"JAVA"
);
pluginNode5
.
setHandlerName
(
"END"
);
pluginNode5
.
setRunParam
(
"END"
);
pluginNodeConfig5
.
setFailedRetryCount
(
2
);
pluginNodeConfig5
.
setFailedRetryInterval
(
TimeUnit
.
MINUTES
.
toSeconds
(
2
));
pluginNodeConfig5
.
setNodeCron
(
"0 0/7 * * * ? *"
);
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment