refactor: 优化事务工具

zhou-hao · zhou-hao · commit e9a3104db358 · 2025-12-19T11:31:48.000+08:00
diff --git a/hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/configuration/R2dbcSqlExecutorConfiguration.java b/hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/configuration/R2dbcSqlExecutorConfiguration.java
@@ -5,16 +5,17 @@
 import org.hswebframework.ezorm.rdb.executor.reactive.ReactiveSqlExecutor;
 import org.hswebframework.ezorm.rdb.executor.reactive.ReactiveSyncSqlExecutor;
 import org.hswebframework.web.crud.sql.DefaultR2dbcExecutor;
+import org.hswebframework.web.crud.utils.TransactionUtils;
+import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
+import org.springframework.transaction.ReactiveTransactionManager;
 
 @AutoConfiguration
-@AutoConfigureAfter(name = "org.springframework.boot.autoconfigure.r2dbc.R2dbcAutoConfiguration")
+@AutoConfigureAfter(name = "org.springframeworrk.boot.autoconfigure.r2dbc.R2dbcAutoConfiguration")
 @ConditionalOnBean(ConnectionFactory.class)
 public class R2dbcSqlExecutorConfiguration {
     @Bean
@@ -31,4 +32,10 @@ public ReactiveSqlExecutor reactiveSqlExecutor(EasyormProperties properties) {
     public SyncSqlExecutor syncSqlExecutor(ReactiveSqlExecutor reactiveSqlExecutor) {
         return ReactiveSyncSqlExecutor.of(reactiveSqlExecutor);
     }
+
+    @Bean
+    public SmartInitializingSingleton transactionUtilsSetup(ReactiveTransactionManager transactionManager){
+        TransactionUtils.setup(transactionManager);
+        return ()->{};
+    }
 }
diff --git a/hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/utils/TransactionUtils.java b/hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/utils/TransactionUtils.java
@@ -3,29 +3,70 @@
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.transaction.NoTransactionException;
-import org.springframework.transaction.reactive.TransactionContextManager;
+import org.springframework.transaction.ReactiveTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionManager;
 import org.springframework.transaction.reactive.TransactionSynchronization;
 import org.springframework.transaction.reactive.TransactionSynchronizationManager;
+import org.springframework.transaction.reactive.TransactionalOperator;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.function.Function;
 
+import static org.springframework.transaction.TransactionDefinition.PROPAGATION_REQUIRES_NEW;
+
 @Slf4j
 public class TransactionUtils {
 
+    static TransactionManager transactionManager;
+
+    public static void setup(TransactionManager transactionManager) {
+        TransactionUtils.transactionManager = transactionManager;
+    }
+
+    public static <T> Mono<T> tryRunInTransaction(Mono<T> task, TransactionDefinition definition) {
+        if (transactionManager instanceof ReactiveTransactionManager tx) {
+            TransactionalOperator requiresNew =
+                TransactionalOperator.create(
+                    tx,
+                    definition);
+            return requiresNew.transactional(task);
+        }
+        return task;
+    }
+
+    public static <T> Flux<T> tryRunInTransaction(Flux<T> task, TransactionDefinition definition) {
+        if (transactionManager instanceof ReactiveTransactionManager tx) {
+            TransactionalOperator requiresNew =
+                TransactionalOperator.create(
+                    tx,
+                    definition);
+            return requiresNew.transactional(task);
+        }
+        return task;
+    }
+
     public static Mono<Void> afterCommit(Mono<Void> task) {
         return TransactionUtils.registerSynchronization(
             new TransactionSynchronization() {
                 @Override
                 @NonNull
                 public Mono<Void> afterCommit() {
-                    return task;
+                    return tryRunInTransaction(task, new DefaultTransactionDefinition(PROPAGATION_REQUIRES_NEW));
                 }
             },
             TransactionSynchronization::afterCommit
         );
     }
 
+    /**
+     * @param synchronization   TransactionSynchronization
+     * @param whenNoTransaction TransactionSynchronization
+     * @return TransactionSynchronization
+     * @see TransactionUtils#tryRunInTransaction(Flux, TransactionDefinition)
+     */
     public static Mono<Void> registerSynchronization(TransactionSynchronization synchronization,
                                                      Function<TransactionSynchronization, Mono<Void>> whenNoTransaction) {
         return TransactionSynchronizationManager
diff --git a/hsweb-commons/hsweb-commons-crud/src/test/java/org/hswebframework/web/crud/CrudTests.java b/hsweb-commons/hsweb-commons-crud/src/test/java/org/hswebframework/web/crud/CrudTests.java
@@ -1,27 +1,45 @@
 package org.hswebframework.web.crud;
 
+import lombok.SneakyThrows;
 import org.hswebframework.web.crud.entity.CustomTestEntity;
 import org.hswebframework.web.crud.entity.TestEntity;
 import org.hswebframework.web.crud.events.EntityBeforeModifyEvent;
 import org.hswebframework.web.crud.service.CustomTestCustom;
 import org.hswebframework.web.crud.service.TestEntityService;
+import org.hswebframework.web.crud.utils.TransactionUtils;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.context.event.EventListener;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.transaction.reactive.TransactionalOperator;
+import reactor.core.Disposable;
+import reactor.core.Disposables;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
+import reactor.util.context.Context;
 
-@SpringBootTest(classes = {TestApplication.class,TestEntityService.class, CustomTestCustom.class})
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+@SpringBootTest(classes = {TestApplication.class, TestEntityService.class, CustomTestCustom.class},
+    properties = {
+        "spring.r2dbc.pool.enabled=true",
+        "spring.r2dbc.pool.max-size=32",
+        "logging.level.org.springframework.r2dbc.connection=debug"
+    })
 @RunWith(SpringJUnit4ClassRunner.class)
 public class CrudTests {
 
     @Autowired
     private TestEntityService service;
 
+    @Autowired
+    private TransactionalOperator transactionalOperator;
 
     @Test
     public void test() {
@@ -53,4 +71,52 @@ public void test() {
                .expectNext(1)
                .verifyComplete();
     }
+
+    @Test
+    @SneakyThrows
+    public void testMultiThread() {
+        Flux.range(0, 100)
+            .map(e -> {
+                CustomTestEntity entity = new CustomTestEntity();
+                entity.setExt("xxx-" + e);
+                entity.setAge(1);
+                entity.setName("mt-" + e);
+                return entity;
+            })
+            .cast(TestEntity.class)
+            .as(service::save)
+            .block();
+
+        Disposable.Swap disposable = Disposables.swap();
+
+        CountDownLatch latch = new CountDownLatch(50);
+        disposable.update(
+            service
+                .createQuery()
+                .like(CustomTestEntity::getName, "mt-%")
+                .fetch()
+                .flatMap(e -> service
+                    .updateById(e.getId(), e)
+                    .flatMap(i -> TransactionUtils
+                        .afterCommit(Mono.deferContextual((c) -> service
+                             .updateById(e.getId(), e)
+                             .doOnNext(x -> {
+                                 latch.countDown();
+                                 if (latch.getCount() <= 0) {
+                                     disposable.dispose();
+                                 }
+                             })
+                             //.as(transactionalOperator::transactional)
+                             .subscribeOn(Schedulers.boundedElastic())
+                             .then())))
+                    // .subscribeOn(Schedulers.boundedElastic())
+                )
+               // .as(transactionalOperator::transactional)
+                .subscribe()
+        );
+        Assert.assertTrue(latch.await(20, TimeUnit.SECONDS));
+
+        Thread.sleep(2000);
+
+    }
 }
diff --git a/hsweb-commons/hsweb-commons-crud/src/test/resources/application.yml b/hsweb-commons/hsweb-commons-crud/src/test/resources/application.yml
@@ -3,8 +3,11 @@ logging:
     org.hswebframework: debug
     org.springframework.transaction: debug
     org.springframework.data.r2dbc.connectionfactory: debug
-#spring:
-#  r2dbc:
+    "org.springframework.transaction.reactive": debug
+spring:
+  r2dbc:
+      pool:
+          max-acquire-time: 1s
 #
 easyorm:
   default-schema: PUBLIC
diff --git a/hsweb-core/src/main/java/org/hswebframework/web/bean/FastBeanCopier.java b/hsweb-core/src/main/java/org/hswebframework/web/bean/FastBeanCopier.java
@@ -1,6 +1,7 @@
 package org.hswebframework.web.bean;
 
 import com.google.common.collect.Maps;
+import io.netty.util.internal.ConcurrentSet;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.SneakyThrows;
@@ -24,6 +25,7 @@
 import java.lang.reflect.Field;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -550,7 +552,7 @@ public void setBeanFactory(BeanFactory beanFactory) {
 
         public Collection<?> newCollection(Class<?> targetClass) {
 
-            if (targetClass == List.class) {
+            if (targetClass == List.class || targetClass == Collection.class) {
                 return new ArrayList<>();
             } else if (targetClass == ConcurrentHashMap.KeySetView.class) {
                 return ConcurrentHashMap.newKeySet();
@@ -560,7 +562,7 @@ public Collection<?> newCollection(Class<?> targetClass) {
                 return new LinkedList<>();
             } else {
                 try {
-                    return (Collection<?>) targetClass.newInstance();
+                    return (Collection<?>) targetClass.getConstructor().newInstance();
                 } catch (Exception e) {
                     throw new UnsupportedOperationException("不支持的类型:" + targetClass, e);
                 }
