如何基于ThreadPoolExecutor创建线程池并操作

如何基于ThreadPoolExecutor创建线程池并操作

日常工作中很多地方很多效率极低的操作，往往可以改串行为并行，执行效率往往提高数倍，废话不多说先上代码

1、用到的guava坐标

com.google.guava

guava

18.0

2、创建一个枚举保证线程池是单例

package com.hao.service;

import java.util.concurrent.LinkedBlockingQueue;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

public enum ExecutorManager {

INSTANCE;

private ExecutorManager() {

}

private static int AVAILABLEPROCESSORS = Runtime.getRuntime().availableProcessors();

public static final ThreadPoolExecutor threadPoolExecutor =

new ThreadPoolExecutor(AVAILABLEPRhttp://OCESSORS * 50, AVAILABLEPROCESSORS * 80, 0L, TimeUnit.MILLISECONDS,

new LinkedBlockingQueue(AVAILABLEPROCESSORS * 2000),

new ThreadFactoryBuilder().setNameFormat("ExecutorManager-pool-Thread-%d").build());

}

3、创建一个方法类

package com.hao.service;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.LinkedList;

import java.util.List;

import java.util.concurrent.Callable;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.springframework.stereotype.Service;

import com.google.common.base.Preconditions;

@Service

public class ExecutorContexhttp://t {

public ExecutorService executorService;

private int DEFAULT_WAIT_SECONDS = 2;

@PostConstruct

public void init() {

executorService = ExecutorManager.threadPoolExecutor;

}

public List waitAllFutures(List> calls, int milliseconds) throws Exception {

Preconditions.checkArgument(null != calls && !calls.isEmpty(), "callable empty.");

LatchedCallables latchAndCallables = wrapCallables(calls);

List> futurres = new LinkedList<>();

for (CountdownedCallable callable : latchAndCallables.wrappedCallables) {

if (null != callable) {

futurres.add(executorService.submit(callable));

}

}

List rets = new ArrayList<>();

if (latchAndCallables.latch.await(milliseconds, TimeUnit.MILLISECONDS)) {

for (CountdownedCallable call : latchAndCallables.wrappedCallables) {

rets.add(call.getResult());

}

} else {

for (Future future : futurres) {

if (!future.isDone()) {

future.cancel(true);

}

}

}

return rets;

}

public List waitAllCallables(List> calls, int seconds) throws Exception {

Preconditions.checkArgument(null != calls && !calls.isEmpty(), "callable empty.");

LatchedCallables latchAndCallables = wrapCallables(calls);

for (CountdownedCallable callable : latchAndCallables.wrappedCallables) {

executorService.submit(callable);

}

List rets = new ArrayList<>();

if (latchAndCallables.latch.await(seconds, TimeUnit.SECONDS)) {

for (CountdownedCallable call : latchAndCallables.wrappedCallables) {

rets.add(call.getResult());

}

}

return rets;

}

public List waitAllCallables(@SuppressWarnings("unchecked") Callable... calls) throws Exception {

Preconditions.checkNotNull(calls, "callable empty.");

return waitAllCallables(Arrays.asList(calls), DEFAULT_WAIT_SECONDS);

}

private static LatchedCallables wrapCallables(List> callables) {

CountDownLatch latch = new CountDownLatch(callables.size());

List> wrapped = new ArrayList<>(callables.size());

for (Callable callable : callables) {

wrapped.add(new CountdownedCallable<>(callable, latch));

}

LatchedCallables returnVal = new LatchedCallables<>();

returnVal.latch = latch;

returnVal.wrappedCallables = wrapped;

return returnVal;

}

public static class LatchedCallables {

public CountDownLatch latch;

public List> wrappedCallables;

}

public static class CountdownedCallable implements Callable {

private final Callable wrapped;

private final CountDownLatch latch;

private T result;

public CountdownedCallable(Callable wrapped, CountDownLatch latch) {

this.wrapped = wrapped;

this.latch = latch;

}

@Override

public T call() throws Exception {

try {

result = wrapped.call();

return result;

} finally {

latch.countDown();

}

}

public T getResult() {

return result;

}

}

}

4、创建一个测试类

package com.hao;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.Callable;

import org.junit.Test;

import org.springframework.beans.factory.annotation.Autowired;

import com.hao.bean.Employee;

import com.hao.service.EmployeeService;

import com.hao.service.ExecutorContext;

public class ExecutorTest extends BaseTest {

@Autowired

ExecutorContext executorContext;

@Autowired

EmployeeService employeeService;

@Test

public void test01() {

long t0 = System.currentTimeMillis();

List employees = new ArrayList();

try {

List> calls = new ArrayList>();

Callable able1 = new Callable() {

@Override

public Integer call() throws Exception {

Thread.sleep(5000);

Employee employee = employeeService.getById(1L);

employees.add(employee);

return 1;

}

};

calls.add(able1);

Callable able2 = new Callable() {

@Override

public Integer call() throws Exception {

Thread.sleep(5000);

Employee employee = employeeService.getById(2L);

employees.add(employee);

return 2;

}

};

calls.add(able2);

Callable able3 = new Callable() {

@Override

public Integer call() throws Exception {

Thread.sleep(5000);

Employee employee = employeeService.getById(3L);

employees.add(employee);

return 3;

}

};

calls.add(able3);

executorContext.waitAllCallables(calls, 5000);

} catch (Exception e) {

e.printStackTrace();

}

for (Employee employee : employees) {

System.out.prrYTBlyyYintln(employee);

}

System.out.println(System.currentTimeMillis() - t0);

}

}

5、执行结果如下

次工具类的好处在于能够像使用普通 service一样使用线程池完成并行操作，当然不要忘记将 ExecutorContext 置于能被sping扫描到的地方，

否则不能直接使用@Autowired 依赖注入

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。