如何基于ThreadPoolExecutor创建线程池并操作
更新时间:2020年11月10日 09:25:26 作者:牛鼻子老赵
这篇文章主要介绍了如何基于ThreadPoolExecutor创建线程池并操作,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
日常工作中很多地方很多效率极低的操作,往往可以改串行为并行,执行效率往往提高数倍,废话不多说先上代码
1、用到的guava坐标
<dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>18.0</version> </dependency>
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(AVAILABLEPROCESSORS * 50, AVAILABLEPROCESSORS * 80, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(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 ExecutorContext { public ExecutorService executorService; private int DEFAULT_WAIT_SECONDS = 2; @PostConstruct public void init() { executorService = ExecutorManager.threadPoolExecutor; } public <T> List<T> waitAllFutures(List<Callable<T>> calls, int milliseconds) throws Exception { Preconditions.checkArgument(null != calls && !calls.isEmpty(), "callable empty."); LatchedCallables<T> latchAndCallables = wrapCallables(calls); List<Future<T>> futurres = new LinkedList<>(); for (CountdownedCallable<T> callable : latchAndCallables.wrappedCallables) { if (null != callable) { futurres.add(executorService.submit(callable)); } } List<T> rets = new ArrayList<>(); if (latchAndCallables.latch.await(milliseconds, TimeUnit.MILLISECONDS)) { for (CountdownedCallable<T> call : latchAndCallables.wrappedCallables) { rets.add(call.getResult()); } } else { for (Future<T> future : futurres) { if (!future.isDone()) { future.cancel(true); } } } return rets; } public <T> List<T> waitAllCallables(List<Callable<T>> calls, int seconds) throws Exception { Preconditions.checkArgument(null != calls && !calls.isEmpty(), "callable empty."); LatchedCallables<T> latchAndCallables = wrapCallables(calls); for (CountdownedCallable<T> callable : latchAndCallables.wrappedCallables) { executorService.submit(callable); } List<T> rets = new ArrayList<>(); if (latchAndCallables.latch.await(seconds, TimeUnit.SECONDS)) { for (CountdownedCallable<T> call : latchAndCallables.wrappedCallables) { rets.add(call.getResult()); } } return rets; } public <T> List<T> waitAllCallables(@SuppressWarnings("unchecked") Callable<T>... calls) throws Exception { Preconditions.checkNotNull(calls, "callable empty."); return waitAllCallables(Arrays.asList(calls), DEFAULT_WAIT_SECONDS); } private static <T> LatchedCallables<T> wrapCallables(List<Callable<T>> callables) { CountDownLatch latch = new CountDownLatch(callables.size()); List<CountdownedCallable<T>> wrapped = new ArrayList<>(callables.size()); for (Callable<T> callable : callables) { wrapped.add(new CountdownedCallable<>(callable, latch)); } LatchedCallables<T> returnVal = new LatchedCallables<>(); returnVal.latch = latch; returnVal.wrappedCallables = wrapped; return returnVal; } public static class LatchedCallables<T> { public CountDownLatch latch; public List<CountdownedCallable<T>> wrappedCallables; } public static class CountdownedCallable<T> implements Callable<T> { private final Callable<T> wrapped; private final CountDownLatch latch; private T result; public CountdownedCallable(Callable<T> 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<Employee> employees = new ArrayList<Employee>(); try { List<Callable<Integer>> calls = new ArrayList<Callable<Integer>>(); Callable<Integer> able1 = new Callable<Integer>() { @Override public Integer call() throws Exception { Thread.sleep(5000); Employee employee = employeeService.getById(1L); employees.add(employee); return 1; } }; calls.add(able1); Callable<Integer> able2 = new Callable<Integer>() { @Override public Integer call() throws Exception { Thread.sleep(5000); Employee employee = employeeService.getById(2L); employees.add(employee); return 2; } }; calls.add(able2); Callable<Integer> able3 = new Callable<Integer>() { @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.println(employee); } System.out.println(System.currentTimeMillis() - t0); } }
5、执行结果如下
次工具类的好处在于能够像使用普通 service一样使用线程池完成并行操作,当然不要忘记将 ExecutorContext 置于能被sping扫描到的地方,
否则不能直接使用@Autowired 依赖注入
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。
相关文章
rabbitmq使用springboot实现direct模式(最新推荐)
这篇文章主要介绍了rabbitmq使用springboot实现direct模式,本文通过示例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下2023-07-07springboot2.x默认使用的代理是cglib代理操作
这篇文章主要介绍了springboot2.x默认使用的代理是cglib代理操作,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教2021-08-08SpringBoot+MyBatisPlus+MySQL8实现树形结构查询
这篇文章主要为大家详细介绍了SpringBoot+MyBatisPlus+MySQL8实现树形结构查询,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下2021-06-06
最新评论