溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》

在ThreadPoolExecutor環(huán)境中如何實(shí)現(xiàn)創(chuàng)建一個(gè)線程池

發(fā)布時(shí)間:2020-11-11 15:23:22 來(lái)源:億速云 閱讀:203 作者:Leah 欄目:開發(fā)技術(shù)

這篇文章給大家介紹在ThreadPoolExecutor環(huán)境中如何實(shí)現(xiàn)創(chuàng)建一個(gè)線程池,內(nèi)容非常詳細(xì),感興趣的小伙伴們可以參考借鑒,希望對(duì)大家能有所幫助。

1、用到的guava坐標(biāo)

<dependency>
      <groupId>com.google.guava</groupId>
      <artifactId>guava</artifactId>
      <version>18.0</version>
    </dependency>

2、創(chuàng)建一個(gè)枚舉保證線程池是單例

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、創(chuàng)建一個(gè)方法類

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、創(chuàng)建一個(gè)測(cè)試類

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、執(zhí)行結(jié)果如下

在ThreadPoolExecutor環(huán)境中如何實(shí)現(xiàn)創(chuàng)建一個(gè)線程池

次工具類的好處在于能夠像使用普通 service一樣使用線程池完成并行操作,當(dāng)然不要忘記將 ExecutorContext 置于能被sping掃描到的地方,

否則不能直接使用@Autowired 依賴注入

關(guān)于在ThreadPoolExecutor環(huán)境中如何實(shí)現(xiàn)創(chuàng)建一個(gè)線程池就分享到這里了,希望以上內(nèi)容可以對(duì)大家有一定的幫助,可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò),可以把它分享出去讓更多的人看到。

向AI問(wèn)一下細(xì)節(jié)

免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI