应用调优常用技巧-線程池

应用调优常用技巧 - 线程池
- 線程池的好處
- 核心API-操作類
- 核心API-監控類
- 2-2 线程池BlockingQueue详解、选择与调优
- - 調優技巧
- 2-3 线程池ScheduledThreadPoolExecutor详解
- 2-4 线程池ForkJoinPool详解
- 2-5 线程池Executors讲解
- 2-6 线程池调优实战
- - 1.线程数调优
  - 2. BlockingQueue調優
- 2-7 線程池總結

应用调优常用技巧 - 线程池

線程池的應用：

Eureka Client里面，每隔30秒发送个心跳，其实就是个定时任务型的线程池；
Tomcat处理请求时，也有线程池；
Spring Task也是用的线程池实现的定时任务
你用@Async实现异步的时候，底层也是个线程池，当然视频里讲了@Async以及手动用线程池实现异步两种方式。
比如你要处理一批数据，比如100万，就可以弄个线程池，每个线程处理一部分数据，并发进行，从而提升处理速度。

線程池的好處

重用已存在的縣城
控制并發
功能強大

核心API-操作類

用的最多的是：execute和submit

execute()：提交任務，交給線程池執行
submit()：提交任務，能夠返回執行結果
shutdown()：關閉線程池，等待任務都執行
shutdownNow()：關閉線程池，不等待任務執行完（很少使用，因為過於暴力了）

核心API-監控類

getTaskCount()：返回線程已執行和未執行的任務總數
getComplatedTaskCount(）：已完成的任務總數
getPollSize()：線程池當前線程的數量
getActiveCount()：線程池中正在執行任務的線程數量

2-2 线程池BlockingQueue详解、选择与调优

API中啊add的相關代碼

public class LinkedBlockingQueueTest {public static void main(String[] args) {LinkedBlockingQueue<Object> queue =new LinkedBlockingQueue<>(1);queue.add("abc");boolean def = queue.offer("def");System.out.println(def);queue.add("g");}
}
無界隊列消耗很大，因為他沒有界限。

調優技巧

合理设置corePoolSize、maximumPoolSize、 workQueue的容量。
下面代碼中corePoolSize=5，maximumPoolSize=10（不是10L），workQueue=100。這樣多餘的任務會在隊列裡面排隊，線程的個數會保持在一個比較合理的範圍。這樣線程的開銷就比較小了。
假如經常發生阻塞，說明隊列經常滿，可以考慮重新設置線程數量executor.setMaximumPoolSize(50);
假如隊列的容量設置的比較小，通常需要把線程池的容量設置的大一點，這樣cpu的使用率會高一點。
如果線程池的容量設置的非常大，而隊列比較小，二任務提交的非常多的時候，就會創建很多個線程去執行，此時會降低系統的吞吐量，因為線程切換的開銷變大了。

public static void main(String[] args) throws ExecutionException, InterruptedException {ThreadPoolExecutor executor =new ThreadPoolExecutor(5,10,// 默认情况下指的是非核心线程的空闲时间// 如果allowCoreThreadTimeOut=true：核心线程/非核心线程允许的空闲时间10L,TimeUnit.SECONDS,new LinkedBlockingQueue<>(100),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());executor.allowCoreThreadTimeOut(true);// 核心线程 -> 正式员工// 非核心线程 -> 临时工executor.execute(new Runnable() {@Overridepublic void run() {System.out.println("线程池测试");}});executor.execute(new Runnable() {@Overridepublic void run() {System.out.println("线程池测试2");}});Future<String> future = executor.submit(new Callable<String>() {@Overridepublic String call() throws Exception {return "测试submit";}});String s = future.get();System.out.println(s);}

2-3 线程池ScheduledThreadPoolExecutor详解

实际项目中应用比较多的是下面这两个方法

scheduleAtFixedRate
ScheduleAtFixedRate

public class ScheduledThreadPoolExecutorTest {public static void main(String[] args) throws ExecutionException, InterruptedException {ScheduledThreadPoolExecutor executor= new ScheduledThreadPoolExecutor(10,Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
//    // 延时3秒之后再去执行任务
//    executor.schedule(
//      new Runnable() {//        @Override
//        public void run() {//          System.out.println("aaa");
//        }
//      },
//      3,
//      TimeUnit.SECONDS
//    );
//
//    // 延时4秒之后再去执行任务，可以返回执行结果
//    ScheduledFuture<String> future = executor.schedule(new Callable<String>() {//      @Override
//      public String call() throws Exception {//        return "bbb";
//      }
//    }, 4, TimeUnit.SECONDS);
//    String s = future.get();
//    System.out.println(s);executor.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {System.out.println("scheduleAtFixedRate" + new Date());try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}}},// 第一次执行任务时，延时多久0,// 每个多久执行这个任务3, TimeUnit.SECONDS);executor.scheduleWithFixedDelay(new Runnable() {@Overridepublic void run() {System.out.println("scheduleWithFixedDelay" + new Date());try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}}},// 第一次执行任务时，延时多久0,// 每次执行完任务之后，延迟多久再次执行这个任务3, TimeUnit.SECONDS);}
}

Timer与ScheduledThreadPoolExecutor对比，博主写的很好
对比结果：Timer是单线程的，在实际项目中要尽量使用ScheduledThreadPoolExecutor，慎用Timer

2-4 线程池ForkJoinPool详解

ForkJoinPoll在實際項目中使用並不多，但是可以幫助我們閱讀JDK源碼。（鄙人並不懂）

public class ForkJoinPoolTest {// ForkJoinPool实现1-100的求和public static void main(String[] args) throws ExecutionException, InterruptedException {ForkJoinPool pool = new ForkJoinPool();ForkJoinTask<Integer> task = pool.submit(new MyTask(1, 100));Integer sum = task.get();System.out.println(sum);}
}class MyTask extends RecursiveTask<Integer> {public MyTask(int start, int end) {this.start = start;this.end = end;}// 当前任务计算的起始private int start;// 当前任务计算的结束private int end;// 阈值，如果end-start在阈值以内，那么就不用再去细分任务public static final int threshold = 2;@Overrideprotected Integer compute() {int sum = 0;boolean needFork = (end - start) > threshold;if (needFork) {int middle = (start + end) / 2;MyTask leftTask = new MyTask(start, middle);MyTask rightTask = new MyTask(middle + 1, end);// 执行子任务leftTask.fork();rightTask.fork();// 子任务执行完成之后的结果Integer leftResult = leftTask.join();Integer rightResult = rightTask.join();sum = leftResult + rightResult;} else {for (int i = start; i <= end; i++) {sum += i;}}return sum;}
}

2-5 线程池Executors讲解

創建線程池的工廠。

public class ExecutorsTest {public static void main(String[] args) {// 可以選擇上圖中不同的線程池ExecutorService pool = Executors.newCachedThreadPool();pool.execute(new Runnable() {@Overridepublic void run() {System.out.println("aaa");}});}
}

2-6 线程池调优实战

線程池調優

线程数调优
BlockingQueue调优

1.线程数调优

任务可以分為以下幾幾種

cpu密集型任務這個任務大部分時間都在進行cpu計算，如排序
調優經驗公式：N+1

查看本機器cpu數量：

public class ThreadPoolCoreTest {public static void main(String[] args) {int i = Runtime.getRuntime().availableProcessors();System.out.println(i);}
}

輸出結果是8

8
Process finished with exit code 0

那麼就可以根據公式：N+1，就可以將線程的數量設置為 9
為什麼設置為9呢，8不好嗎？如果設置為8個線程，某一個線程突然出現暫停或者中斷的話，那麼8個cpu就會有一個出現空閒，多出來的那個線程，就會充分利用cpu的空閒時間。就像踢足球有一個替補隊員。

IO密集型任務大部分時間在和IO交互，如操作數據庫
調優經驗公式：2N。
線程處理IO，是不佔用cpu的，所以處理IO的時候這個線程的CPU資源，就可以交給其他線程去使用，因此就可以多設置一些線程，業界比較認可的經驗值是2N，對於我的機器線程數就可以配製成16
混合型任務實際項目中，混合型任務比較多
調優公式：N * U *（1+WT/ST）

N：非常好獲取
U：是我們的期待，也非常好設置
WT和ST怎麼獲取呢？技巧：我們可以運行一個項目，然後打開終端，輸入jvisualvm命令，就可以打開java VisualVM了

然後選擇自己剛運行的項目，雙擊打開，然後Profile–>cpu，性能分析會運行一段時間。這樣ST和WT就知道了

2. BlockingQueue調優

得做估算，做下面兩個估算。這樣比較麻煩，可以使用懶人工具

單個任務佔用內存
線程池計劃佔用內存

懶人工具，裡面有一個工具類，可以幫助我們調優線程池

https://www.javacodegeeks.com/2012/03/threading-stories-about-robust-thread.html

線程池調優工具類：


import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;/*** 線程池調優工具類* A class that calculates the optimal thread pool boundaries. It takes the desired target utilization and the desired* work queue memory consumption as input and retuns thread count and work queue capacity.* * @author Niklas Schlimm* */
public abstract class PoolSizeCalculator {/*** The sample queue size to calculate the size of a single {@link Runnable} element.*/private final int SAMPLE_QUEUE_SIZE = 1000;/*** Accuracy of test run. It must finish within 20ms of the testTime otherwise we retry the test. This could be* configurable.*/private final int EPSYLON = 20;/*** Control variable for the CPU time investigation.*/private volatile boolean expired;/*** Time (millis) of the test run in the CPU time calculation.*/private final long testtime = 3000;/*** Calculates the boundaries of a thread pool for a given {@link Runnable}.* * @param targetUtilization*            the desired utilization of the CPUs (0 <= targetUtilization <= 1)* @param targetQueueSizeBytes*            the desired maximum work queue size of the thread pool (bytes)*/protected void calculateBoundaries(BigDecimal targetUtilization, BigDecimal targetQueueSizeBytes) {calculateOptimalCapacity(targetQueueSizeBytes);Runnable task = creatTask();start(task);start(task); // warm up phaselong cputime = getCurrentThreadCPUTime();start(task); // test intervallcputime = getCurrentThreadCPUTime() - cputime;long waittime = (testtime * 1000000) - cputime;calculateOptimalThreadCount(cputime, waittime, targetUtilization);}private void calculateOptimalCapacity(BigDecimal targetQueueSizeBytes) {long mem = calculateMemoryUsage();BigDecimal queueCapacity = targetQueueSizeBytes.divide(new BigDecimal(mem), RoundingMode.HALF_UP);System.out.println("Target queue memory usage (bytes): " + targetQueueSizeBytes);System.out.println("createTask() produced " + creatTask().getClass().getName() + " which took " + mem+ " bytes in a queue");System.out.println("Formula: " + targetQueueSizeBytes + " / " + mem);System.out.println("* Recommended queue capacity (bytes): " + queueCapacity);}/*** Brian Goetz' optimal thread count formula, see 'Java Concurrency in Practice' (chapter 8.2)* * @param cpu*            cpu time consumed by considered task* @param wait*            wait time of considered task* @param targetUtilization*            target utilization of the system*/private void calculateOptimalThreadCount(long cpu, long wait, BigDecimal targetUtilization) {BigDecimal waitTime = new BigDecimal(wait);BigDecimal computeTime = new BigDecimal(cpu);BigDecimal numberOfCPU = new BigDecimal(Runtime.getRuntime().availableProcessors());BigDecimal optimalthreadcount = numberOfCPU.multiply(targetUtilization).multiply(new BigDecimal(1).add(waitTime.divide(computeTime, RoundingMode.HALF_UP)));System.out.println("Number of CPU: " + numberOfCPU);System.out.println("Target utilization: " + targetUtilization);System.out.println("Elapsed time (nanos): " + (testtime * 1000000));System.out.println("Compute time (nanos): " + cpu);System.out.println("Wait time (nanos): " + wait);System.out.println("Formula: " + numberOfCPU + " * " + targetUtilization + " * (1 + " + waitTime + " / "+ computeTime + ")");System.out.println("* Optimal thread count: " + optimalthreadcount);}/*** Runs the {@link Runnable} over a period defined in {@link #testtime}. Based on Heinz Kabbutz' ideas* (http://www.javaspecialists.eu/archive/Issue124.html).* * @param task*            the runnable under investigation*/public void start(Runnable task) {long start = 0;int runs = 0;do {if (++runs > 5) {throw new IllegalStateException("Test not accurate");}expired = false;start = System.currentTimeMillis();Timer timer = new Timer();timer.schedule(new TimerTask() {public void run() {expired = true;}}, testtime);while (!expired) {task.run();}start = System.currentTimeMillis() - start;timer.cancel();} while (Math.abs(start - testtime) > EPSYLON);collectGarbage(3);}private void collectGarbage(int times) {for (int i = 0; i < times; i++) {System.gc();try {Thread.sleep(10);} catch (InterruptedException e) {Thread.currentThread().interrupt();break;}}}/*** Calculates the memory usage of a single element in a work queue. Based on Heinz Kabbutz' ideas* (http://www.javaspecialists.eu/archive/Issue029.html).* * @return memory usage of a single {@link Runnable} element in the thread pools work queue*/public long calculateMemoryUsage() {BlockingQueue<Runnable> queue = createWorkQueue();for (int i = 0; i < SAMPLE_QUEUE_SIZE; i++) {queue.add(creatTask());}long mem0 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();long mem1 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();queue = null;collectGarbage(15);mem0 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();queue = createWorkQueue();for (int i = 0; i < SAMPLE_QUEUE_SIZE; i++) {queue.add(creatTask());}collectGarbage(15);mem1 = Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory();return (mem1 - mem0) / SAMPLE_QUEUE_SIZE;}/*** Create your runnable task here.* * @return an instance of your runnable task under investigation*/protected abstract Runnable creatTask();/*** Return an instance of the queue used in the thread pool.* * @return queue instance*/protected abstract BlockingQueue<Runnable> createWorkQueue();/*** Calculate current cpu time. Various frameworks may be used here, depending on the operating system in use. (e.g.* http://www.hyperic.com/products/sigar). The more accurate the CPU time measurement, the more accurate the results* for thread count boundaries.* * @return current cpu time of current thread*/protected abstract long getCurrentThreadCPUTime();}

示例類，需要繼承上面的調優工具類，運行一下會得到結果。
其中的Runnable 方法，指的就是我們實際項目中需要運行的任務，然後由他來估算，運行這些任務的線程線程數應該配多大，BlockingQueue應該配置多大

public class MyPoolSizeCalculator extends PoolSizeCalculator {public static void main(String[] args) {MyPoolSizeCalculator calculator = new MyPoolSizeCalculator();calculator.calculateBoundaries(// CPU目标利用率new BigDecimal(1.0),// blockingqueue占用的内存大小，bytenew BigDecimal(100000));ThreadPoolExecutor executor =new ThreadPoolExecutor(8,8,// 默认情况下指的是非核心线程的空闲时间// 如果allowCoreThreadTimeOut=true：核心线程/非核心线程允许的空闲时间10L,TimeUnit.SECONDS,new LinkedBlockingQueue<>(2500),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());}protected long getCurrentThreadCPUTime() {// 当前线程占用的总时间return ManagementFactory.getThreadMXBean().getCurrentThreadCpuTime();}protected Runnable creatTask() {return new AsynchronousTask();}protected BlockingQueue createWorkQueue() {return new LinkedBlockingQueue<>();}}class AsynchronousTask implements Runnable {@Overridepublic void run() {System.out.println(Thread.currentThread().getName());}
}

運行結果如下，建議線程數配置成8（* Optimal thread count: 8）

Target queue memory usage (bytes): 100000
createTask() produced com.imooc.jvm.threadpool.AsynchronousTask which took 40 bytes in a queue
Formula: 100000 / 40
* Recommended queue capacity (bytes): 2500
Number of CPU: 8
Target utilization: 1
Elapsed time (nanos): 3000000000
Compute time (nanos): 3000000000
Wait time (nanos): 0
Formula: 8 * 1 * (1 + 0 / 3000000000)
* Optimal thread count: 8Process finished with exit code 0

計算BlockingQueue，把示例類中的System.out.println(Thread.currentThread().getName());注釋掉，再次運行就可以了。為什麼要注釋掉呢？我也不清楚。實際用的過程中，我們需要把AsynchronousTask中的方法設置為空。
運行結果如下：建議設置成2500（）

Target queue memory usage (bytes): 100000
createTask() produced com.imooc.jvm.threadpool.AsynchronousTask which took 40 bytes in a queue
Formula: 100000 / 40
* Recommended queue capacity (bytes): 2500
Number of CPU: 8
Target utilization: 1
Elapsed time (nanos): 3000000000
Compute time (nanos): 2984375000
Wait time (nanos): 15625000
Formula: 8 * 1 * (1 + 15625000 / 2984375000)
* Optimal thread count: 8

自己的項目需要根據實際情況進行計算，比如：
業務評估，之後根據公式進行計算，這個上面有說到。
逐步調整的話，就是通過壓測，將工具計算出來的結果,比如說為8，我們可以在測試一下7,9，10等，在公式計算出來的結果周邊進行滑動，多次對比，然後找到最優解。

2-7 線程池總結

需要掌握的有一下內容，可以幫助我們應付面試，在實際工作中也會有用。