ruoyi-vue-plus-线程相关

# 线程相关

# 线程池

线程池配置

线程池配置文件配置

# 全局线程池相关配置 ThreadPoolProperties类
thread-pool:
  # 是否开启线程池
  enabled: true
  # 队列最大长度
  queueCapacity: 128
  # 线程池维护线程所允许的空闲时间
  keepAliveSeconds: 300

线程池配置类

点击展开

@Configuration
public class ThreadPoolConfig {

    /**
     * 核心线程数 = cpu 核心数 + 1
     */
    private final int core = Runtime.getRuntime().availableProcessors() + 1;

    @Autowired
    private ThreadPoolProperties threadPoolProperties;

    @Bean(name = "threadPoolTaskExecutor")
    @ConditionalOnProperty(prefix = "thread-pool", name = "enabled", havingValue = "true")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(core);
        executor.setMaxPoolSize(core * 2);
        executor.setQueueCapacity(threadPoolProperties.getQueueCapacity());
        executor.setKeepAliveSeconds(threadPoolProperties.getKeepAliveSeconds());
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }

    /**
     * 执行周期性或定时任务
     */
    @Bean(name = "scheduledExecutorService")
    protected ScheduledExecutorService scheduledExecutorService() {
        return new ScheduledThreadPoolExecutor(core,
            new BasicThreadFactory.Builder().namingPattern("schedule-pool-%d").daemon(true).build(),
            new ThreadPoolExecutor.CallerRunsPolicy()) {
            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                super.afterExecute(r, t);
                Threads.printException(r, t);
            }
        };
    }
}

示例

点击展开

@Slf4j
@SaIgnore
@RestController
@RequestMapping("/thread")
@RequiredArgsConstructor
public class ThreadPoolController {
	
    // 线程池
    private final ThreadPoolTaskExecutor threadPoolTaskExecutor;

    // 定时线程池
    private final ScheduledExecutorService scheduledExecutorService;

    // 线程池应用测试
    @GetMapping("/test1")
    public String test() {
        log.info("线程池应用测试");
        // 命令 执行任务
        threadPoolTaskExecutor.execute(this::task);
        // 命令 执行任务 含返回值
        threadPoolTaskExecutor.submit(this::task);
        return "test1";
    }

    // 定时线程池测试
    @GetMapping("/test2")
    public String test2() {
        log.info("定时线程池测试");
        scheduledExecutorService.schedule(this::task, 10, TimeUnit.SECONDS);
        return "test2";
    }

    // 调用有参方法时
    @GetMapping("/test3")
    public String test3() {
        log.info("调用有参方法时");
        Future<String> submit = threadPoolTaskExecutor.submit(this::task2);
        try {
            String msg = submit.get();
            log.info(msg);
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
        return "test3";
    }

    // get()方法 线程测试
    @GetMapping("/test4")
    public String test4() {
        for (int i = 0; i < 20; i++) {
            Future<String> submit = threadPoolTaskExecutor.submit(this::task3);
            try {
                String msg = submit.get();
                log.info("{} => {}", i, msg);
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
        return "test4";
    }

    @GetMapping("/test5")
    public String test5() {
        List<Future<String>> res = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            Future<String> submit = threadPoolTaskExecutor.submit(this::task3);
            res.add(submit);
        }

        try {
            int i = 0;
            for (Future<String> re : res) {
                String msg = re.get();
                log.info("{} => {}", i++, msg);
            }
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
        return "test5";
    }

    // 任务方法
    public void task() {
        log.info("任务执行");
    }

    // 含返回值任务方法
    public String task2() {
        log.info("任务执行");
        return "task2";
    }

    // 真实业务场景(执行时间)
    public String task3() {
        log.info("任务执行");
        ThreadUtil.sleep(1000);
        return "task3";
    }

}

# Async注解

在Spring中应用异步功能 , 只需在配置类中启用 @EnableAsync注解即可

大致应用

配置类配置应用
在注册有Bean的类上写上方法
在方法上加上 @Async注解
测试即可

配置

配置异步方法需要继承重写 AsyncConfigurerSupport类 , 主要重写

getAsyncExecutor() : 异步执行获取的线程池
getAsyncUncaughtExceptionHandler() : 线程池异常处理

提示

@EnableAsync注解参数proxyTargetClass标识采用CGLIB代理 , 异步调用时必须要通过 Spring Ioc容器拿取Bean对象(Bean管理) . 意味着异步的方法要写在Bean对象内

PS : 默认采用的是 JDK自带代理 , 性能和通用性远不及于 CGLIB代理

点击展开

// 启用异步代理
@EnableAsync(proxyTargetClass = true)
@Configuration
public class AsyncConfig extends AsyncConfigurerSupport {

    /**
     * 以下方式根据名字注入Bean 
     * 采用的是API提供的类 , 因此不能用对象类型获取 
     */
    @Autowired
    @Qualifier("scheduledExecutorService")
    private ScheduledExecutorService scheduledExecutorService;

    /**
     * 自定义 @Async 注解使用系统线程池
     */
    @Override
    public Executor getAsyncExecutor() {
        return scheduledExecutorService;
    }

    /**
     * 异步执行异常处理
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (throwable, method, objects) -> {
            throwable.printStackTrace();
            StringBuilder sb = new StringBuilder();
            sb.append("Exception message - ").append(throwable.getMessage())
                .append(", Method name - ").append(method.getName());
            if (ArrayUtil.isNotEmpty(objects)) {
                sb.append(", Parameter value - ").append(Arrays.toString(objects));
            }
            throw new ServiceException(sb.toString());
        };
    }

}

示例代码

Controller层入口

点击展开

@Slf4j
@SaIgnore
@AllArgsConstructor
@RestController
@RequestMapping("/async")
public class AsyncTestController {

    private final AsyncServiceImpl asyncService;

    /**
     * 异步测试
     * 结论: 必须通过注入Bean对象调用当中的异步方法 , 否则异步不会生效
     */
    @RequestMapping("/test1")
    public String test1() {
        log.info("test1");
        asyncService.taskApply();
        taskApply();
        return "test1";
    }

    @Async
    public void taskApply() {
        log.info("内部方法异步测试");
    }

    /**
     * 延迟测试
     * 结论: 异步线程的任务,不会影响主线程的执行时长
     */
    @RequestMapping("/test2")
    public String test2() {
        log.info("test2");
        asyncService.taskDelay();
        return "test2";
    }

    /**
     * 异常测试
     */

    /**
     * 异步异常测试 (主线程)
     * 结论: 主线程异常不会影响 异步线程的执行
     */
    @RequestMapping("/test3")
    public void test3() {
        log.info("test3");
        asyncService.taskApply();
        throw new RuntimeException("异常");
    }

    /**
     * 异步异常测试 (异步线程异常)
     * 结论: 异步线程异常不会影响 主线程的执行
     */
    @RequestMapping("/test4")
    public void test4() {
        log.info("test4");
        asyncService.taskException();
    }

    /**
     * 事务测试
     */

    /**
     * 事务操作测试 (主线程)
     * 结论: 主线程事务不会对异步线程进行回滚 (主线程异常如此)
     */
    @RequestMapping("/test5")
    @Transactional
    public void test5() {
        log.info("test5");
        asyncService.taskTransaction();
        throw new RuntimeException("异常");
    }

    /**
     * 事务测试 (异步线程异常)
     * 结论: 主线程和异步线程的事务是隔离互补干扰
     */
    @RequestMapping("/test6")
    @Transactional
    public void test6() {
        log.info("test6");
        asyncService.taskTransactionException();
    }

}

Service层业务

点击展开

@Slf4j
@Service
@AllArgsConstructor
public class AsyncServiceImpl {
	// 框架自带
    private final ITestDemoService demoService;

    // 异步任务
    @Async
    public void taskApply() {
        log.info("taskApply");
    }

    // 延迟执行 (模拟业务场景)
    @Async
    public void taskDelay() {
        ThreadUtil.sleep(2000);
        log.info("taskDelay");
    }

    // 异常异步
    @Async
    public void taskException() {
        log.info("taskException");
        throw new RuntimeException("异常");
    }

    // 事务测试 (正常)
    @Async
    public void taskTransaction() {
        log.info("taskTransaction");
        demoService.insertByBo(new TestDemoBo());
    }

    // 事务测试 (异常)
    @Async
    @Transactional
    public void taskTransactionException() {
        log.info("taskTransactionException");
        demoService.insertByBo(new TestDemoBo());
        throw new RuntimeException("异常");
    }

}

#ruoyi-vue-plus #oss

← ruoyi-vue-plus-日志功能 ruoyi-vue-plus-OSS功能→