commit

2024-12-13 09:37:20 +08:00
parent 014beea7d4
commit c418e37618
20 changed files with 2043 additions and 230 deletions
@@ -0,0 +1,125 @@
+package io.v.nutz.base.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.nutz.dao.Dao;
+import org.nutz.ioc.loader.annotation.Inject;
+import org.nutz.ioc.loader.annotation.IocBean;
+import org.nutz.lang.Lang;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.List;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * @version 1.0
+ * @Author zzr
+ * @name：ThreadPoolConfig
+ * @Date 2024/12/6 14:35
+ * @注释
+ */
+@Slf4j
+@IocBean
+public class ThreadPoolConfig {
+
+    @Inject
+    private Dao dao;
+
+    /**
+     * 核心线程数（默认线程数）
+     */
+    private static final int CORE_POOL_SIZE = 2 * Runtime.getRuntime().availableProcessors() + 1;
+
+    /**
+     * 最大线程数
+     */
+    private static final int MAX_POOL_SIZE = 128;
+
+    /**
+     * 允许线程空闲时间（单位：默认为秒）
+     */
+    private static final int KEEP_ALIVE_TIME = 5;
+
+    /**
+     * 任务的等待时间
+     */
+    private static final int AWAIT_TERMINATION_TIME = 30;
+
+    /**
+     * 缓冲队列数
+     */
+    private static final int QUEUE_CAPACITY = 1200;
+
+    /**
+     * 线程池名前缀
+     */
+    private static final String THREAD_NAME_PREFIX = "dd3s-thread-pool";
+
+    /**
+     * bean的名称，默认为首字母小写的方法名
+     * spring管理的线程池，顶级父类也是Executor
+     */
+    @IocBean(name = "executorService")
+    public ThreadPoolTaskExecutor executorService() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(CORE_POOL_SIZE);
+        executor.setMaxPoolSize(MAX_POOL_SIZE);
+        executor.setQueueCapacity(QUEUE_CAPACITY);
+        executor.setKeepAliveSeconds(KEEP_ALIVE_TIME);
+        executor.setThreadNamePrefix(THREAD_NAME_PREFIX);
+        executor.setAwaitTerminationSeconds(AWAIT_TERMINATION_TIME);
+        executor.setWaitForTasksToCompleteOnShutdown(true);
+
+        // 线程池对拒绝任务的处理策略
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        executor.initialize();
+        return executor;
+    }
+
+    /**
+     * 异步执行
+     *
+     * @param list                           泛型list,任意实体类集合
+     * @param batchSize                      期望单次操作的数量，默认200
+     * @param mode                           insert插入 ； update更新  ；  insertOrUpdate插入或更新
+     * @param isFastInsertOrUpdateIgnoreNull mode为新增此值为true代表fastInsert；mode为修改时此值为true代表updateIgnoreNull
+     * @param <T>
+     */
+    public <T> void asyncExecute(List<T> list, Integer batchSize, String mode, boolean isFastInsertOrUpdateIgnoreNull) {
+        if (Lang.isEmpty(list)) {
+            return;
+        }
+        if (batchSize == null || batchSize <= 0) {
+            batchSize = 200;
+        }
+        for (int i = 0; i < list.size(); i += batchSize) {
+            final int end = Math.min(i + batchSize, list.size());
+            List<T> batch = list.subList(i, end);
+            executorService().execute(() -> {
+                try {
+                    switch(mode) {
+                        case "insert" -> {
+                            if (isFastInsertOrUpdateIgnoreNull) {
+                                dao.fastInsert(batch);
+                            } else {
+                                dao.insert(batch);
+                            }
+                        }
+                        case "update" -> {
+                            if (isFastInsertOrUpdateIgnoreNull) {
+                                dao.updateIgnoreNull(batch);
+                            } else {
+                                dao.update(batch);
+                            }
+                        }
+                        case "insertOrUpdate" -> dao.insertOrUpdate(batch);
+                    }
+                } catch (Exception e) {
+                    // 日志记录异常信息
+                    System.err.println("Error occurred while inserting batch: " + e.getMessage());
+                    e.printStackTrace();
+                }
+            });
+        }
+    }
+
+}
@@ -0,0 +1,82 @@
+package io.v.nutz.base.utils;
+
+import io.v.nutz.base.config.ThreadPoolConfig;
+import org.nutz.ioc.loader.annotation.Inject;
+import org.nutz.ioc.loader.annotation.IocBean;
+
+import java.util.List;
+
+/**
+ * @version 1.0
+ * @Author zzr
+ * @name：ManyAddOrRenewUtil
+ * @Date 2024/7/24 16:16
+ * @注释
+ */
+@IocBean
+public class ManyAddOrRenewUtil {
+
+    @Inject
+    private ThreadPoolConfig threadPoolConfig;
+
+
+    /**
+     * 批量快速插入，异步执行
+     *
+     * @param list      泛型list,任意实体类集合
+     * @param batchSize 期望单次操作的数量，默认200
+     * @param <T>
+     */
+    public <T> void asyncExecuteFastInsert(List<T> list, Integer batchSize) {
+        threadPoolConfig.asyncExecute(list, batchSize, "insert", true);
+    }
+
+
+    /**
+     * 批量插入(非快速)，异步执行
+     *
+     * @param list      泛型list,任意实体类集合
+     * @param batchSize 期望单次操作的数量，默认200
+     * @param <T>
+     */
+    public <T> void asyncExecuteInsert(List<T> list, Integer batchSize) {
+        threadPoolConfig.asyncExecute(list, batchSize, "insert", true);
+    }
+
+
+    /**
+     * 批量修改（忽略空值），异步执行
+     *
+     * @param list      泛型list,任意实体类集合
+     * @param batchSize 期望单次操作的数量，默认200
+     * @param <T>
+     */
+    public <T> void asyncExecuteUpdateIgnoreNull(List<T> list, Integer batchSize) {
+        threadPoolConfig.asyncExecute(list, batchSize, "update", true);
+    }
+
+
+    /**
+     * 批量修改(不忽略空值)，异步执行
+     *
+     * @param list      泛型list,任意实体类集合
+     * @param batchSize 期望单次操作的数量，默认200
+     * @param <T>
+     */
+    public <T> void asyncExecuteUpdate(List<T> list, Integer batchSize) {
+        threadPoolConfig.asyncExecute(list, batchSize, "update", false);
+    }
+
+
+    /**
+     * 新增或修改，异步执行
+     *
+     * @param list      泛型list,任意实体类集合
+     * @param batchSize 期望单次操作的数量，默认200
+     * @param <T>
+     */
+    public <T> void asyncExecuteInsertOrUpdate(List<T> list, Integer batchSize) {
+        threadPoolConfig.asyncExecute(list, batchSize, "insertOrUpdate", false);
+    }
+
+}