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
22c83f51
Commit
22c83f51
authored
Jan 08, 2020
by
huangfusuper
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
【修改遗留】修改添加任务时,判断上级节点是否完成
parent
2bfb89a8
Hide whitespace changes
Inline
Side-by-side
Showing
8 changed files
with
118 additions
and
14 deletions
+118
-14
JobTaskRunLogMapper.java
...re/src/main/java/com/byit/mapper/JobTaskRunLogMapper.java
+7
-0
JobTaskRunLogService.java
.../src/main/java/com/byit/service/JobTaskRunLogService.java
+8
-0
NodeDependencyService.java
...src/main/java/com/byit/service/NodeDependencyService.java
+18
-0
NodeService.java
...dmin-core/src/main/java/com/byit/service/NodeService.java
+1
-0
JobTaskRunLogServiceImpl.java
.../java/com/byit/service/impl/JobTaskRunLogServiceImpl.java
+5
-0
NodeDependencyServiceImpl.java
...java/com/byit/service/impl/NodeDependencyServiceImpl.java
+28
-0
JobScheduleHelper.java
...core/src/main/java/com/byit/thread/JobScheduleHelper.java
+41
-13
JobTaskRunLogMapper.xml
...in-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
+10
-1
No files found.
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskRunLogMapper.java
View file @
22c83f51
...
@@ -14,6 +14,13 @@ import java.util.List;
...
@@ -14,6 +14,13 @@ import java.util.List;
@Repository
@Repository
public
interface
JobTaskRunLogMapper
{
public
interface
JobTaskRunLogMapper
{
/**
/**
* 查询没有结束的节点
* @param nodIds
* @return
*/
int
findJobTaskRunLogNotEndNodeByRunCodeCount
(
@Param
(
"nodIds"
)
List
<
Integer
>
nodIds
);
/**
* 查根据flowId和RunId查询一批节点
* 查根据flowId和RunId查询一批节点
* @param flowId
* @param flowId
* @param runId
* @param runId
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskRunLogService.java
View file @
22c83f51
...
@@ -2,6 +2,7 @@ package com.byit.service;
...
@@ -2,6 +2,7 @@ package com.byit.service;
import
com.byit.model.JobTaskRunLog
;
import
com.byit.model.JobTaskRunLog
;
import
com.byit.model.JobTaskRunLogWithBLOBs
;
import
com.byit.model.JobTaskRunLogWithBLOBs
;
import
org.apache.ibatis.annotations.Param
;
import
java.util.List
;
import
java.util.List
;
...
@@ -12,6 +13,13 @@ import java.util.List;
...
@@ -12,6 +13,13 @@ import java.util.List;
* @date: 2019/12/20 19:42
* @date: 2019/12/20 19:42
**/
**/
public
interface
JobTaskRunLogService
{
public
interface
JobTaskRunLogService
{
/**
* 查询没有结束的节点
* @param nodIds
* @return
*/
int
findJobTaskRunLogNotEndNodeByRunCodeCount
(
List
<
Integer
>
nodIds
);
/**
/**
* 查根据flowId和RunId查询一批节点
* 查根据flowId和RunId查询一批节点
* @param flowId
* @param flowId
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/NodeDependencyService.java
0 → 100644
View file @
22c83f51
package
com
.
byit
.
service
;
import
java.util.List
;
/**
* @author huangfu
*/
public
interface
NodeDependencyService
{
/**
* 根据节点id查询本节点依赖的节点
* @param nodeId
* @return
*/
List
<
Integer
>
findDependIdByNodeId
(
Integer
nodeId
);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/NodeService.java
View file @
22c83f51
...
@@ -6,4 +6,5 @@ package com.byit.service;
...
@@ -6,4 +6,5 @@ package com.byit.service;
* @create: 2019-12-24 10:38
* @create: 2019-12-24 10:38
*/
*/
public
interface
NodeService
{
public
interface
NodeService
{
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobTaskRunLogServiceImpl.java
View file @
22c83f51
...
@@ -29,6 +29,11 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
...
@@ -29,6 +29,11 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
}
}
@Override
@Override
public
int
findJobTaskRunLogNotEndNodeByRunCodeCount
(
List
<
Integer
>
nodIds
)
{
return
jobTaskRunLogMapper
.
findJobTaskRunLogNotEndNodeByRunCodeCount
(
nodIds
);
}
@Override
public
List
<
JobTaskRunLogWithBLOBs
>
findJobTaskRunLogWithBLOBsByFlowIdAndRunId
(
Integer
flowId
,
String
runId
)
{
public
List
<
JobTaskRunLogWithBLOBs
>
findJobTaskRunLogWithBLOBsByFlowIdAndRunId
(
Integer
flowId
,
String
runId
)
{
return
jobTaskRunLogMapper
.
findJobTaskRunLogWithBLOBsByFlowIdAndRunId
(
flowId
,
runId
);
return
jobTaskRunLogMapper
.
findJobTaskRunLogWithBLOBsByFlowIdAndRunId
(
flowId
,
runId
);
}
}
...
...
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/NodeDependencyServiceImpl.java
0 → 100644
View file @
22c83f51
package
com
.
byit
.
service
.
impl
;
import
com.byit.mapper.NodeDependencyMapper
;
import
com.byit.service.NodeDependencyService
;
import
lombok.extern.slf4j.Slf4j
;
import
org.springframework.beans.factory.annotation.Autowired
;
import
org.springframework.stereotype.Service
;
import
javax.xml.ws.Action
;
import
java.util.List
;
/**
* @author Administrator
*/
@Service
@Slf4j
public
class
NodeDependencyServiceImpl
implements
NodeDependencyService
{
private
final
NodeDependencyMapper
nodeDependencyMapper
;
public
NodeDependencyServiceImpl
(
NodeDependencyMapper
nodeDependencyMapper
)
{
this
.
nodeDependencyMapper
=
nodeDependencyMapper
;
}
@Override
public
List
<
Integer
>
findDependIdByNodeId
(
Integer
nodeId
)
{
return
nodeDependencyMapper
.
findDependIdByNodeId
(
nodeId
);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JobScheduleHelper.java
View file @
22c83f51
...
@@ -5,8 +5,10 @@ import com.byit.job.WorkRoulette;
...
@@ -5,8 +5,10 @@ import com.byit.job.WorkRoulette;
import
com.byit.model.JobTask
;
import
com.byit.model.JobTask
;
import
com.byit.model.JobTaskRunLogWithBLOBs
;
import
com.byit.model.JobTaskRunLogWithBLOBs
;
import
com.byit.model.JobTaskSchedule
;
import
com.byit.model.JobTaskSchedule
;
import
com.byit.service.JobTaskRunLogService
;
import
com.byit.service.JobTaskScheduleService
;
import
com.byit.service.JobTaskScheduleService
;
import
com.byit.service.JobTaskService
;
import
com.byit.service.JobTaskService
;
import
com.byit.service.NodeDependencyService
;
import
com.byit.service.impl.JobTaskRunLogServiceImpl
;
import
com.byit.service.impl.JobTaskRunLogServiceImpl
;
import
com.byit.task.JavaBeanJobTask
;
import
com.byit.task.JavaBeanJobTask
;
import
com.byit.util.SpringUtil
;
import
com.byit.util.SpringUtil
;
...
@@ -39,6 +41,8 @@ public class JobScheduleHelper{
...
@@ -39,6 +41,8 @@ public class JobScheduleHelper{
private
final
JobTaskService
jobTaskService
;
private
final
JobTaskService
jobTaskService
;
private
final
JobTaskScheduleService
jobTaskScheduleService
;
private
final
JobTaskScheduleService
jobTaskScheduleService
;
private
final
NodeDependencyService
nodeDependencyService
;
private
final
JobTaskRunLogService
jobTaskRunLogService
;
/**
/**
* 读取任务节点的预读
* 读取任务节点的预读
...
@@ -67,15 +71,25 @@ public class JobScheduleHelper{
...
@@ -67,15 +71,25 @@ public class JobScheduleHelper{
private
volatile
boolean
scheduleThreadToStop
=
false
;
private
volatile
boolean
scheduleThreadToStop
=
false
;
@Autowired
@Autowired
public
JobScheduleHelper
(
JobTaskScheduleService
jobTaskScheduleService
,
JobTaskService
jobTaskService
)
{
public
JobScheduleHelper
(
JobTaskScheduleService
jobTaskScheduleService
,
JobTaskService
jobTaskService
,
NodeDependencyService
nodeDependencyService
,
JobTaskRunLogService
jobTaskRunLogService
)
{
this
.
jobTaskScheduleService
=
jobTaskScheduleService
;
this
.
jobTaskScheduleService
=
jobTaskScheduleService
;
this
.
jobTaskService
=
jobTaskService
;
this
.
jobTaskService
=
jobTaskService
;
this
.
nodeDependencyService
=
nodeDependencyService
;
this
.
jobTaskRunLogService
=
jobTaskRunLogService
;
}
}
/**
/**
* 启动两条线程
* 启动两条线程
*/
*/
public
void
start
(){
public
void
start
(){
jobInfoThreadStart
();
scheduleThreadStart
();
}
/**
* 任务节点扫描
*/
public
void
jobInfoThreadStart
(){
jobInfoThread
=
new
Thread
(()->{
jobInfoThread
=
new
Thread
(()->{
try
{
try
{
TimeUnit
.
MILLISECONDS
.
sleep
(
PRE_READ_MS
-
System
.
currentTimeMillis
()%
1000
);
TimeUnit
.
MILLISECONDS
.
sleep
(
PRE_READ_MS
-
System
.
currentTimeMillis
()%
1000
);
...
@@ -117,12 +131,23 @@ public class JobScheduleHelper{
...
@@ -117,12 +131,23 @@ public class JobScheduleHelper{
* 大概思路,根据任务流id,从任务流执行回溯表查询该任务流的所有节点,查看上级节点是否已经执行成功
* 大概思路,根据任务流id,从任务流执行回溯表查询该任务流的所有节点,查看上级节点是否已经执行成功
* //TODO 需要修改 判断父节点是否执行完毕 注意 父节点是一个集合
* //TODO 需要修改 判断父节点是否执行完毕 注意 父节点是一个集合
*/
*/
if
(
true
){
log
.
debug
(
"任务:{}"
,
jobTask
);
//虚节点的状态
JobTaskSchedule
jobTaskSchedule
=
new
JobTaskSchedule
();
if
(
"0"
.
equals
(
jobTask
.
getIsVirtual
())){
BeanUtils
.
copyProperties
(
jobTask
,
jobTaskSchedule
);
//在日志表里面创建一条记录
jobTaskSchedules
.
add
(
jobTaskSchedule
);
//根据 map_flow_id查询当前的版本的工作流 使用祝工作流的runId 保存到执行记录表和任务表
}
else
{
List
<
Integer
>
dependIdByNodeId
=
nodeDependencyService
.
findDependIdByNodeId
(
jobTask
.
getNodeId
());
int
jobTaskRunLogNotEndNodeByRunCodeCount
=
jobTaskRunLogService
.
findJobTaskRunLogNotEndNodeByRunCodeCount
(
dependIdByNodeId
);
if
(
"start"
.
equals
(
jobTask
.
getNodeName
())
||
(
jobTaskRunLogNotEndNodeByRunCodeCount
==
0
)){
log
.
debug
(
"任务:{}"
,
jobTask
);
JobTaskSchedule
jobTaskSchedule
=
new
JobTaskSchedule
();
BeanUtils
.
copyProperties
(
jobTask
,
jobTaskSchedule
);
jobTaskSchedules
.
add
(
jobTaskSchedule
);
}
}
}
});
});
if
(
CollectionUtil
.
isNotEmpty
(
jobTaskSchedules
))
{
if
(
CollectionUtil
.
isNotEmpty
(
jobTaskSchedules
))
{
...
@@ -199,15 +224,17 @@ public class JobScheduleHelper{
...
@@ -199,15 +224,17 @@ public class JobScheduleHelper{
//设置名字
//设置名字
jobInfoThread
.
setName
(
"myth-job,admin JobScheduleHelper#jobInfoThread"
);
jobInfoThread
.
setName
(
"myth-job,admin JobScheduleHelper#jobInfoThread"
);
jobInfoThread
.
start
();
jobInfoThread
.
start
();
}
/**
* 开始操作任务排期表:
* 1. 扫描五秒要执行的数据
* 2.加载时,创建任务日志,传入调度时间
* 3.加载到任务调度轮盘
* 4.删除数据
*/
private
void
scheduleThreadStart
(){
/**
* 开始操作任务排期表:
* 1. 扫描五秒要执行的数据
* 2.加载时,创建任务日志,传入调度时间
* 3.加载到任务调度轮盘
* 4.删除数据
*/
scheduleThread
=
new
Thread
(()
->{
scheduleThread
=
new
Thread
(()
->{
try
{
try
{
TimeUnit
.
MILLISECONDS
.
sleep
(
4000
-
System
.
currentTimeMillis
()%
1000
);
TimeUnit
.
MILLISECONDS
.
sleep
(
4000
-
System
.
currentTimeMillis
()%
1000
);
...
@@ -320,6 +347,7 @@ public class JobScheduleHelper{
...
@@ -320,6 +347,7 @@ public class JobScheduleHelper{
scheduleThread
.
setDaemon
(
true
);
scheduleThread
.
setDaemon
(
true
);
scheduleThread
.
start
();
scheduleThread
.
start
();
}
}
public
void
doStop
(){
public
void
doStop
(){
this
.
jobInfoThreadToStop
=
true
;
this
.
jobInfoThreadToStop
=
true
;
try
{
try
{
...
...
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
View file @
22c83f51
...
@@ -39,6 +39,15 @@
...
@@ -39,6 +39,15 @@
<sql
id=
"Blob_Column_List"
>
<sql
id=
"Blob_Column_List"
>
run_msg, trigger_msg
run_msg, trigger_msg
</sql>
</sql>
<select
id=
"findJobTaskRunLogNotEndNodeByRunCodeCount"
resultType=
"java.lang.Integer"
>
select count(id) from job_task_run_log where node_id in
<foreach
item=
"nodeId"
collection=
"nodIds"
open=
"("
separator=
","
close=
")"
>
#{nodeId}
</foreach>
and (run_code = '0' or run_code = '5')
</select>
<select
id=
"findJobTaskRunLogWithBLOBsByFlowIdAndRunId"
resultMap=
"ResultMapWithBLOBs"
>
<select
id=
"findJobTaskRunLogWithBLOBsByFlowIdAndRunId"
resultMap=
"ResultMapWithBLOBs"
>
select
select
<include
refid=
"Base_Column_List"
/>
<include
refid=
"Base_Column_List"
/>
...
@@ -56,7 +65,7 @@
...
@@ -56,7 +65,7 @@
<select
id=
"findNotEndVirtualNode"
resultMap=
"BaseResultMap"
>
<select
id=
"findNotEndVirtualNode"
resultMap=
"BaseResultMap"
>
select
<include
refid=
"Base_Column_List"
/>
from job_task_run_log
select
<include
refid=
"Base_Column_List"
/>
from job_task_run_log
where is_virtual='0' and
(run_code is null or run_code = '')
where is_virtual='0' and
run_code = '0'
</select>
</select>
<select
id=
"findJobTaskRunLogByLogId"
parameterType=
"java.lang.Integer"
resultMap=
"ResultMapWithBLOBs"
>
<select
id=
"findJobTaskRunLogByLogId"
parameterType=
"java.lang.Integer"
resultMap=
"ResultMapWithBLOBs"
>
...
...
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