线程池（二）—-ThreadPoolTaskExecutor

ThreadPoolTaskExecutor是一个spring的线程池技术，其实，它的实现方式完全是使用ThreadPoolExecutor进行实现。

应用举例：假设有三个耗时任务，一个返回结果，一个不返回结果，一个不返回结果且批量执行

1、ThreadPoolTaskExecutorConfig：

将ThreadPoolTaskExecutor做为一个bean，通过spring的注入，保证只会初始化一次。

package exceldemo.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
public class ThreadPoolTaskExecutorConfig {
private static int CORE_POOL_SIZE = 5;
private static int MAX_POOL_SIZE = 1000;
@Bean(name=”threadPoolTaskExecutor”)
public ThreadPoolTaskExecutor serviceJobTaskExecutor(){
ThreadPoolTaskExecutor poolTaskExecutor = new ThreadPoolTaskExecutor();
//线程池维护线程的最少数量
poolTaskExecutor.setCorePoolSize(CORE_POOL_SIZE);
//线程池维护线程的最大数量
poolTaskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
//线程池所使用的缓冲队列
poolTaskExecutor.setQueueCapacity(200);
//线程池维护线程所允许的空闲时间
poolTaskExecutor.setKeepAliveSeconds(30000);
poolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
System.out.println(poolTaskExecutor);
return poolTaskExecutor;
}
}

2、服务：

package exceldemo.service.impl;

import exceldemo.dto.User;
import exceldemo.service.UserService;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

@Service(“userService”)
public class UserServiceImpl implements UserService {
@Override
public List<User> getByIds(List<Integer> ids) {

List<User> users = new ArrayList<>();
for(Integer id : ids){
User user = new User();
user.setAge(id);
user.setUserName(“用户”+id);
//耗时操作
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
users.add(user);
}
return users;
}

@Override
public void addUserAction() {

//耗时操作
try {
Thread.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(“行为记录”);
}
}

3、需要提交的task：

package exceldemo.task;

import exceldemo.dto.User;
import exceldemo.service.UserService;

import java.util.List;
import java.util.concurrent.Callable;

public class UserTask implements Callable<List<User>> {

private List<Integer> userIds;

private UserService userService;

public UserTask(List<Integer> queryIds, UserService userService) {
this.userService = userService;
this.userIds = queryIds;
//System.out.println(“ids “+queryIds.size());
}

@Override
public List<User> call() throws Exception {
//System.out.println(“*******************”);
List<User> users = userService.getByIds(userIds);
return users;
}
}

异步任务：

package exceldemo.task;

import java.util.List;

public class DemoTask implements Runnable {

private List<Integer> ids ;

public DemoTask(List<Integer> ids){
this.ids = ids;
}

@Override
public void run() {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
//System.out.println(“run “+ids.size());
}
}

4、controller：

package exceldemo.rest;

import exceldemo.dto.User;
import exceldemo.service.UserService;
import exceldemo.task.UserTask;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

@RestController
@RequestMapping(“/user1”)
public class UserController {

@Autowired
private UserService userService;

@Autowired
private ThreadPoolTaskExecutor threadPoolTaskExecutor;

@RequestMapping(“/getAllSc”)
public List<User> getAllSc(){

List<Integer> ids = new ArrayList<>();
for(int i = 0;i<=500;i++){
ids.add(i);
}
//1、异步通知
threadPoolTaskExecutor.execute(new Runnable() {
@Override
public void run() {
userService.addUserAction();
}
});

//2、批量异步通知
List<Integer> childIds = new ArrayList<>();
for (int i = 0; i < ids.size(); i += 100) {
int startIndex = i;
int endIndex = startIndex + 100 > ids.size() ? ids.size() : startIndex + 100;
DemoTask task = new DemoTask(ids.subList(startIndex, endIndex));
threadPoolTaskExecutor.execute(task);
}

//3、异步获取所有用户
long startTime = new Date().getTime();
List<User> users = new ArrayList<>();
List<Future> futures = new ArrayList<>();

for (int i = 0; i < ids.size(); i += 100) {
int startIndex = i;
int endIndex = startIndex + 100 > ids.size() ? ids.size() : startIndex + 100;
UserTask task = new UserTask(ids.subList(startIndex, endIndex),userService);
Future<List<User>> future = threadPoolTaskExecutor.submit(task);
futures.add(future);
}
//取数据
try{
for(Future future : futures){
users.addAll((List<User>) future.get());
}
}catch (Exception e){

}

long endTime = new Date().getTime();
System.out.println(“耗时”+(endTime-startTime));
return users;
}
}