疯狂的狮子Li
2 years ago
parent
c03c72c7d0
commit
594165da6a
@ -1,56 +0,0 @@
|
||||
package org.dromara.job.processors;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
import tech.powerjob.common.utils.NetUtils;
|
||||
import tech.powerjob.worker.core.processor.ProcessResult;
|
||||
import tech.powerjob.worker.core.processor.TaskContext;
|
||||
import tech.powerjob.worker.core.processor.TaskResult;
|
||||
import tech.powerjob.worker.core.processor.sdk.BroadcastProcessor;
|
||||
import tech.powerjob.worker.log.OmsLogger;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 广播处理器 示例
|
||||
*
|
||||
* @author tjq
|
||||
* @since 2020/4/17
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class BroadcastProcessorDemo implements BroadcastProcessor {
|
||||
|
||||
@Override
|
||||
public ProcessResult preProcess(TaskContext context) {
|
||||
System.out.println("===== BroadcastProcessorDemo#preProcess ======");
|
||||
context.getOmsLogger().info("BroadcastProcessorDemo#preProcess, current host: {}", NetUtils.getLocalHost());
|
||||
if ("rootFailed".equals(context.getJobParams())) {
|
||||
return new ProcessResult(false, "console need failed");
|
||||
} else {
|
||||
return new ProcessResult(true);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public ProcessResult process(TaskContext taskContext) throws Exception {
|
||||
OmsLogger logger = taskContext.getOmsLogger();
|
||||
System.out.println("===== BroadcastProcessorDemo#process ======");
|
||||
logger.info("BroadcastProcessorDemo#process, current host: {}", NetUtils.getLocalHost());
|
||||
long sleepTime = 1000;
|
||||
try {
|
||||
sleepTime = Long.parseLong(taskContext.getJobParams());
|
||||
} catch (Exception e) {
|
||||
logger.warn("[BroadcastProcessor] parse sleep time failed!", e);
|
||||
}
|
||||
Thread.sleep(Math.max(sleepTime, 1000));
|
||||
return new ProcessResult(true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ProcessResult postProcess(TaskContext context, List<TaskResult> taskResults) {
|
||||
System.out.println("===== BroadcastProcessorDemo#postProcess ======");
|
||||
context.getOmsLogger().info("BroadcastProcessorDemo#postProcess, current host: {}, taskResult: {}", NetUtils.getLocalHost(), taskResults);
|
||||
return new ProcessResult(true, "success");
|
||||
}
|
||||
}
|
||||
@ -1,41 +0,0 @@
|
||||
package org.dromara.job.processors;
|
||||
|
||||
import com.alibaba.fastjson.JSONObject;
|
||||
import org.springframework.stereotype.Component;
|
||||
import tech.powerjob.official.processors.util.CommonUtils;
|
||||
import tech.powerjob.worker.core.processor.ProcessResult;
|
||||
import tech.powerjob.worker.core.processor.TaskContext;
|
||||
import tech.powerjob.worker.core.processor.sdk.BasicProcessor;
|
||||
import tech.powerjob.worker.log.OmsLogger;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* LogTestProcessor
|
||||
*
|
||||
* @author tjq
|
||||
* @since 2022/9/18
|
||||
*/
|
||||
@Component
|
||||
public class LogTestProcessor implements BasicProcessor {
|
||||
|
||||
@Override
|
||||
public ProcessResult process(TaskContext context) throws Exception {
|
||||
|
||||
final OmsLogger omsLogger = context.getOmsLogger();
|
||||
final String parseParams = CommonUtils.parseParams(context);
|
||||
final JSONObject config = Optional.ofNullable(JSONObject.parseObject(parseParams)).orElse(new JSONObject());
|
||||
|
||||
final long loopTimes = Optional.ofNullable(config.getLong("loopTimes")).orElse(1000L);
|
||||
|
||||
for (int i = 0; i < loopTimes; i++) {
|
||||
omsLogger.debug("[DEBUG] one DEBUG log in {}", new Date());
|
||||
omsLogger.info("[INFO] one INFO log in {}", new Date());
|
||||
omsLogger.warn("[WARN] one WARN log in {}", new Date());
|
||||
omsLogger.error("[ERROR] one ERROR log in {}", new Date());
|
||||
}
|
||||
|
||||
return new ProcessResult(true);
|
||||
}
|
||||
}
|
||||
@ -1,93 +0,0 @@
|
||||
package org.dromara.job.processors;
|
||||
|
||||
import com.google.common.collect.Lists;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Getter;
|
||||
import lombok.NoArgsConstructor;
|
||||
import org.dromara.common.json.utils.JsonUtils;
|
||||
import org.springframework.stereotype.Component;
|
||||
import tech.powerjob.worker.core.processor.ProcessResult;
|
||||
import tech.powerjob.worker.core.processor.TaskContext;
|
||||
import tech.powerjob.worker.core.processor.sdk.MapProcessor;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ThreadLocalRandom;
|
||||
|
||||
/**
|
||||
* Map处理器 示例
|
||||
*
|
||||
* @author tjq
|
||||
* @since 2020/4/18
|
||||
*/
|
||||
@Component
|
||||
public class MapProcessorDemo implements MapProcessor {
|
||||
|
||||
|
||||
/**
|
||||
* 每一批发送任务大小
|
||||
*/
|
||||
private static final int BATCH_SIZE = 100;
|
||||
/**
|
||||
* 发送的批次
|
||||
*/
|
||||
private static final int BATCH_NUM = 5;
|
||||
|
||||
@Override
|
||||
public ProcessResult process(TaskContext context) throws Exception {
|
||||
|
||||
log.info("============== MapProcessorDemo#process ==============");
|
||||
log.info("isRootTask:{}", isRootTask());
|
||||
log.info("taskContext:{}", JsonUtils.toJsonString(context));
|
||||
|
||||
if (isRootTask()) {
|
||||
log.info("==== MAP ====");
|
||||
List<SubTask> subTasks = Lists.newLinkedList();
|
||||
for (int j = 0; j < BATCH_NUM; j++) {
|
||||
SubTask subTask = new SubTask();
|
||||
subTask.siteId = j;
|
||||
subTask.itemIds = Lists.newLinkedList();
|
||||
subTasks.add(subTask);
|
||||
for (int i = 0; i < BATCH_SIZE; i++) {
|
||||
subTask.itemIds.add(i + j * 100);
|
||||
}
|
||||
}
|
||||
map(subTasks, "MAP_TEST_TASK");
|
||||
return new ProcessResult(true, "map successfully");
|
||||
} else {
|
||||
|
||||
log.info("==== PROCESS ====");
|
||||
SubTask subTask = (SubTask) context.getSubTask();
|
||||
for (Integer itemId : subTask.getItemIds()) {
|
||||
if (Thread.interrupted()) {
|
||||
// 任务被中断
|
||||
log.info("job has been stop! so stop to process subTask: {} => {}", subTask.getSiteId(), itemId);
|
||||
break;
|
||||
}
|
||||
log.info("processing subTask: {} => {}", subTask.getSiteId(), itemId);
|
||||
int max = Integer.MAX_VALUE >> 7;
|
||||
for (int i = 0; ; i++) {
|
||||
// 模拟耗时操作
|
||||
if (i > max) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
// 测试在 Map 任务中追加上下文
|
||||
context.getWorkflowContext().appendData2WfContext("Yasuo", "A sword's poor company for a long road.");
|
||||
boolean b = ThreadLocalRandom.current().nextBoolean();
|
||||
if (context.getCurrentRetryTimes() >= 1) {
|
||||
// 重试的话一定会成功
|
||||
b = true;
|
||||
}
|
||||
return new ProcessResult(b, "RESULT:" + b);
|
||||
}
|
||||
}
|
||||
|
||||
@Getter
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public static class SubTask {
|
||||
private Integer siteId;
|
||||
private List<Integer> itemIds;
|
||||
}
|
||||
}
|
||||
@ -1,35 +0,0 @@
|
||||
package org.dromara.job.processors;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
import tech.powerjob.worker.core.processor.ProcessResult;
|
||||
import tech.powerjob.worker.core.processor.TaskContext;
|
||||
import tech.powerjob.worker.core.processor.sdk.BasicProcessor;
|
||||
import tech.powerjob.worker.log.OmsLogger;
|
||||
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* @author Echo009
|
||||
* @since 2022/4/27
|
||||
*/
|
||||
@Component
|
||||
public class SimpleProcessor implements BasicProcessor {
|
||||
|
||||
@Override
|
||||
public ProcessResult process(TaskContext context) throws Exception {
|
||||
|
||||
OmsLogger logger = context.getOmsLogger();
|
||||
|
||||
String jobParams = Optional.ofNullable(context.getJobParams()).orElse("S");
|
||||
logger.info("Current context:{}", context.getWorkflowContext());
|
||||
logger.info("Current job params:{}", jobParams);
|
||||
|
||||
// 测试中文问题 #581
|
||||
if (jobParams.contains("CN")) {
|
||||
return new ProcessResult(true, "任务成功啦！！！");
|
||||
}
|
||||
|
||||
return jobParams.contains("F") ? new ProcessResult(false) : new ProcessResult(true, "yeah!");
|
||||
|
||||
}
|
||||
}
|
||||
@ -1,36 +0,0 @@
|
||||
package org.dromara.job.workflow;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
import tech.powerjob.worker.core.processor.ProcessResult;
|
||||
import tech.powerjob.worker.core.processor.TaskContext;
|
||||
import tech.powerjob.worker.core.processor.sdk.BasicProcessor;
|
||||
import tech.powerjob.worker.log.OmsLogger;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 工作流测试
|
||||
*
|
||||
* @author tjq
|
||||
* @since 2020/6/2
|
||||
*/
|
||||
@Component
|
||||
@Slf4j
|
||||
public class WorkflowStandaloneProcessor implements BasicProcessor {
|
||||
|
||||
@Override
|
||||
public ProcessResult process(TaskContext context) throws Exception {
|
||||
OmsLogger logger = context.getOmsLogger();
|
||||
logger.info("current jobParams: {}", context.getJobParams());
|
||||
logger.info("current context: {}", context.getWorkflowContext());
|
||||
log.info("jobParams:{}", context.getJobParams());
|
||||
log.info("currentContext:{}", JSON.toJSONString(context));
|
||||
|
||||
// 尝试获取上游任务
|
||||
Map<String, String> workflowContext = context.getWorkflowContext().fetchWorkflowContext();
|
||||
log.info("工作流上下文数据:{}", workflowContext);
|
||||
return new ProcessResult(true, context.getJobId() + " process successfully.");
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue