diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
index 4693e35831..dc3b6022ad 100644
--- a/RELEASE-NOTES.md
+++ b/RELEASE-NOTES.md
@@ -2,6 +2,8 @@
### 功能提升
+1. [ISSUE #352](https://github.com/dangdangdotcom/elastic-job/issues/352) Cloud作业本地运行
+
### 缺陷修正
1. [ISSUE #322](https://github.com/dangdangdotcom/elastic-job/issues/322) elastic-job-cloud-scheduler调度任务评估资源时考虑对executor的资源使用情况
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalCloudJobConfiguration.java b/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalCloudJobConfiguration.java
new file mode 100644
index 0000000000..0d7dab8c81
--- /dev/null
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalCloudJobConfiguration.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 1999-2015 dangdang.com.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.dangdang.ddframe.job.cloud.executor.local;
+
+import com.dangdang.ddframe.job.config.JobRootConfiguration;
+import com.dangdang.ddframe.job.config.JobTypeConfiguration;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * 本地云作业配置.
+ *
+ * @author gaohongtao
+ */
+@RequiredArgsConstructor
+@AllArgsConstructor
+@Getter
+public final class LocalCloudJobConfiguration implements JobRootConfiguration {
+
+ private final JobTypeConfiguration typeConfig;
+
+ private final LocalCloudJobExecutionType executionType;
+
+ private final int shardingItem;
+
+ private String beanName;
+
+ private String applicationContext;
+
+ /**
+ * 获取作业名称.
+ *
+ * @return 作业名称
+ */
+ public String getJobName() {
+ return typeConfig.getCoreConfig().getJobName();
+ }
+}
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalCloudJobExecutionType.java b/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalCloudJobExecutionType.java
new file mode 100644
index 0000000000..2976b2ee2a
--- /dev/null
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalCloudJobExecutionType.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 1999-2015 dangdang.com.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.dangdang.ddframe.job.cloud.executor.local;
+
+/**
+ * 本地作业执行类型.
+ *
+ * @author gaohongtao
+ */
+public enum LocalCloudJobExecutionType {
+
+ DAEMON, TRANSIENT
+}
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalTaskExecutor.java b/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalTaskExecutor.java
new file mode 100644
index 0000000000..56b151b251
--- /dev/null
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/main/java/com/dangdang/ddframe/job/cloud/executor/local/LocalTaskExecutor.java
@@ -0,0 +1,115 @@
+/*
+ * Copyright 1999-2015 dangdang.com.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.dangdang.ddframe.job.cloud.executor.local;
+
+import com.dangdang.ddframe.job.api.ElasticJob;
+import com.dangdang.ddframe.job.api.ShardingContext;
+import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
+import com.dangdang.ddframe.job.api.simple.SimpleJob;
+import com.dangdang.ddframe.job.config.JobCoreConfiguration;
+import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
+import com.dangdang.ddframe.job.exception.JobConfigurationException;
+import com.dangdang.ddframe.job.exception.JobSystemException;
+import com.dangdang.ddframe.job.executor.ShardingContexts;
+import com.dangdang.ddframe.job.util.config.ShardingItemParameters;
+import com.dangdang.ddframe.job.util.json.GsonFactory;
+import com.google.common.base.Strings;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.exec.CommandLine;
+import org.apache.commons.exec.DefaultExecutor;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static com.dangdang.ddframe.job.api.JobType.DATAFLOW;
+import static com.dangdang.ddframe.job.api.JobType.SIMPLE;
+
+/**
+ * 本地作业执行器.
+ *
+ * @author gaohongtao
+ */
+@RequiredArgsConstructor
+public final class LocalTaskExecutor {
+
+ private final LocalCloudJobConfiguration localCloudJobConfiguration;
+
+ /**
+ * 本地执行作业.
+ */
+ public void execute() {
+ if (SIMPLE == localCloudJobConfiguration.getTypeConfig().getJobType()) {
+ getJobInstance(SimpleJob.class).execute(getShardingContext());
+ } else if (DATAFLOW == localCloudJobConfiguration.getTypeConfig().getJobType()) {
+ processDataflow();
+ } else {
+ processScript();
+ }
+ }
+
+ private T getJobInstance(final Class clazz) {
+ if (Strings.isNullOrEmpty(localCloudJobConfiguration.getApplicationContext())) {
+ String jobClass = localCloudJobConfiguration.getTypeConfig().getJobClass();
+ try {
+ return clazz.cast(Class.forName(jobClass).newInstance());
+ } catch (final ReflectiveOperationException ex) {
+ throw new JobSystemException("Elastic-Job: Class '%s' initialize failure, the error message is '%s'.", jobClass, ex.getMessage());
+ }
+ } else {
+ return clazz.cast(new ClassPathXmlApplicationContext(localCloudJobConfiguration.getApplicationContext()).getBean(localCloudJobConfiguration.getBeanName()));
+ }
+ }
+
+ private ShardingContext getShardingContext() {
+ JobCoreConfiguration coreConfig = localCloudJobConfiguration.getTypeConfig().getCoreConfig();
+ String shardingItem = new ShardingItemParameters(coreConfig.getShardingItemParameters()).getMap().get(localCloudJobConfiguration.getShardingItem());
+ Map shardingItemMap = new HashMap<>(1);
+ if (!Strings.isNullOrEmpty(shardingItem)) {
+ shardingItemMap.put(localCloudJobConfiguration.getShardingItem(), shardingItem);
+ }
+ return new ShardingContext(new ShardingContexts("foo", localCloudJobConfiguration.getJobName(), coreConfig
+ .getShardingTotalCount(), coreConfig.getJobParameter(), shardingItemMap), localCloudJobConfiguration.getShardingItem());
+ }
+
+ @SuppressWarnings("unchecked")
+ private void processDataflow() {
+ final ShardingContext shardingContext = getShardingContext();
+ DataflowJob dataflowJob = getJobInstance(DataflowJob.class);
+ List data = dataflowJob.fetchData(shardingContext);
+ if (null != data && !data.isEmpty()) {
+ dataflowJob.processData(shardingContext, data);
+ }
+ }
+
+ private void processScript() {
+ final String scriptCommandLine = ((ScriptJobConfiguration) localCloudJobConfiguration.getTypeConfig()).getScriptCommandLine();
+ if (Strings.isNullOrEmpty(scriptCommandLine)) {
+ throw new JobConfigurationException("Cannot find script command line for job '%s', job is not executed.", localCloudJobConfiguration.getJobName());
+ }
+ CommandLine commandLine = CommandLine.parse(scriptCommandLine);
+ commandLine.addArgument(GsonFactory.getGson().toJson(getShardingContext()), false);
+ try {
+ new DefaultExecutor().execute(commandLine);
+ } catch (final IOException ex) {
+ throw new JobConfigurationException("Execute script failure.", ex);
+ }
+ }
+}
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/AllCloudExecutorTests.java b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/AllCloudExecutorTests.java
index cb6c845e74..888d3cdb09 100644
--- a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/AllCloudExecutorTests.java
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/AllCloudExecutorTests.java
@@ -17,6 +17,7 @@
package com.dangdang.ddframe.job.cloud.executor;
+import com.dangdang.ddframe.job.cloud.executor.local.AllLocalExecutorTests;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.junit.runner.RunWith;
@@ -29,7 +30,8 @@
DaemonTaskSchedulerTest.class,
JobConfigurationContextTest.class,
TaskExecutorTest.class,
- TaskExecutorThreadTest.class
+ TaskExecutorThreadTest.class,
+ AllLocalExecutorTests.class
})
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class AllCloudExecutorTests {
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/AllLocalExecutorTests.java b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/AllLocalExecutorTests.java
new file mode 100644
index 0000000000..d5e31ecd35
--- /dev/null
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/AllLocalExecutorTests.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 1999-2015 dangdang.com.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.dangdang.ddframe.job.cloud.executor.local;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses(LocalTaskExecutorTest.class)
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class AllLocalExecutorTests {
+}
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/LocalTaskExecutorTest.java b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/LocalTaskExecutorTest.java
new file mode 100644
index 0000000000..4bbdf2f577
--- /dev/null
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/LocalTaskExecutorTest.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 1999-2015 dangdang.com.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.dangdang.ddframe.job.cloud.executor.local;
+
+import com.dangdang.ddframe.job.cloud.executor.local.fixture.TestDataflowJob;
+import com.dangdang.ddframe.job.cloud.executor.local.fixture.TestSimpleJob;
+import com.dangdang.ddframe.job.config.JobCoreConfiguration;
+import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
+import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
+import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
+import com.dangdang.ddframe.job.exception.JobConfigurationException;
+import com.dangdang.ddframe.job.exception.JobSystemException;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+
+import static com.dangdang.ddframe.job.cloud.executor.local.LocalCloudJobExecutionType.DAEMON;
+import static com.dangdang.ddframe.job.cloud.executor.local.LocalCloudJobExecutionType.TRANSIENT;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+public class LocalTaskExecutorTest {
+
+ @Before
+ public void setUp() throws Exception {
+ TestSimpleJob.setShardingContext(null);
+ TestDataflowJob.setInput(null);
+ TestDataflowJob.setOutput(null);
+ }
+
+ @Test
+ public void assertSimpleJob() throws Exception {
+ new LocalTaskExecutor(new LocalCloudJobConfiguration(new SimpleJobConfiguration(JobCoreConfiguration
+ .newBuilder(TestSimpleJob.class.getSimpleName(), "*/2 * * * * ?", 3).build(), TestSimpleJob.class.getName()), TRANSIENT, 1)).execute();
+ assertThat(TestSimpleJob.getShardingContext().getJobName(), is(TestSimpleJob.class.getSimpleName()));
+ assertThat(TestSimpleJob.getShardingContext().getShardingItem(), is(1));
+ assertThat(TestSimpleJob.getShardingContext().getShardingTotalCount(), is(3));
+ assertNull(TestSimpleJob.getShardingContext().getShardingParameter());
+ assertThat(TestSimpleJob.getShardingContext().getJobParameter(), is(""));
+ }
+
+ @Test(expected = JobSystemException.class)
+ public void assertNotExistsJobClass() throws Exception {
+ new LocalTaskExecutor(new LocalCloudJobConfiguration(new SimpleJobConfiguration(JobCoreConfiguration
+ .newBuilder("not exist", "*/2 * * * * ?", 3).build(), "not exist"), TRANSIENT, 1)).execute();
+ }
+
+ @Test
+ public void assertSpringSimpleJob() throws Exception {
+ new LocalTaskExecutor(new LocalCloudJobConfiguration(new SimpleJobConfiguration(JobCoreConfiguration
+ .newBuilder(TestSimpleJob.class.getSimpleName(), "*/2 * * * * ?", 3)
+ .shardingItemParameters("0=Beijing,1=Shanghai,2=Guangzhou").jobParameter("dbName=dangdang").build(), TestSimpleJob.class
+ .getName()), DAEMON, 1, "testSimpleJob", "applicationContext.xml")).execute();
+ assertThat(TestSimpleJob.getShardingContext().getJobName(), is(TestSimpleJob.class.getSimpleName()));
+ assertThat(TestSimpleJob.getShardingContext().getShardingTotalCount(), is(3));
+ assertThat(TestSimpleJob.getShardingContext().getJobParameter(), is("dbName=dangdang"));
+ assertThat(TestSimpleJob.getShardingParameters().size(), is(1));
+ assertThat(TestSimpleJob.getShardingParameters().iterator().next(), is("Shanghai"));
+ }
+
+ @Test
+ public void assertDataflow() throws Exception {
+ TestDataflowJob.setInput(Arrays.asList("1", "2", "3"));
+ new LocalTaskExecutor(new LocalCloudJobConfiguration(new DataflowJobConfiguration(JobCoreConfiguration
+ .newBuilder(TestDataflowJob.class.getSimpleName(), "*/2 * * * * ?", 10).build(), TestDataflowJob.class.getName(), false), TRANSIENT, 5)).execute();
+ assertFalse(TestDataflowJob.getOutput().isEmpty());
+ for (String each : TestDataflowJob.getOutput()) {
+ assertTrue(each.endsWith("-d"));
+ }
+ }
+
+ @Test(expected = JobConfigurationException.class)
+ public void assertScriptEmpty() throws Exception {
+ new LocalTaskExecutor(new LocalCloudJobConfiguration(new ScriptJobConfiguration(JobCoreConfiguration
+ .newBuilder(TestDataflowJob.class.getSimpleName(), "*/2 * * * * ?", 10).build(), ""), TRANSIENT, 5)).execute();
+ }
+
+ @Test(expected = JobConfigurationException.class)
+ public void assertScriptNotExists() throws Exception {
+ new LocalTaskExecutor(new LocalCloudJobConfiguration(new ScriptJobConfiguration(JobCoreConfiguration
+ .newBuilder(TestDataflowJob.class.getSimpleName(), "*/2 * * * * ?", 10).build(), "not_exists_file param1"), TRANSIENT, 5)).execute();
+ }
+}
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/fixture/TestDataflowJob.java b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/fixture/TestDataflowJob.java
new file mode 100644
index 0000000000..8758a8f3b6
--- /dev/null
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/fixture/TestDataflowJob.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 1999-2015 dangdang.com.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.dangdang.ddframe.job.cloud.executor.local.fixture;
+
+import com.dangdang.ddframe.job.api.ShardingContext;
+import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.List;
+
+public class TestDataflowJob implements DataflowJob {
+
+ @Setter
+ private static List input;
+
+ @Getter
+ @Setter
+ private static List output;
+
+ @Override
+ public List fetchData(final ShardingContext shardingContext) {
+ return input;
+ }
+
+ @Override
+ public void processData(final ShardingContext shardingContext, final List data) {
+ output = Lists.transform(input, new Function() {
+ @Override
+ public String apply(final String input) {
+ return input + "-d";
+ }
+ });
+ }
+}
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/fixture/TestSimpleJob.java b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/fixture/TestSimpleJob.java
new file mode 100644
index 0000000000..ab33eac50d
--- /dev/null
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/java/com/dangdang/ddframe/job/cloud/executor/local/fixture/TestSimpleJob.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 1999-2015 dangdang.com.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.dangdang.ddframe.job.cloud.executor.local.fixture;
+
+import com.dangdang.ddframe.job.api.ShardingContext;
+import com.dangdang.ddframe.job.api.simple.SimpleJob;
+import com.google.common.base.Strings;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+public class TestSimpleJob implements SimpleJob {
+
+ @Getter
+ @Setter
+ private static ShardingContext shardingContext;
+
+ @Getter
+ private static Set shardingParameters = new ConcurrentSkipListSet<>();
+
+ @Override
+ public void execute(final ShardingContext shardingContext) {
+ TestSimpleJob.shardingContext = shardingContext;
+ if (!Strings.isNullOrEmpty(shardingContext.getShardingParameter())) {
+ shardingParameters.add(shardingContext.getShardingParameter());
+ }
+ }
+}
diff --git a/elastic-job-cloud/elastic-job-cloud-executor/src/test/resources/applicationContext.xml b/elastic-job-cloud/elastic-job-cloud-executor/src/test/resources/applicationContext.xml
index 033830355a..2228a19673 100644
--- a/elastic-job-cloud/elastic-job-cloud-executor/src/test/resources/applicationContext.xml
+++ b/elastic-job-cloud/elastic-job-cloud-executor/src/test/resources/applicationContext.xml
@@ -5,4 +5,5 @@
http://www.springframework.org/schema/beans/spring-beans.xsd
">