refactor: 优化

Author: zhou-hao

hsweb-easy-orm-rdb/src/main/java/org/hswebframework/ezorm/rdb/executor/jdbc/JdbcReactiveSqlExecutor.java

@@ -42,7 +42,11 @@ public Mono<Integer> update(Publisher<SqlRequest> request) {
                         .toFlux(request)
                         .map(sql -> doUpdate(ctx.getOrDefault(Logger.class, log), connection, sql))
                         .reduce(Math::addExact)))
-            .last(0);
+            .last(0)
+            // 使用弹性线程池来执行jdbc操作
+            .subscribeOn(Schedulers.boundedElastic())
+            // 下游切换为parallel线程池,弹性线程池数据不可控，在虚拟线程等场景下，可能影响下游基于ThreadLocal等场景的缓存性能。
+            .publishOn(Schedulers.parallel());
 
     }
 
@@ -55,14 +59,15 @@ public Mono<Void> execute(Publisher<SqlRequest> request) {
                     doInConnection(connection -> this
                         .toFlux(request)
                         .doOnNext(sql -> doExecute(ctx.getOrDefault(Logger.class, log), connection, sql))))
-
+            .subscribeOn(Schedulers.boundedElastic())
+            .publishOn(Schedulers.parallel())
             .then();
     }
 
     @Override
     public <E> Flux<E> select(Publisher<SqlRequest> request, ResultWrapper<E, ?> wrapper) {
         return Flux
-            .deferContextual(ctx -> {
+            .<E>deferContextual(ctx -> {
                 Logger logger = ctx.getOrDefault(Logger.class, log);
                 return Flux
                     .create(sink -> {
@@ -92,7 +97,8 @@ public <E> Flux<E> select(Publisher<SqlRequest> request, ResultWrapper<E, ?> wra
                         sink
                             .onDispose(disposable);
                     });
-            });
+            })
+            .publishOn(Schedulers.parallel());
 
     }
 

hsweb-easy-orm-rdb/src/test/java/org/hswebframework/ezorm/rdb/TestJdbcReactiveSqlExecutor.java

@@ -3,9 +3,12 @@
 import lombok.AllArgsConstructor;
 import org.hswebframework.ezorm.rdb.executor.SqlRequest;
 import org.hswebframework.ezorm.rdb.executor.jdbc.JdbcReactiveSqlExecutor;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.sql.Connection;
+import java.util.function.Function;
 
 @AllArgsConstructor
 public class TestJdbcReactiveSqlExecutor extends JdbcReactiveSqlExecutor {
@@ -22,5 +25,13 @@ public Mono<Connection> getConnection() {
                 );
     }
 
-
+    @Override
+    protected <T> Flux<T> doInConnection(Function<Connection, Publisher<T>> handler) {
+        return Flux
+            .using(
+                provider::getConnection,
+                handler,
+                provider::releaseConnect
+            );
+    }
 }

hsweb-easy-orm-rdb/src/test/java/org/hswebframework/ezorm/rdb/TestReactiveSqlExecutor.java

@@ -3,9 +3,13 @@
 import io.r2dbc.spi.Connection;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.ezorm.rdb.executor.reactive.r2dbc.R2dbcReactiveSqlExecutor;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.SignalType;
 
+import java.util.function.Function;
+
 @Slf4j
 public class TestReactiveSqlExecutor extends R2dbcReactiveSqlExecutor {
 
@@ -33,12 +37,20 @@ public TestReactiveSqlExecutor(R2dbcConnectionProvider provider) {
     @Override
     protected Mono<Connection> getConnection() {
         return provider.getConnection()
-                .doOnNext(connection -> log.debug("get connection {}", connection));
+                       .doOnNext(connection -> log.debug("get connection {}", connection));
+    }
+
+    @Override
+    protected <T> Flux<T> doInConnection(Function<Connection, Publisher<T>> handler) {
+        return Flux.usingWhen(
+            provider.getConnection(),
+            handler,
+            c -> Mono.fromRunnable(() -> provider.releaseConnection(c)));
     }
 
     @Override
     protected void releaseConnection(SignalType type, Connection connection) {
-         log.debug("release connection {}", connection);
+        log.debug("release connection {}", connection);
         provider.releaseConnection(connection);
     }
 

hsweb-easy-orm-rdb/src/test/java/org/hswebframework/ezorm/rdb/executor/jdbc/JdbcReactiveSqlExecutorTest.java

@@ -40,7 +40,8 @@ public void testExecute() {
                 .update(Flux.range(0, 10)
                         .doOnNext(i -> System.out.println())
                         .map(num -> prepare("insert into test (id) values (?) ", num)))
-                .doOnError(Throwable::printStackTrace);
+                .doOnError(Throwable::printStackTrace)
+                .doFinally(s-> System.out.println(Thread.currentThread()));
 
         StepVerifier.create(counter)
                 .expectNext(10)