From 45c5299cb1a830ee1fb5ef81a6375cbb4c003003 Mon Sep 17 00:00:00 2001 From: zhouhao Date: Fri, 1 Sep 2017 15:40:11 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=BC=82=E6=AD=A5=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../hsweb-concurrent-async-job/README.md | 24 ++++ .../hsweb-concurrent-async-job/pom.xml | 37 ++++++ .../web/async/AsyncJobException.java | 41 +++++++ .../AsyncJobServiceAutoConfiguration.java | 49 ++++++++ .../web/async/BatchAsyncJobContainer.java | 10 +- .../async/SpringTransactionSupportJob.java | 52 ++++++++ .../SpringTransactionSupportJobWrapper.java | 24 ++++ .../TransactionBatchAsyncJobContainer.java | 114 ++++++++++++++++++ .../TransactionSupportAsyncJobService.java | 30 +++++ ...ortJob.java => TransactionSupportJob.java} | 5 +- ...java => TransactionSupportJobWrapper.java} | 4 +- .../TranslationBatchAsyncJobContainer.java | 82 ------------- .../main/resources/META-INF/spring.factories | 3 + ...TransactionSupportAsyncJobServiceTest.java | 66 ++++++++++ .../src/test/resources/application.yml | 18 +++ 15 files changed, 470 insertions(+), 89 deletions(-) create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/README.md create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/AsyncJobException.java create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/AsyncJobServiceAutoConfiguration.java create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/SpringTransactionSupportJob.java create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/SpringTransactionSupportJobWrapper.java create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionBatchAsyncJobContainer.java create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportAsyncJobService.java rename hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/{TranslationSupportJob.java => TransactionSupportJob.java} (55%) rename hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/{TranslationSupportJobWrapper.java => TransactionSupportJobWrapper.java} (53%) delete mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationBatchAsyncJobContainer.java create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/main/resources/META-INF/spring.factories create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/test/java/org/hswebframework/web/async/TransactionSupportAsyncJobServiceTest.java create mode 100644 hsweb-concurrent/hsweb-concurrent-async-job/src/test/resources/application.yml diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/README.md b/hsweb-concurrent/hsweb-concurrent-async-job/README.md new file mode 100644 index 000000000..b5abc6364 --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/README.md @@ -0,0 +1,24 @@ +# 异步任务工具,支持多线程事务 +```xml + + + org.hswebframework.web + hsweb-concurrent-async-job + ${project.verion} + +``` + +```java + @Autowired + private AsyncJobService asyncJobService; + + + public void testJob(){ + List results= asyncJobService.batch() + .submit(()->...) //提交job + .submit(()->...) //提交另外一个job + .submit(()->...,true) //提交支持事务的job + .getResult(); + } + +``` \ No newline at end of file diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/pom.xml b/hsweb-concurrent/hsweb-concurrent-async-job/pom.xml index 1e04aebb6..227a4b17b 100644 --- a/hsweb-concurrent/hsweb-concurrent-async-job/pom.xml +++ b/hsweb-concurrent/hsweb-concurrent-async-job/pom.xml @@ -11,5 +11,42 @@ hsweb-concurrent-async-job + + + org.springframework + spring-tx + + + org.springframework.boot + spring-boot-starter + + + org.slf4j + slf4j-api + + + + ch.qos.logback + logback-classic + provided + + + + org.hswebframework.web + hsweb-tests + ${project.version} + test + + + org.codehaus.groovy + groovy-all + test + + + com.h2database + h2 + test + + \ No newline at end of file diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/AsyncJobException.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/AsyncJobException.java new file mode 100644 index 000000000..b7b0c7257 --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/AsyncJobException.java @@ -0,0 +1,41 @@ +package org.hswebframework.web.async; + +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * TODO 完成注释 + * + * @author zhouhao + */ +public class AsyncJobException extends RuntimeException { + List errors; + + public AsyncJobException(List errors) { + this.errors = errors; + } + + public AsyncJobException(String message, List errors) { + super(message); + this.errors = errors; + } + + public List getErrors() { + return errors; + } + + @Override + public StackTraceElement[] getStackTrace() { + if (null != errors) { + List stackTraceElements = errors.stream() + .map(Exception::getStackTrace) + .flatMap(Stream::of) + .collect(Collectors.toList()); + stackTraceElements.addAll(Arrays.asList(super.getStackTrace())); + return stackTraceElements.toArray(new StackTraceElement[stackTraceElements.size()]); + } + return super.getStackTrace(); + } +} diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/AsyncJobServiceAutoConfiguration.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/AsyncJobServiceAutoConfiguration.java new file mode 100644 index 000000000..47dacaf3d --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/AsyncJobServiceAutoConfiguration.java @@ -0,0 +1,49 @@ +package org.hswebframework.web.async; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * TODO 完成注释 + * + * @author zhouhao + */ +@Configuration +@ConditionalOnMissingBean(AsyncJobService.class) +public class AsyncJobServiceAutoConfiguration { + + @Value("${hsweb.async.job.maxThreadPoolSize:-1}") + private int maxThreadPoolSize = -1; + + @Bean + @ConditionalOnMissingBean(ExecutorService.class) + public ExecutorService executorService() { + if (maxThreadPoolSize == -1) { + maxThreadPoolSize = Runtime.getRuntime().availableProcessors() * 50; + } + return Executors.newFixedThreadPool(maxThreadPoolSize); + } + + @Bean + @ConditionalOnMissingBean(TransactionSupportJobWrapper.class) + public TransactionSupportJobWrapper transactionSupportJobWrapper() { + return new SpringTransactionSupportJobWrapper(); + } + + @Bean + public AsyncJobService asyncJobService(ExecutorService executorService, TransactionSupportJobWrapper transactionSupportJobWrapper) { + TransactionSupportAsyncJobService asyncJobService = new TransactionSupportAsyncJobService(); + + asyncJobService.setTranslationSupportJobWrapper(transactionSupportJobWrapper); + asyncJobService.setExecutorService(executorService); + + return asyncJobService; + + } + +} diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/BatchAsyncJobContainer.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/BatchAsyncJobContainer.java index 937d4d58b..9ced5a349 100644 --- a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/BatchAsyncJobContainer.java +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/BatchAsyncJobContainer.java @@ -4,13 +4,17 @@ import java.util.List; import java.util.concurrent.Callable; /** - * TODO 完成注释 - * * @author zhouhao */ public interface BatchAsyncJobContainer { - BatchAsyncJobContainer submit(Callable callable); + default BatchAsyncJobContainer submit(Callable callable) { + submit(callable, false); + return this; + } + + BatchAsyncJobContainer submit(Callable callable, boolean enableTransaction); + List getResult() throws Exception; } diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/SpringTransactionSupportJob.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/SpringTransactionSupportJob.java new file mode 100644 index 000000000..c7b60c2a5 --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/SpringTransactionSupportJob.java @@ -0,0 +1,52 @@ +package org.hswebframework.web.async; + +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; + +import java.util.concurrent.Callable; + +/** + * TODO 完成注释 + * + * @author zhouhao + */ +public class SpringTransactionSupportJob implements TransactionSupportJob { + + private TransactionStatus transactionStatus; + + private TransactionTemplate transactionTemplate; + + private boolean rollback = false; + + private Callable target; + + public SpringTransactionSupportJob(TransactionTemplate transactionTemplate, Callable job) { + this.transactionTemplate = transactionTemplate; + this.target = job; + } + + @Override + public void rollBackOnly() { + rollback = true; + } + + @Override + public void commit() { + //do noting + if (transactionStatus != null) { + if (rollback) { + transactionTemplate.getTransactionManager().rollback(transactionStatus); + } else { + transactionTemplate.getTransactionManager().commit(transactionStatus); + } + } + } + + @Override + public V call() throws Exception { + transactionStatus = transactionTemplate.getTransactionManager().getTransaction(transactionTemplate); + if (rollback) transactionStatus.setRollbackOnly(); + return target.call(); + } +} diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/SpringTransactionSupportJobWrapper.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/SpringTransactionSupportJobWrapper.java new file mode 100644 index 000000000..c1858ab9a --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/SpringTransactionSupportJobWrapper.java @@ -0,0 +1,24 @@ +package org.hswebframework.web.async; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.transaction.support.TransactionTemplate; + +import java.util.concurrent.Callable; + +/** + * @author zhouhao + */ +public class SpringTransactionSupportJobWrapper implements TransactionSupportJobWrapper { + + private TransactionTemplate transactionTemplate; + + @Autowired + public void setTransactionTemplate(TransactionTemplate transactionTemplate) { + this.transactionTemplate = transactionTemplate; + } + + @Override + public TransactionSupportJob wrapper(Callable callable) { + return new SpringTransactionSupportJob<>(transactionTemplate, callable); + } +} diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionBatchAsyncJobContainer.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionBatchAsyncJobContainer.java new file mode 100644 index 000000000..b900e2aea --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionBatchAsyncJobContainer.java @@ -0,0 +1,114 @@ +package org.hswebframework.web.async; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; + +/** + * @author zhouhao + */ +public class TransactionBatchAsyncJobContainer implements BatchAsyncJobContainer { + + private ExecutorService executorService; + private TransactionSupportJobWrapper translationSupportJobWrapper; + private static final Logger logger = LoggerFactory.getLogger(TransactionBatchAsyncJobContainer.class); + + public TransactionBatchAsyncJobContainer(ExecutorService executorService, TransactionSupportJobWrapper translationSupportJobWrapper) { + this.executorService = executorService; + this.translationSupportJobWrapper = translationSupportJobWrapper; + } + + private List exceptions = new ArrayList<>(); + + private AtomicInteger failCounter = new AtomicInteger(); + + private AtomicInteger transactionJobOverCounter = new AtomicInteger(); + + private CountDownLatch countDownLatch = new CountDownLatch(1); + + private List futures = new ArrayList<>(); + + private int transactionJobNumber = 0; + + public void setExecutorService(ExecutorService executorService) { + this.executorService = executorService; + } + + @Override + public BatchAsyncJobContainer submit(Callable callable, boolean enableTransaction) { + if (!enableTransaction) { + if (logger.isDebugEnabled()) { + logger.debug("submit not transaction support job {}", transactionJobNumber); + } + futures.add(executorService.submit(callable)); + return this; + } + transactionJobNumber++; + if (logger.isDebugEnabled()) { + logger.debug("submit transaction support job {}", transactionJobNumber); + } + int tmpJobFlag = transactionJobNumber; + + TransactionSupportJob translationJob = translationSupportJobWrapper.wrapper(callable); + Callable proxy = () -> { + V value = null; + try { + if (failCounter.get() > 0) { + return null; + } + value = translationJob.call(); + transactionJobOverCounter.incrementAndGet(); + if (logger.isDebugEnabled()) { + logger.debug("transaction support job {} success,wait...", tmpJobFlag); + } + //等待其他任务完成 + countDownLatch.await(); + if (failCounter.get() > 0) { + if (logger.isDebugEnabled()) { + logger.debug("transaction support job {} success,but other job failed, do rollback only!", tmpJobFlag); + } + translationJob.rollBackOnly(); + } else { + if (logger.isDebugEnabled()) { + logger.debug("transaction support job {} success,commit.", tmpJobFlag); + } + } + translationJob.commit(); + } catch (Exception e) { + exceptions.add(e); + failCounter.incrementAndGet(); + transactionJobOverCounter.incrementAndGet(); + logger.warn("transaction support job {} fail.", tmpJobFlag, e); + } + return value; + }; + futures.add(executorService.submit(proxy)); + return this; + } + + @Override + public List getResult() throws Exception { + while (transactionJobOverCounter.get() != transactionJobNumber) { + Thread.sleep(50); + } + countDownLatch.countDown(); + if (!exceptions.isEmpty()) { + throw new AsyncJobException(exceptions); + } + + return futures.stream().map(this::getValue).collect(Collectors.toList()); + } + + private Object getValue(Future future) { + try { + return future.get(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } +} diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportAsyncJobService.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportAsyncJobService.java new file mode 100644 index 000000000..583a7e8ae --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportAsyncJobService.java @@ -0,0 +1,30 @@ +package org.hswebframework.web.async; + +import org.springframework.beans.factory.annotation.Autowired; + +import java.util.concurrent.ExecutorService; + +/** + * @author zhouhao + */ +public class TransactionSupportAsyncJobService implements AsyncJobService { + + private ExecutorService executorService; + + private TransactionSupportJobWrapper translationSupportJobWrapper; + + @Autowired(required = false) + public void setExecutorService(ExecutorService executorService) { + this.executorService = executorService; + } + + @Autowired + public void setTranslationSupportJobWrapper(TransactionSupportJobWrapper translationSupportJobWrapper) { + this.translationSupportJobWrapper = translationSupportJobWrapper; + } + + @Override + public BatchAsyncJobContainer batch() { + return new TransactionBatchAsyncJobContainer(executorService, translationSupportJobWrapper); + } +} diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationSupportJob.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportJob.java similarity index 55% rename from hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationSupportJob.java rename to hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportJob.java index 13ee0373a..e0f5be275 100644 --- a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationSupportJob.java +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportJob.java @@ -7,8 +7,9 @@ import java.util.concurrent.Callable; * * @author zhouhao */ -public interface TranslationSupportJob extends Callable { +public interface TransactionSupportJob extends Callable { - void rollBack(); + void rollBackOnly(); + void commit(); } diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationSupportJobWrapper.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportJobWrapper.java similarity index 53% rename from hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationSupportJobWrapper.java rename to hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportJobWrapper.java index ec5332dd4..afdd748a6 100644 --- a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationSupportJobWrapper.java +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TransactionSupportJobWrapper.java @@ -7,6 +7,6 @@ import java.util.concurrent.Callable; * * @author zhouhao */ -public interface TranslationSupportJobWrapper { - TranslationSupportJob wrapper(Callable callable); +public interface TransactionSupportJobWrapper { + TransactionSupportJob wrapper(Callable callable); } diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationBatchAsyncJobContainer.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationBatchAsyncJobContainer.java deleted file mode 100644 index 835b4ba96..000000000 --- a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/java/org/hswebframework/web/async/TranslationBatchAsyncJobContainer.java +++ /dev/null @@ -1,82 +0,0 @@ -package org.hswebframework.web.async; - -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Supplier; -import java.util.stream.Collectors; - -/** - * @author zhouhao - */ -public class TranslationBatchAsyncJobContainer implements BatchAsyncJobContainer { - - private ExecutorService executorService; - - private List exceptions = new ArrayList<>(); - - private Supplier supportJobSupplierBuilder; - - private CountDownLatch downLatch = new CountDownLatch(1); - - private AtomicInteger failCounter = new AtomicInteger(); - - private AtomicInteger overCounter = new AtomicInteger(); - - private List futures = new ArrayList<>(); - - public void setExecutorService(ExecutorService executorService) { - this.executorService = executorService; - } - - @Override - public BatchAsyncJobContainer submit(Callable callable) { - TranslationSupportJob translationJob = supportJobSupplierBuilder - .get() - .wrapper(callable); - Callable proxy = () -> { - try { - if (failCounter.get() > 0) { - return null; - } - V val = translationJob.call(); - overCounter.incrementAndGet(); - downLatch.await(); - if (failCounter.get() > 0) { - translationJob.rollBack(); - return null; - } - return val; - } catch (Exception e) { - exceptions.add(e); - failCounter.incrementAndGet(); - overCounter.incrementAndGet(); - } - return null; - }; - futures.add(executorService.submit(proxy)); - return this; - } - - @Override - public List getResult() throws Exception { - while (overCounter.get() != futures.size()) { - Thread.sleep(10); - } - downLatch.countDown(); - if (!exceptions.isEmpty()) { - throw new RuntimeException(); - } - - return futures.stream().map(this::getValue).collect(Collectors.toList()); - } - - private Object getValue(Future future) { - try { - return future.get(); - } catch (Exception e) { - throw new RuntimeException(e); - } - } -} diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/main/resources/META-INF/spring.factories b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..8fe3b1ec0 --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/main/resources/META-INF/spring.factories @@ -0,0 +1,3 @@ +# Auto Configure +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +org.hswebframework.web.async.AsyncJobServiceAutoConfiguration \ No newline at end of file diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/test/java/org/hswebframework/web/async/TransactionSupportAsyncJobServiceTest.java b/hsweb-concurrent/hsweb-concurrent-async-job/src/test/java/org/hswebframework/web/async/TransactionSupportAsyncJobServiceTest.java new file mode 100644 index 000000000..277c26655 --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/test/java/org/hswebframework/web/async/TransactionSupportAsyncJobServiceTest.java @@ -0,0 +1,66 @@ +package org.hswebframework.web.async; + +import org.hsweb.ezorm.rdb.executor.SqlExecutor; +import org.hswebframework.web.tests.SimpleWebApplicationTests; +import org.junit.Assert; +import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.transaction.support.TransactionTemplate; + +import javax.sql.DataSource; +import java.sql.SQLException; +import java.util.List; +import java.util.concurrent.Executors; + +import static org.junit.Assert.*; + +/** + * TODO 完成注释 + * + * @author zhouhao + */ +public class TransactionSupportAsyncJobServiceTest extends SimpleWebApplicationTests { + + @Autowired + private SqlExecutor sqlExecutor; + + @Autowired + private AsyncJobService asyncJobService; + + @Configuration + @EnableConfigurationProperties(DataSourceProperties.class) + public static class DataSourceConfig { + @Bean + @ConfigurationProperties(prefix = "spring.datasource") + public DataSource dataSource(DataSourceProperties properties) { + return properties.initializeDataSourceBuilder().build(); + } + } + + @Test + public void test() throws Exception { + + sqlExecutor.exec("create table test(id varchar(32))"); + + try { + BatchAsyncJobContainer jobContainer = asyncJobService.batch(); + + for (int i = 0; i < 100; i++) { + jobContainer.submit(() -> sqlExecutor.insert("insert into test values('test')", null), true); + } + jobContainer.submit(() -> { + Thread.sleep(200); + throw new RuntimeException(); + }, true); + System.out.println(jobContainer.getResult().size()); + } catch (Exception ignore) { + } + Assert.assertTrue(sqlExecutor.list("select * from test").isEmpty()); + } +} \ No newline at end of file diff --git a/hsweb-concurrent/hsweb-concurrent-async-job/src/test/resources/application.yml b/hsweb-concurrent/hsweb-concurrent-async-job/src/test/resources/application.yml new file mode 100644 index 000000000..f2eac679f --- /dev/null +++ b/hsweb-concurrent/hsweb-concurrent-async-job/src/test/resources/application.yml @@ -0,0 +1,18 @@ +spring: + aop: + auto: true + datasource: + url : jdbc:h2:mem:async_test_mem + username : sa + password : + type: com.alibaba.druid.pool.DruidDataSource + driver-class-name : org.h2.Driver + initial-size: 100 + max-active: 2000 +hsweb: + app: + name: 异步任务 + version: 3.0.0 +logging: + level: + org.hswebframework: debug \ No newline at end of file