Netfiex Conductor安装入门指南以及切换为mysql数据源

1. 通过git下载源码到本地 git@github.com:Netflix/conductor.git。
2. 由于Conductor依赖于gradle编译，我们可以通过两种方式安装gradle.
   

   • 通过命令gradlew server自动下载，有点慢不建议。
   • 下载安装包，解压并配置环境变量到\bin目录问题。

3. 如果你有数据源为mysql的需求可以按如下进行更改。

    public interface Configuration { String DB_PROPERTY_NAME = "db"; //修改数据源类型为mysql String DB_DEFAULT_VALUE = "mysql"; } public interface MySQLConfiguration extends Configuration { String JDBC_URL_PROPERTY_NAME = "jdbc.url"; //这里修改数据源地址 String JDBC_URL_DEFAULT_VALUE = "jdbc:mysql://localhost:3306/conductor?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false"; //这里修改用户名 String JDBC_USER_NAME_PROPERTY_NAME = "jdbc.username"; String JDBC_USER_NAME_DEFAULT_VALUE = "root"; //这里修改密码 String JDBC_PASSWORD_PROPERTY_NAME = "jdbc.password"; String JDBC_PASSWORD_DEFAULT_VALUE = ""; String FLYWAY_ENABLED_PROPERTY_NAME = "flyway.enabled"; boolean FLYWAY_ENABLED_DEFAULT_VALUE = true; String FLYWAY_TABLE_PROPERTY_NAME = "flyway.table"; Optional 
       
         FLYWAY_TABLE_DEFAULT_VALUE = Optional.empty(); // The defaults are currently in line with the HikariConfig defaults, which are unfortunately private. //这里修改max,min两个连接数 String CONNECTION_POOL_MAX_SIZE_PROPERTY_NAME = "conductor.mysql.connection.pool.size.max"; int CONNECTION_POOL_MAX_SIZE_DEFAULT_VALUE = 10; String CONNECTION_POOL_MINIMUM_IDLE_PROPERTY_NAME = "conductor.mysql.connection.pool.idle.min"; int CONNECTION_POOL_MINIMUM_IDLE_DEFAULT_VALUE = 3; 
       

   }

4. 修改数据源文件完成,cd 进入conductor下载的源码目录，执行gradlew build -x test跳过测试构建项目，默认生产的包会生成在（/conductor/server/build/libs）。
5. 不带参数启动项目命令：java -jar conductor-server-2.6.0-SNAPSHOT.jar。
6. 访问localhost:8080查看页面。
   [外链图片转存失败(img-BcPPMhZV-61)(https://i.imgur.com/lsJpK2k.png)]
   
   
7. 生成Ui页面，ui依赖于nodejs,以及gulp的插件。
8. 下载安装nodejs安装包，要求高于8以上版本。
9. 全局安装gulp 命令：$ npm install –global gulp
10. cd 进入项目conductor/ui 命令：$ npm install –save-dev gulp生成项目依赖的gulp。
11. 执行命令：gulp watch等待构建成功，然后访问localhost:3000页面查看Conductor的管理界面。
    [外链图片转存失败(img-GzHJTrpf-62)(https://i.imgur.com/3Zeglyt.png)]
    
    
12. 修改数据源为mysql

开始编写第一个task，定义好两个task。

[ { "name": "mytask1", "retryCount": 3, "timeoutSeconds": 1200, "inputKeys": [ "ai1", "ai2" ], "outputKeys": [ "ao1", "ao2" ], "timeoutPolicy": "TIME_OUT_WF", "retryLogic": "FIXED", "retryDelaySeconds": 600, "responseTimeoutSeconds": 3600 }, { "name": "mytask2", "retryCount": 3, "timeoutSeconds": 1200, "inputKeys": [ "bi1", "bi2" ], "outputKeys": [ "bo1", "bo2" ], "timeoutPolicy": "TIME_OUT_WF", "retryLogic": "FIXED", "retryDelaySeconds": 600, "responseTimeoutSeconds": 3600 } ] 

[ { "name": "myworkflow1", "description": "my workflow for test", "version": 1, "tasks": [ { "name": "mytask1", "taskReferenceName": "node1", "type": "SIMPLE", "inputParameters": { "ai1": "${workflow.input.wi1}", "ai2": "${workflow.input.wi2}" } }, { "name": "mytask2", "taskReferenceName": "node2", "type": "SIMPLE", "inputParameters": { "bi1": "${node1.output.ao1}", "bi2": "${node1.output.ao2}" } } ], "outputParameters": { "ao1": "${node1.output.ao1}", "ao2": "${node1.output.ao2}", "bo1": "${node2.output.bo1}", "bo2": "${node2.output.bo2}" }, "schemaVersion": 2 } ] 

[外链图片转存失败(img-wOELVVDd-63)(https://i.imgur.com/jjSPrp0.png)]

编写worker代码来真正执行任务

package com.zd.demo; import com.netflix.conductor.client.http.TaskClient; import com.netflix.conductor.client.task.WorkflowTaskCoordinator; import com.netflix.conductor.client.worker.Worker; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class DemoApplication { public static void main(String[] args) { TaskClient taskClient = new TaskClient(); taskClient.setRootURI("http://localhost:8080/api/"); //Point this to the server API int threadCount = 2; //number of threads used to execute workers. To avoid starvation, should be same or more than number of workers Worker worker1 = new SampleWorker("mytask1"); Worker worker2 = new SampleWorker2("mytask2"); //Create WorkflowTaskCoordinator WorkflowTaskCoordinator.Builder builder = new WorkflowTaskCoordinator.Builder(); WorkflowTaskCoordinator coordinator = builder.withWorkers(worker1, worker2).withThreadCount(threadCount).withTaskClient(taskClient).build(); coordinator.init(); } class SampleWorker implements Worker { private String taskDefName; public SampleWorker(String taskDefName) { this.taskDefName = taskDefName; } @Override public String getTaskDefName() { return taskDefName; } @Override public TaskResult execute(Task task) { System.out.printf("Executing %s\n", taskDefName); System.out.println("ai1:" + task.getInputData().get("ai1")); System.out.println("ai2:" + task.getInputData().get("ai2")); TaskResult result = new TaskResult(task); result.setStatus(TaskResult.Status.COMPLETED); //Register the output of the task result.getOutputData().put("ao1", String.valueOf(task.getInputData().get("ai1")) + " from ai1"); result.getOutputData().put("ao2", String.valueOf(task.getInputData().get("ai2")) + " from ai2"); return result; } class SampleWorker2 implements Worker { private String taskDefName; public SampleWorker2(String taskDefName) { this.taskDefName = taskDefName; } @Override public String getTaskDefName() { return taskDefName; } @Override public TaskResult execute(Task task) { System.out.printf("Executing %s\n", taskDefName); System.out.println("bi1:" + task.getInputData().get("bi1")); System.out.println("bi2:" + task.getInputData().get("bi2")); TaskResult result = new TaskResult(task); result.setStatus(TaskResult.Status.COMPLETED); //Register the output of the task result.getOutputData().put("bo1", String.valueOf(task.getInputData().get("bi1")) + " from bi1"); result.getOutputData().put("bo2", String.valueOf(task.getInputData().get("bi2")) + " from bi2"); return result; } } 

运行main函数