在ThreadPoolExecutor环境中如何实现创建一个线程池
创始人
2024-11-19 00:01:40
0次

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(AVAILABLEPROCESSORS * 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 ExecutorContext {

  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.println(employee);
    }
    System.out.println(System.currentTimeMillis() - t0);
  }

}

5、执行结果如下

在ThreadPoolExecutor环境中如何实现创建一个线程池

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

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

相关内容

热门资讯

空调连续工作多久需要停机休息 空调连续工作的时间取决于多个因素,包括空调的型号、功率、使用环境和维护情况等。一般来说,空调可以连续...
空调哪个符号是除湿的 空调除湿功能通常用一个水滴形状的符号表示。不同品牌和型号的空调遥控器可能会有所差异,但一般来说,除湿...
格力空调多长时间加一次氟利昂 1、一般情况下,空调对氟利昂的使用量是非常小的,所以在空调使用了四年左右的时候再考虑加也没问题。2、...
空调用多长时间需要加氟一次 一般情况下是五年。当出现以下这几种情况时,空调才需要加氟。1、空调的室内机和室外机各有两个接口,当接...
格调生活·C标思享荟:卡萨帝在... 9月12日的郑州龙湖金融岛,夜风轻拂,灯光璀璨。“C标时刻共传奇——卡萨帝思享荟暨20周年品牌盛典郑...
品牌价值破千亿、高端份额持续领... 9月12日,“C标时刻共传奇——卡萨帝思享荟暨20周年品牌盛典郑州站”在郑州龙湖金融岛举办。2026...
空调主板坏了修800还是换新的 空调主板坏了是修还是换新,需要考虑多个因素,以下是一些建议:1. 维修成本:如果维修费用较低,比如 ...
空调开多久需要休息多长时间 问题:空调开多久需要休息多长时间回答:1、在使用空调的过程中,一般空调开7~8个小时左右需要休息2~...
中央空调多长时间清洗一次 1、一般空调开机前清洗一次即可,空调开机中间时段再清洗一次,空调关机时清洗一次,这样比较合理。2、空...
中央空调氟利昂多久加一次 看空调制热和制冷程度,很多人的观点认为氟利昂必须一年放一次,其实这个观点是错误的,理解上面有点偏差,...