springboot2.3整合mybatis-plus，实现xxljob执行器基于线程池的异步多线程任务

近期，由于工作的需要，需要开发一款基于xxljob调度器的执行器任务。该执行器主要的业务是负责从流水表中定期采集数据后，按照不同的维度将流水数据分类统计并处理后存入oracle数据库。由于所有的数据处理维度并没有相互的依赖关联关系，这里采用了基于线程池的异步多线程调度执行。

1、开发环境

虚拟机vmware16
centos7
jdk8
springboot2.3
xxljob2.3.0
mybatis-plus
1
2
3
4
5
6

2、具体实施

我们这次主要采用了springboot2.3线程池非定时任务类ThreadPoolTaskExecutor。该类的主要优点是将多多线程和线程池同时实现，不用再像以往使用多线程的时候创建Thread类或实现Runnable接口，使用线程池创建Executors。而我们现在只需要在启动文件添加@EnableAsync就可以使用多线程，在线程业务类文件或者线程业务方法上添加@Async就可以使用ThreadPoolTaskExecutor提供的线程池，同时该线程池都是异步调用，非常实用。

2.1 创建线程池

创建线程池配置文件AsyncConfig。内容如下：

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    @Bean
    public ThreadPoolTaskExecutor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        // 核心线程数：线程池创建的时候初始化的线程数
        taskExecutor.setCorePoolSize(10);
        // 最大线程数：线程池最大的线程数，只有缓冲队列满了之后才会申请超过核心线程数的线程
        taskExecutor.setMaxPoolSize(100);
        // 缓冲队列：用来缓冲执行任务的队列
        taskExecutor.setQueueCapacity(50);
        // 线程池关闭：等待所有任务都完成再关闭
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        // 等待时间：等待5秒后强制停止
        taskExecutor.setAwaitTerminationSeconds(5);
        // 允许空闲时间：超过核心线程之外的线程到达60秒后会被销毁
        taskExecutor.setKeepAliveSeconds(60);
        // 线程名称前缀
        taskExecutor.setThreadNamePrefix("taskExecutor--");

        taskExecutor.initialize();
        return taskExecutor;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

2.2 开启多线程任务

开启多线程任务有俩种方法，
第一种可以在启动类上添加@EnableAsync注解

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    @Override
    @Bean
    public ThreadPoolTaskExecutor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
		……
        taskExecutor.initialize();
        return taskExecutor;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13

第二种可以在线程池配置文件上添加@EnableAsync

@SpringBootApplication
@EnableAsync
public class ProcessFlowExecutorApplication {

	public static void main(String[] args) {
        SpringApplication.run(ProcessFlowExecutorApplication.class, args);
	}
}
1
2
3
4
5
6
7
8

我本次就添加到了线程池配置文件上。

2.3 service层使用@Async注解实现异步线程池任务

service层接口代码如下：

public interface ChannelService extends IService<Channel> {
    public void saveBatchChannel(int i);
}
1
2
3

service层实现类代码如下：

@Service
public class ChannelServiceImpl extends ServiceImpl<ChannelMapper, Channel> implements ChannelService {
    @Async
    public void saveBatchChannel(int i) {
        System.out.println(" 执行异步任务：" + i);
    }
}
1
2
3
4
5
6
7

2.4 controller层调用业务层

@RestController
public class IndexController {
    @Autowired
    private ChannelService channelService;

    @RequestMapping("/userlist")
    public void getUserList() {
		for (int i = 0; i < 10; i++) {
            channelService.saveBatchChannel(i);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

2.5 打印的结果为

3 结束语
好了，今天的文章就写到这里吧，如果您感觉该文章实用，点赞收藏一下吧，您的鼓励是我最大的动力。如果觉得哪里有问题，可以评论区留言，欢迎批评指正。相互学习。