Skip to content

B
byit-myth-job
Commit 4307d71e by huangfusuper
Options

调度中心集成rpc框架

parent b280b623
Showing with 594 additions and 172 deletions
byit-myth-admin/pom.xml
......@@ -46,6 +46,11 @@
<version>20.0</version>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-api</artifactId>
</dependency>
</dependencies>
<build>
......
byit-myth-admin/src/main/java/com/byit/AdminApplication.java
package com.byit;
import com.byit.rpc.remoting.provider.annotation.RpcService;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
......@@ -10,6 +11,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
* @date: 2019/11/20 11:44
**/
@SpringBootApplication
@RpcService(http_type = true)
public class AdminApplication {
public static void main(String[] args) {
SpringApplication.run(AdminApplication.class,args);
......
byit-myth-admin/src/main/java/com/byit/config/DemoRegisterConfig.java 0 → 100644
package com.byit.config;
import com.byit.rpc.registry.impl.RegistryServiceRegistry;
import com.byit.rpc.remoting.invoker.impl.RpcSpringInvokerFactory;
import com.byit.rpc.remoting.provider.impl.RpcSpringProviderFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @author liyuan
* create date on 2019-11-14 15:09
* description:
* <p></p>
* modified By
**/
@Configuration
@Slf4j
public class DemoRegisterConfig {
@Value("${myth-rpc.registry.address}")
private String address;
@Value("${myth-rpc.registry.biz}")
private String biz;
@Value("${myth-rpc.registry.env}")
private String env;
@Value("${myth-rpc.registry.port}")
private int port;
@Bean
public RpcSpringInvokerFactory invokerFactory(){
RpcSpringInvokerFactory invokerFactory = new RpcSpringInvokerFactory();
invokerFactory.setServiceRegistryClass(RegistryServiceRegistry.class);
invokerFactory.setServiceRegistryParam(new HashMap<String,String>(){{
put(RegistryServiceRegistry.REGISTRY_ADDRESS,address);
put(RegistryServiceRegistry.BIZ, biz);
put(RegistryServiceRegistry.ENV,env);
}});
log.info(">>>>>>>>>> 例子项目 invoker config 初始化成功");
return invokerFactory;
}
@Bean
public RpcSpringProviderFactory rpcSpringProviderFactory() {
RpcSpringProviderFactory providerFactory = new RpcSpringProviderFactory();
providerFactory.setPort(port);
providerFactory.setServiceRegistryClass(RegistryServiceRegistry.class);
providerFactory.setServiceRegistryParam(new HashMap<String, String>() {{
put(RegistryServiceRegistry.REGISTRY_ADDRESS, address);
put(RegistryServiceRegistry.BIZ, biz);
put(RegistryServiceRegistry.ENV, env);
}});
log.info(">>>>>>>>>>>>>>> byit-demo-service 向注册中心注册config初始化完成");
return providerFactory;
}
}
byit-myth-admin/src/main/java/com/byit/controller/JobController.java
package com.byit.controller;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.JobRunResultDto;
import com.byit.job.dto.PluginBeanJobInfo;
import com.byit.model.JobTask;
import com.byit.model.RunRecording;
import com.byit.rpc.remoting.invoker.annotation.RpcReference;
import com.byit.service.JobTaskService;
import com.byit.service.RunScriptService;
import com.byit.service.TestServiceImpl;
import com.byit.thread.LogCallbackThread;
import com.byit.thread.LogScanHelper;
import com.byit.thread.RunRecordingScanHelper;
......@@ -26,6 +31,9 @@ import java.util.List;
@RequestMapping("job")
public class JobController {
@Autowired
private RunScriptService runScriptService;
private final JobTaskService jobTaskService;
@Autowired
......@@ -55,8 +63,10 @@ public class JobController {
@Autowired
private RunRecordingScanHelper runRecordingScanThread;
@GetMapping("test")
public String test(){
runRecordingScanThread.start();
return "ASDSADSADSADSAD";
public DispatchResponseDto test(){
DispatchResponseDto dispatchResponseDto = runScriptService.runScript(null);
System.out.println("---------------------");
return dispatchResponseDto;
}
}
byit-myth-admin/src/main/java/com/byit/service/TestServiceImpl.java 0 → 100644
package com.byit.service;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.rpc.remoting.invoker.annotation.RpcReference;
import org.springframework.stereotype.Service;
@Service
public class TestServiceImpl {
@RpcReference
private ScriptExecutorService scriptExecutorService;
public DispatchResponseDto test(){
return scriptExecutorService.runPythonScript(null);
}
}
byit-myth-admin/src/main/resources/application-dev.yml 0 → 100644
spring:
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://10.0.10.118:3306/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
username: root
password: 123456
mail:
host: smtp.163.com #我自己的SMTP服务器地址
username: huangfusuper@163.com #登录用户名
password: huangfu0110 #授权密码
default-encoding: UTF-8
protocol: smtp #协议
properties:
from: huangfusuper@163.com #真实邮箱
mybatis:
mapper-locations: /mapper/*.xml
myth-rpc:
registry:
address: http://localhost:8080/myth-register
env: huangfu
biz: byit-myth-job
port: 6665
logging:
path: /data/mythjob
file: myth_log_file
\ No newline at end of file
byit-myth-admin/src/main/resources/application.yml
server:
port: 8998
spring:
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://10.0.10.118:3306/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
username: root
password: 123456
mail:
host: smtp.163.com #我自己的SMTP服务器地址
username: huangfusuper@163.com #登录用户名
password: huangfu0110 #授权密码
default-encoding: UTF-8
protocol: smtp #协议
properties:
from: huangfusuper@163.com #真实邮箱
mybatis:
mapper-locations: /mapper/*.xml
logging:
path: /data/mythjob
file: myth_log_file
profiles:
active: dev
application:
name: myth-job-admin
byit-myth-core/myth-admin-core/pom.xml
......@@ -65,6 +65,16 @@
<version>1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>byit-myth-rpc</artifactId>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-api</artifactId>
</dependency>
</dependencies>
<build>
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/RunScriptService.java 0 → 100644
package com.byit.service;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.ScriptDto;
/**
* @author huanfu
*/
public interface RunScriptService {
/**
* 运行脚本的方法
* @param scriptDto
* @return
*/
DispatchResponseDto runScript(ScriptDto scriptDto);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunScriptServiceImpl.java 0 → 100644
package com.byit.service.impl;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.ScriptDto;
import com.byit.rpc.remoting.invoker.annotation.RpcReference;
import com.byit.service.RunScriptService;
import org.springframework.stereotype.Service;
/**
* @author huangfu
*/
@Service
public class RunScriptServiceImpl implements RunScriptService {
@RpcReference
private ScriptExecutorService scriptExecutorService;
@Override
public DispatchResponseDto runScript(ScriptDto scriptDto) {
return scriptExecutorService.runPythonScript(scriptDto);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java
......@@ -62,11 +62,6 @@ public class JavaBeanJobTask implements TimerTask {
}
private void runJob(JobTaskSchedule mythJobTaskSchedule) throws InterruptedException{
if(!"start".equals(mythJobTaskSchedule.getNodeName())){
log.error("{}----------开始执行调度了吗？我要开始睡觉了-------------------",mythJobTaskSchedule.getNodeName());
Thread.sleep(100000);
}
//根据负责均衡方案获取对应IP
RpcLoadBalance rpcInvokerRouter = LoadBalance.match(mythJobTaskSchedule.getRoutingStrategy( ), LoadBalance.ROUND).rpcInvokerRouter;
try{
......@@ -75,7 +70,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:8080/job/callbackRes");
adminSenPluginDto.setCallbackUrl("http://127.0.0.1:8998/job/callbackRes");
adminSenPluginDto.setJobHandelName(jobHandelName);
adminSenPluginDto.setJobParam(param);
adminSenPluginDto.setRunId(runId);
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java 0 → 100644
package com.byit.task;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.enums.NodePropertyEnum;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.ScriptDto;
import com.byit.job.enums.JobResultEnum;
import com.byit.model.JobTaskSchedule;
import com.byit.rpc.remoting.invoker.annotation.RpcReference;
import com.byit.service.RunScriptService;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Controller;
/**
* @author Administrator
*/
@Slf4j
public class ScriptExecutorJobTask implements TimerTask {
private JobTaskSchedule mythJobTaskSchedule;
public JobTaskSchedule getMythJobTaskSchedule() {
return mythJobTaskSchedule;
}
public void setMythJobTaskSchedule(JobTaskSchedule mythJobTaskSchedule) {
this.mythJobTaskSchedule = mythJobTaskSchedule;
}
@Override
public void run(Timeout timeout) throws Exception {
//获取任务级别 1最低 2最高
String priority = mythJobTaskSchedule.getPriority();
if(NodePropertyEnum.ADVANCED_NODE.getCode().equals(priority)){
log.debug("--------检测到高级节点--------");
MythJobAutoConfigure.ADVANCED_JOB_THREAD_POOL.execute(()->{
runJob(mythJobTaskSchedule);
});
}else{
MythJobAutoConfigure.LOW_LEVEL_JOB_THREAD_POOL.execute(()->{
log.debug("--------检测到低级节点--------");
runJob(mythJobTaskSchedule);
});
}
}
private void runJob(JobTaskSchedule mythJobTaskSchedule) {
RunScriptService runScriptService = SpringUtil.getBean(RunScriptService.class);
ScriptDto scriptDto = new ScriptDto();
scriptDto.setLogId(mythJobTaskSchedule.getLogId());
scriptDto.setCommand(mythJobTaskSchedule.getRunCommand());
scriptDto.setParam(mythJobTaskSchedule.getRunParam());
scriptDto.setRunId(mythJobTaskSchedule.getRunId());
scriptDto.setRemotePath(mythJobTaskSchedule.getScriptUrls());
DispatchResponseDto dispatchResponseDto = runScriptService.runScript(scriptDto);
System.out.println(dispatchResponseDto);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JobScheduleHelper.java
......@@ -10,6 +10,7 @@ import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.mapservice.RunRecordingAndJobTaskService;
import com.byit.service.mapservice.TaskAndLogServer;
import com.byit.task.JavaBeanJobTask;
import com.byit.task.ScriptExecutorJobTask;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
......@@ -326,16 +327,21 @@ public class JobScheduleHelper{
jobTaskSchedules.forEach(mythJobTaskSchedule ->{
//如果是重跑就有logId
Integer logId = mythJobTaskSchedule.getLogId();
if ("JAVA".equals(mythJobTaskSchedule.getJobType())) {
if(logId == null){
logId = saveLog(mythJobTaskSchedule);
}
mythJobTaskSchedule.setLogId(logId);
Long triggerTime = mythJobTaskSchedule.getTriggerTime();
if ("JAVA".equals(mythJobTaskSchedule.getJobType())) {
//构建调度执行器
JavaBeanJobTask javaBeanJobTask = new JavaBeanJobTask(mythJobTaskSchedule);
jobTaskScheduleService.delete(mythJobTaskSchedule.getId());
WorkRoulette.addJob(javaBeanJobTask,mythJobTaskSchedule.getTriggerTime());
WorkRoulette.addJob(javaBeanJobTask,triggerTime);
}else if("SCRIPT".equals(mythJobTaskSchedule.getJobType())){
ScriptExecutorJobTask scriptExecutorJobTask = new ScriptExecutorJobTask();
scriptExecutorJobTask.setMythJobTaskSchedule(mythJobTaskSchedule);
WorkRoulette.addJob(scriptExecutorJobTask,triggerTime);
}
jobTaskScheduleService.delete(mythJobTaskSchedule.getId());
});
}else{
preReadSuc = false;
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/DispatchResponseDto.java
package com.byit.job.dto;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.*;
import java.io.Serializable;
/**
* @program: byit-myth-job->DispatchResponseDto
......@@ -15,7 +14,8 @@ import lombok.NoArgsConstructor;
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class DispatchResponseDto {
@ToString
public class DispatchResponseDto implements Serializable {
private String code;
private String msg;
private String content;
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/ScriptDto.java 0 → 100644
package com.byit.job.dto;
import lombok.Data;
import java.io.Serializable;
/**
* 脚本传输的dto
* @author huangfu
*/
@Data
public class ScriptDto implements Serializable {
/**
* 日志的id
*/
private Integer logId;
/**
* 运行标识
*/
private String runId;
/**
* 运行所需参数
*/
private String param;
/**
* 执行命令
*/
private String command;
/**
* 脚本路径
*/
private String remotePath;
}
byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/Test.java deleted 100644 → 0
package com.byit.executor;
import com.byit.executor.jobExecutor.process.MythJobProcess;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.util.*;
/**
* @description:
* @author: gml
* @create: 2020-01-15 11:10
*/
@Slf4j
public class Test {
public static void main(String[] args) throws IOException, InterruptedException {
List<MythJobProcess> list = new ArrayList<>();
Thread thread = new Thread(() -> {
List<String> cmdList = new ArrayList<>();
cmdList = Arrays.asList("python D:\\workspace\\pycharmWorkSpace\\test\\com.test\\Test.py".split(" "));
Map<String, String> env = new HashMap<>();
env.put("python", "C:\\Program Files\\Python38");
MythJobProcess mythJobProcess = new MythJobProcess(cmdList, env, "D:\\workspace\\pycharmWorkSpace\\test\\com.test");
list.add(mythJobProcess);
try {
String log = mythJobProcess.run();
System.out.println("运行日志" + log);
} catch (IOException e) {
e.printStackTrace();
}
});
thread.start();
Thread.sleep(30000);
list.forEach(process -> {
process.hardKill();
});
}
}
byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/conf/FileSystemContext.java
......@@ -32,11 +32,6 @@ public class FileSystemContext {
fileSystem.fileRemove(remPath);
}
public void fileAppend(String filePath, byte[] fileBuffer) throws IOException, MyException {
fileSystem.fileAppend(filePath,fileBuffer);
}
public Map<String, String> getFileMate(String filePath) throws IOException, MyException {
return fileSystem.getFileMate(filePath);
}
......
byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/filesystem/FastDfsFileSystem.java
......@@ -64,15 +64,6 @@ public class FastDfsFileSystem implements FileSystem {
}
@Override
public void fileAppend(String filePath, byte[] fileBuffer) throws IOException, MyException {
TrackerClient trackerClient = new TrackerClient();
TrackerServer trackerServer = trackerClient.getConnection();
StorageClient1 storageClient1 = new StorageClient1(trackerServer, null);
storageClient1.append_file1(filePath,fileBuffer);
trackerServer.close();
}
@Override
public Map<String, String> getFileMate(String filePath) throws IOException, MyException {
TrackerClient trackerClient = new TrackerClient();
TrackerServer trackerServer = trackerClient.getConnection();
......
byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/filesystem/FileSystem.java
......@@ -12,37 +12,39 @@ import java.util.Map;
public interface FileSystem {
/**
* 文件上传接口
* @param fileBuffer 文件的字节流
* @param fileExtName 文件的扩展名称
* @param mateDaTA 文件的元信息
* @return
* @param fileBuffer 文件数组
* @param fileExtName 文件后缀
* @param mateDaTA 文件源信息
* @return 上传路径
* @throws IOException
* @throws MyException
*/
String uploadFile(byte[] fileBuffer, String fileExtName, Map<String,String> mateDaTA) throws IOException, MyException;
/**
* 文件下载操作
* @param remPath 远程文件地址
* @return 文件流
* 文件下载接口
* @param remPath 远程地址
* @return 文件字节数组
* @throws IOException
* @throws MyException
*/
byte[] downloaderFile(String remPath) throws IOException, MyException;
/**
* 文件删除
* @param remPath 远程地址
* @throws IOException
* @throws MyException
*/
void fileRemove(String remPath) throws IOException, MyException;
/**
* 数据追加
* @param filePath
* @param fileBuffer
*/
void fileAppend(String filePath,byte[] fileBuffer) throws IOException, MyException;
/**
* 查询文件原信息
* @param filePath
* @return
* @throws IOException
* @throws MyException
*/
Map<String,String> getFileMate(String filePath) throws IOException, MyException;
......
byit-myth-core/myth-executor-core/src/test/java/com/test/FastDFSTest.java
......@@ -4,6 +4,9 @@ import com.byit.executor.conf.FileSystemContext;
import com.byit.executor.filesystem.FastDfsFileSystem;
import org.apache.commons.io.FileUtils;
import org.csource.common.MyException;
import org.csource.fastdfs.StorageClient1;
import org.csource.fastdfs.TrackerClient;
import org.csource.fastdfs.TrackerServer;
import org.junit.Before;
import org.junit.Test;
......@@ -29,7 +32,7 @@ public class FastDFSTest {
@Test
public void downLoad() throws IOException, MyException {
String filePath = "ddmp/M00/00/00/CgB4Al5GY5GAMGVaAAAHa9wlKKc755.log";
String filePath = "ddmp/M00/00/00/CgB4Al5J-4aAQ8v6AAAHa95jlv4140.log";
Map<String, String> fileMate = fileSystemContext.getFileMate(filePath);
String filename = fileMate.get("filename");
byte[] bytes = fileSystemContext.downloaderFile(filePath);
......@@ -41,9 +44,10 @@ public class FastDFSTest {
}
@Test
public void appendTest() throws IOException, MyException {
String filePath = "ddmp/M00/00/00/CgB4Al5GY5GAMGVaAAAHa9wlKKc755.log";
byte[] bytes = FileUtils.readFileToByteArray(new File("E:\\image/20180101.log"));
fileSystemContext.fileAppend(filePath,bytes);
public void removeTest() throws IOException, MyException {
fileSystemContext.fileRemove("ddmp/M00/00/00/CgB4Al5J-4aAQ8v6AAAHa95jlv4140.log");
}
}
byit-myth-executor/myth-executor-api/src/main/java/com/byit/executor/api/ScriptExecutorService.java 0 → 100644
package com.byit.executor.api;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.ScriptDto;
import com.byit.job.enums.JobResultEnum;
/**
* Python等脚本的执行器服务类
* @author huangfu
*/
public interface ScriptExecutorService {
/**
* Python等脚本的执行
* @param scriptDto 参数
* @return 调用结果
*/
DispatchResponseDto runPythonScript(ScriptDto scriptDto);
}
byit-myth-executor/myth-exector-plugin/src/main/java/com/byit/utils/JobUtils.java → byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java
......@@ -40,6 +40,8 @@ public class JobUtils {
* 暂停工作流
*/
private static final String REQUEST_FLOW_STOP = "/api/flow/stopSchedule";
private static final String SERVER_PORT = "8998";
/**
* 当前项目运行环境 jar file
*/
......@@ -80,7 +82,7 @@ public class JobUtils {
log.info("---------------开始发布工作流,flowName:{}---------------------", pluginPackage.getFlow().getName());
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + "8080" + REQUEST_FLOW_PUBLISH;
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_PUBLISH;
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl, "param=" + JSON.toJSONString(pluginPackage, WriteClassName));
//String addRequestResult = HttpUtil.post(requestUrl, JSON.toJSONString(pluginPackage, WriteClassName))
......@@ -96,7 +98,7 @@ public class JobUtils {
*/
public static String stopFlow(String flowName, String workspaceName){
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + "8080" + REQUEST_FLOW_STOP;
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_FLOW_STOP;
Map<String,String> map = new HashMap<>(5);
map.put("flowName",flowName);
map.put("workspaceName",workspaceName);
......@@ -116,7 +118,7 @@ public class JobUtils {
log.info("---------------开始创建工作空间,workspaceName:{}---------------------", workspaceName);
//请求的路径
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + "8080" + REQUEST_WORKSPACE_ADD;
String requestUrl = REQUEST_PREFIX + "127.0.0.1" + ":" + SERVER_PORT + REQUEST_WORKSPACE_ADD;
//发送请求 添加任务
String addRequestResult = HttpUtil.post(requestUrl, "workspaceName="+ workspaceName);
log.info("--------------------添加任务完成，添加结果为：{}------------------------",addRequestResult);
......
byit-myth-executor/myth-executor-server/pom.xml
......@@ -17,25 +17,16 @@
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-exector-plugin</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>byit-myth-rpc</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
......
byit-myth-executor/myth-executor-server/src/main/java/com/byit/ExecutorApplication.java → byit-myth-executor/myth-executor-server/src/main/java/com/byit/ExecutorServiceApplication.java
......@@ -10,8 +10,8 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
* @date: 2019/11/14 16:03
**/
@SpringBootApplication
public class ExecutorApplication {
public class ExecutorServiceApplication {
public static void main(String[] args) {
SpringApplication.run(ExecutorApplication.class,args);
SpringApplication.run(ExecutorServiceApplication.class,args);
}
}
byit-myth-executor/myth-executor-server/src/main/java/com/byit/config/DemoRegisterConfig.java 0 → 100644
package com.byit.config;
import com.byit.rpc.registry.impl.RegistryServiceRegistry;
import com.byit.rpc.remoting.provider.impl.RpcSpringProviderFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
/**
* @author liyuan
* create date on 2019-11-14 15:09
* description:
* <p></p>
* modified By
**/
@Configuration
@Slf4j
public class DemoRegisterConfig {
@Value("${myth-rpc.remoting.port}")
private int port;
@Value("${myth-rpc.registry.address}")
private String address;
@Value("${myth-rpc.registry.env}")
private String env;
@Value("${myth-rpc.registry.biz}")
private String biz;
@Bean
public RpcSpringProviderFactory rpcSpringProviderFactory() {
RpcSpringProviderFactory providerFactory = new RpcSpringProviderFactory();
providerFactory.setPort(port);
providerFactory.setServiceRegistryClass(RegistryServiceRegistry.class);
providerFactory.setServiceRegistryParam(new HashMap<String, String>() {{
put(RegistryServiceRegistry.REGISTRY_ADDRESS, address);
put(RegistryServiceRegistry.BIZ, biz);
put(RegistryServiceRegistry.ENV, env);
}});
log.info(">>>>>>>>>>>>>>> byit-demo-service 向注册中心注册config初始化完成");
return providerFactory;
}
}
byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java 0 → 100644
package com.byit.service;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.ScriptDto;
import com.byit.job.enums.JobResultEnum;
import com.byit.rpc.remoting.provider.annotation.RpcService;
import org.springframework.stereotype.Service;
/**
* 脚本执行的实现
* @author huangfu
*/
@Service
@RpcService
public class ScriptExecutorServiceImpl implements ScriptExecutorService {
@Override
public DispatchResponseDto runPythonScript(ScriptDto scriptDto) {
System.out.println("--------------3脚本调用开始-----------");
System.out.println("--------------3脚本调用结束-----------");
DispatchResponseDto dispatchResponseDto = DispatchResponseDto.builder()
.code(JobResultEnum.DISPATCH_SUCCESS.getCode())
.msg(JobResultEnum.DISPATCH_SUCCESS.getMsg())
.build();
return dispatchResponseDto;
}
}
byit-myth-executor/myth-executor-server/src/main/resources/application-dev.properties 0 → 100644
# myth-rpc
myth-rpc.remoting.port=7776
myth-rpc.registry.address=http://localhost:8080/myth-register
myth-rpc.registry.env=huangfu
myth-rpc.registry.biz=byit-myth-job
logging.config=classpath:logback.xml
\ No newline at end of file
byit-myth-executor/myth-executor-server/src/main/resources/application-dev.yml deleted 100644 → 0
# myth-rpc
myth-rpc:
remoting:
port: 7001
registry:
address: http://localhost:8080/myth-register
env: dev
biz: ${spring.application.name}
\ No newline at end of file
byit-myth-executor/myth-executor-server/src/main/resources/application.properties 0 → 100644
server.port=7777
spring.application.name=myth-executor-server
spring.profiles.active=dev
byit-myth-executor/myth-executor-server/src/main/resources/application.yml deleted 100644 → 0
server:
port: 7000
spring:
profiles:
active: dev
application:
name: myth-job-server
logging:
config: classpath:logback.xml
\ No newline at end of file
byit-myth-executor/myth-executor-server/src/main/resources/logback.xml
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<!-- 引用现有配置 -->
<include resource="org/springframework/boot/logging/logback/defaults.xml" />
<!-- 控制台 -->
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<configuration debug="false" scan="true" scanPeriod="1 seconds">
<contextName>logback</contextName>
<property name="log.path" value="/Users/liyuan/codes/open/rpc/data/applogs/myth-registry/myth-registry-admin.log"/>
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>${CONSOLE_LOG_PATTERN}</pattern>
<pattern>%d{HH:mm:ss.SSS} %contextName [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<root level="INFO">
<appender-ref ref="CONSOLE" />
<appender name="file" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}</file>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<fileNamePattern>${log.path}.%d{yyyy-MM-dd}.zip</fileNamePattern>
</rollingPolicy>
<encoder>
<pattern>%date %level [%thread] %logger{36} [%file : %line] %msg%n
</pattern>
</encoder>
</appender>
<root level="info">
<appender-ref ref="console"/>
<appender-ref ref="file"/>
</root>
</configuration>
\ No newline at end of file
byit-myth-executor/pom.xml
......@@ -14,7 +14,7 @@
<modules>
<module>myth-executor-api</module>
<module>myth-executor-server</module>
<module>myth-exector-plugin</module>
<module>myth-executor-plugin</module>
</modules>
......
byit-myth-register/myth-register-client/src/main/java/com/byit/registry/client/RegistryBaseClient.java
......@@ -54,7 +54,7 @@ public class RegistryBaseClient {
/**
* registry
*
* 注册表 调用注册的接口完成调用 {@link com.byit.registry.admin.controller.ApiController#registry(java.lang.String)}
* @param registryDataList
* @return
*/
......
byit-myth-register/myth-register-server/src/main/java/com/byit/registry/admin/controller/ApiController.java
......@@ -194,6 +194,7 @@ public class ApiController {
// parse data
RegistryParamVO registryParamVO = null;
try {
//注册数据就是从这里加载的
registryParamVO = JacksonUtil.readValue(data, RegistryParamVO.class);
} catch (Exception e) {
System.out.println(e);
......
byit-myth-register/myth-register-server/src/main/java/com/byit/registry/admin/core/util/PropUtil.java
......@@ -16,7 +16,7 @@ public class PropUtil {
/**
* load prop
*
* 以“ file：”开头的磁盘路径，其他类路径
* @param propertyFileName disk path when start with "file:", other classpath
* @return
*/
......
byit-myth-register/myth-register-server/src/main/resources/application.properties
......@@ -18,9 +18,9 @@ spring.freemarker.settings.number_format=0.##########
mybatis.mapper-locations=classpath:/mybatis-mapper/*Mapper.xml
### myth-registry, datasource
spring.datasource.url=jdbc:mysql://10.0.120.30:3307/myth-registry?Unicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false
spring.datasource.url=jdbc:mysql://10.0.10.118:3306/myth-registry?Unicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false
spring.datasource.username=root
spring.datasource.password=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
spring.datasource.type=org.apache.tomcat.jdbc.pool.DataSource
......
demo-client/byit-demo-client/pom.xml
......@@ -48,6 +48,10 @@
<groupId>org.apache.commons</groupId>
<artifactId>commons-exec</artifactId>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-api</artifactId>
</dependency>
</dependencies>
<build>
......
demo-client/byit-demo-client/src/main/java/com/byit/demo/client/controller/IndexController.java
......@@ -2,10 +2,16 @@ package com.byit.demo.client.controller;
import com.byit.demo.client.api.DemoService;
import com.byit.demo.client.api.dto.UserDTO;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.ScriptDto;
import com.byit.job.enums.JobResultEnum;
import com.byit.rpc.remoting.invoker.annotation.RpcReference;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
/**
* @author liyuan
......@@ -14,15 +20,24 @@ import org.springframework.web.bind.annotation.ResponseBody;
* <p></p>
* modified By
**/
@Controller
@RestController
public class IndexController {
@RpcReference
private DemoService demoService;
@RpcReference
private ScriptExecutorService scriptExecutorService;
@RequestMapping("/demo")
@ResponseBody
public UserDTO say(String name){
return demoService.hello(name);
}
@RequestMapping("/demo1")
@ResponseBody
public DispatchResponseDto say1(String name){
DispatchResponseDto dispatchResponseDto = scriptExecutorService.runPythonScript(null);
return dispatchResponseDto;
}
}
demo-client/byit-demo-client/src/main/java/com/byit/job/ScriptAddFlow.java 0 → 100644
package com.byit.job;
import com.byit.job.dto.plugin.*;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.utils.JobUtils;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
public class ScriptAddFlow {
public static void main(String[] args) {
PluginPackage pluginPackage = new PluginPackage();
pluginPackage.setWorkspaceName("test");
pluginPackage.setFlow(createFlow());
JobUtils.publish(pluginPackage);
}
public static PluginFlow createFlow(){
PluginFlow pluginFlow = new PluginFlow();
PluginFlowConfig build = PluginFlowConfig.builder().alarmEmail("huangfukexing@byitgroup.com")
.alarmlAction("1")
.execType("1")
.flowCron("0 0/1 * * * ? *")
.flowTimeout(TimeUnit.MINUTES.toMillis(30))
.priority("2")
.repeatCount(1)
.scheduleFollow("1")
.build();
pluginFlow.setName("脚本");
pluginFlow.setDesc("脚本");
pluginFlow.setConfig(build);
pluginFlow.setPrincipal("皇甫科星");
pluginFlow.setRePublish(false);
pluginFlow.setAuthor("huangfusuper");
pluginFlow.setNodeList(createNodes());
return pluginFlow;
}
/**
* 创建节点
* @return
*/
public static List<PluginBaseNode> createNodes(){
PluginNode pluginNode1 = new PluginNode();
PluginNodeConfig pluginNodeConfig1 = new PluginNodeConfig();
pluginNode1.setName("start");
pluginNode1.setDesc("我是开始节点，打死你");
pluginNode1.setType("node");
pluginNode1.setAuthor("郭郭");
pluginNode1.setJobType("JAVA");
pluginNode1.setHandlerName("start");
pluginNodeConfig1.setFailedRetryCount(2);
pluginNodeConfig1.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig1.setNodeCron("0 0/7 * * * ? *");
pluginNodeConfig1.setNodeTimeout(-1L);
pluginNodeConfig1.setPluginUrls("http://127.0.0.1:8888");
pluginNodeConfig1.setPriority("2");
pluginNodeConfig1.setRoutingStrategy(LoadBalance.ROUND.name());
pluginNode1.setConfig(pluginNodeConfig1);
PluginNode pluginNode2 = new PluginNode();
PluginNodeConfig pluginNodeConfig2 = new PluginNodeConfig();
pluginNode2.setName("中间节点1");
pluginNode2.setDesc("中间节点1");
pluginNode2.setType("node");
pluginNode2.setAuthor("皇甫");
pluginNode2.setJobType("SCRIPT");
pluginNode2.setHandlerName("addJob");
pluginNode2.setRunParam("add1");
pluginNodeConfig2.setFailedRetryCount(2);
pluginNodeConfig2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig2.setNodeCron("0 0/7 * * * ? *");
pluginNodeConfig2.setNodeTimeout(-1L);
pluginNodeConfig2.setPluginUrls("http://127.0.0.1:8888");
pluginNodeConfig2.setPriority("1");
pluginNodeConfig2.setRoutingStrategy(LoadBalance.ROUND.name());
pluginNode2.setConfig(pluginNodeConfig2);
pluginNode2.setDependNodeNameList(Collections.singletonList("start"));
PluginNode pluginNode5 = new PluginNode();
PluginNodeConfig pluginNodeConfig5 = new PluginNodeConfig();
pluginNode5.setName("end");
pluginNode5.setDesc("结束节点");
pluginNode5.setType("node");
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 * * * ? *");
pluginNodeConfig5.setNodeTimeout(-1L);
pluginNodeConfig5.setPluginUrls("http://127.0.0.1:8888");
pluginNodeConfig5.setPriority("2");
pluginNodeConfig5.setRoutingStrategy(LoadBalance.ROUND.name());
pluginNode5.setConfig(pluginNodeConfig5);
pluginNode5.setDependNodeNameList(Collections.singletonList("中间节点1"));
return Arrays.asList(pluginNode5, pluginNode2, pluginNode1);
}
}
demo-client/byit-demo-client/src/main/java/com/byit/job/TestAddFlow.java
......@@ -59,7 +59,7 @@ public class TestAddFlow {
pluginNodeConfig1.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig1.setNodeCron("0 0/7 * * * ? *");
pluginNodeConfig1.setNodeTimeout(-1L);
pluginNodeConfig1.setPluginUrls("http://127.0.0.1:8888");
pluginNodeConfig1.setPluginUrls("http://127.0.0.1:6666");
pluginNodeConfig1.setPriority("2");
pluginNodeConfig1.setRoutingStrategy(LoadBalance.ROUND.name());
pluginNode1.setConfig(pluginNodeConfig1);
......@@ -77,7 +77,7 @@ public class TestAddFlow {
pluginNodeConfig2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig2.setNodeCron("0 0/7 * * * ? *");
pluginNodeConfig2.setNodeTimeout(-1L);
pluginNodeConfig2.setPluginUrls("http://127.0.0.1:8888");
pluginNodeConfig2.setPluginUrls("http://127.0.0.1:6666");
pluginNodeConfig2.setPriority("1");
pluginNodeConfig2.setRoutingStrategy(LoadBalance.ROUND.name());
pluginNode2.setConfig(pluginNodeConfig2);
......@@ -97,7 +97,7 @@ public class TestAddFlow {
p2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
p2.setNodeCron("0 0/7 * * * ? *");
p2.setNodeTimeout(-1L);
p2.setPluginUrls("http://127.0.0.1:8888");
p2.setPluginUrls("http://127.0.0.1:6666");
p2.setPriority("1");
p2.setRoutingStrategy(LoadBalance.ROUND.name());
binglie.setConfig(p2);
......@@ -118,7 +118,7 @@ public class TestAddFlow {
pluginNodeConfig3.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig3.setNodeCron("0 0/7 * * * ? *");
pluginNodeConfig3.setNodeTimeout(-1L);
pluginNodeConfig3.setPluginUrls("http://127.0.0.1:8888");
pluginNodeConfig3.setPluginUrls("http://127.0.0.1:6666");
pluginNodeConfig3.setPriority("2");
pluginNodeConfig3.setRoutingStrategy(LoadBalance.ROUND.name());
pluginNode3.setConfig(pluginNodeConfig3);
......@@ -139,7 +139,7 @@ public class TestAddFlow {
pluginNodeConfig4.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig4.setNodeCron("0 0/7 * * * ? *");
pluginNodeConfig4.setNodeTimeout(-1L);
pluginNodeConfig4.setPluginUrls("http://127.0.0.1:8888");
pluginNodeConfig4.setPluginUrls("http://127.0.0.1:6666");
pluginNodeConfig4.setPriority("1");
pluginNodeConfig4.setRoutingStrategy(LoadBalance.ROUND.name());
pluginNode4.setConfig(pluginNodeConfig4);
......@@ -158,7 +158,7 @@ public class TestAddFlow {
pluginNodeConfig5.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig5.setNodeCron("0 0/7 * * * ? *");
pluginNodeConfig5.setNodeTimeout(-1L);
pluginNodeConfig5.setPluginUrls("http://127.0.0.1:8888");
pluginNodeConfig5.setPluginUrls("http://127.0.0.1:6666");
pluginNodeConfig5.setPriority("2");
pluginNodeConfig5.setRoutingStrategy(LoadBalance.ROUND.name());
pluginNode5.setConfig(pluginNodeConfig5);
......
demo-client/byit-demo-client/src/main/resources/application-dev.properties
myth-rpc.registry.address=http://localhost:8080/myth-register
myth-rpc.registry.env=liyuan
myth-rpc.registry.env=huangfu
myth-rpc.registry.biz=byit-myth-job
myth-rpc.registry.port=9996
logging.config=classpath:logback.xml
\ No newline at end of file
demo-client/byit-demo-service/src/main/resources/application-dev.properties
# myth-rpc
myth-rpc.remoting.port=9998
myth-rpc.registry.address=http://localhost:8080/myth-register
myth-rpc.registry.env=liyuan
myth-rpc.registry.env=huangfu
myth-rpc.registry.biz=byit-myth-job
logging.config=classpath:logback.xml
\ No newline at end of file
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