谷歌KDD'23工作：如何提升推薦系統Ranking模型訓練穩定性
谷歌在KDD 2023發表了一篇工作，探索了推薦系統ranking模型的訓練穩定性問題，分析了造成訓練穩定性存在問題的潛在原因，以及現有的一些提升模型穩定性方法的不足，并提出了一種新
雖然 Java 對線程的創建、中斷、等待、通知、銷毀、同步等功能提供了很多的支持，但是從操作系統角度來說，頻繁的創建線程和銷毀線程，其實是需要大量的時間和資源的。
例如，當有多個任務同時需要處理的時候，一個任務對應一個線程來執行，以此來提升任務的執行效率，模型圖如下：
圖片
如果任務數非常少，這種模式倒問題不大，但是如果任務數非常的多，可能就會存在很大的問題：
假如把很多任務讓一組線程來執行，而不是一個任務對應一個新線程，這種通過接受任務并進行分發處理的就是線程池。
圖片
線程池內部維護了若干個線程，當沒有任務的時候，這些線程都處于等待狀態；當有新的任務進來時，就分配一個空閑線程執行；當所有線程都處于忙碌狀態時，新任務要么放入隊列中等待，要么增加一個新線程進行處理，要么直接拒絕。
很顯然，這種通過線程池來執行多任務的思路，優勢明顯：
關于這一點，我們可以看一個簡單的對比示例。
/** * 使用一個任務對應一個線程來執行 * @param args */public static void main(String[] args) { long startTime = System.currentTimeMillis(); final Random random = new Random(); List<Integer> list = new CopyOnWriteArrayList<>(); // 一個任務對應一個線程，使用20000個線程執行任務 for (int i = 0; i < 20000; i++) { new Thread(new Runnable() { @Override public void run() { list.add(random.nextInt(100)); } }).start(); } // 等待任務執行完畢 while (true){ if(list.size() >= 20000){ break; } } System.out.println("一個任務對應一個線程，執行耗時：" + (System.currentTimeMillis() - startTime) + "ms");}/** * 使用線程池進行執行任務 * @param args */public static void main(String[] args) { long startTime = System.currentTimeMillis(); final Random random = new Random(); List<Integer> list = new CopyOnWriteArrayList<>(); // 使用線程池進行執行任務，默認4個線程 ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 4, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(20000)); for (int i = 0; i < 20000; i++) { // 提交任務 executor.submit(new Runnable() { @Override public void run() { list.add(random.nextInt(100)); } }); } // 等待任務執行完畢 while (true){ if(list.size() >= 20000){ break; } } System.out.println("使用線程池，執行耗時：" + (System.currentTimeMillis() - startTime) + "ms"); // 關閉線程池 executor.shutdown();}兩者執行耗時情況對比，如下：
一個任務對應一個線程，執行耗時：3073ms---------------------------使用線程池，執行耗時：578ms從結果上可以看出，同樣的任務數，采用線程池和不采用線程池，執行耗時差距非常明顯，一個任務對應一個新的線程來執行，反而效率不如采用 4 個線程的線程池執行的快。
為什么會產生這種現象，下面我們就一起來聊聊線程池。
站在專業的角度講，線程池其實是一種利用池化思想來實現線程管理的技術，它將線程的創建和任務的執行進行解耦，同時復用已經創建的線程來降低頻繁創建和銷毀線程所帶來的資源消耗。通過合理的參數設置，可以實現更低的系統資源使用率、更高的任務并發執行效率。
在 Java 中，線程池最頂級的接口是Executor，名下的實現類關系圖如下：
圖片
關鍵接口和實現類，相關的描述如下：
整個關系圖中，其中ThreadPoolExecutor是線程池最核心的實現類，開發者可以使用它來創建線程池。
ThreadPoolExecutor類的完整構造方法一共有七個參數，理解這些參數的配置對使用好線程池至關重要，完整的構造方法核心源碼如下：
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)各個參數的解讀如下：
創建完線程池之后就可以提交任務了，當有新的任務進來時，線程池就會工作并分配線程去執行任務。
ThreadPoolExecutor的典型用法如下：
// 創建固定大小的線程池ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 4, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(100));// 提交任務executor.execute(task1);executor.execute(task2);executor.execute(task3);...針對任務的提交方式，ThreadPoolExecutor還提供了兩種方法。
ThreadPoolExecutor執行提交的任務流程雖然比較復雜，但是通過對源碼的分析，大致的任務執行流程，可以用如下圖來概括。
整個執行流程，大體步驟如下：
我們再回頭來看上文提到的ThreadPoolExecutor構造方法中的七個參數，這些參數會直接影響線程的執行情況，各個參數的變化情況，可以用如下幾點來概括：
ThreadPoolExecutor執行任務的部分核心源碼如下！
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // 工作線程數量 < corePoolSize，直接創建線程執行任務 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 工作線程數量 >= corePoolSize，將任務添加至阻塞隊列中 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 往阻塞隊列中添加任務的時候，如果線程池非運行狀態，將任務remove，并執行拒絕策略 if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 阻塞隊列已滿，嘗試添加新的線程去執行，如果工作線程數量 >= maximumPoolSize，執行拒絕策略 else if (!addWorker(command, false)) reject(command);}private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 線程池狀態處于非 RUNNING 狀態，添加worker失敗 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; // 判斷線程池中線程數量大于等于該線程池允許的最大線程數量，如果大于則worker失敗，反之cas更新線程池中的線程數 for (;;) { int wc = workerCountOf(c); if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { // 創建工作線程 w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { // 如果線程池處于 RUNNING 狀態并且線程已經啟動，則拋出線程異常啟動 if (t.isAlive()) throw new IllegalThreadStateException(); // 將線程加入已創建的工作線程集合，更新用于追蹤線程池中線程數量 largestPoolSize 字段 workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 啟動線程執行任務 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted;}final void runWorker(Worker w) { // 獲取執行任務線程 Thread wt = Thread.currentThread(); // 獲取執行任務 Runnable task = w.firstTask; // 將worker中的任務置空 w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { // 從當前工作線程種獲取任務，或者循環從阻塞任務隊列中獲取任務 while (task != null || (task = getTask()) != null) { w.lock(); // 雙重檢查線程池是否正在停止，如果線程池停止，并且當前線程能夠中斷，則中斷線程 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { // 前置執行任務鉤子函數 beforeExecute(wt, task); Throwable thrown = null; try { // 執行當前任務 task.run(); } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { // 后置執行任務鉤子函數 afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { // 回收線程 processWorkerExit(w, completedAbruptly); }}final void reject(Runnable command) { // 執行拒絕策略 handler.rejectedExecution(command, this);}當線程池中的線程數大于等于 maximumPoolSize，并且 workQueue 已滿，新任務會被拒絕，使用RejectedExecutionHandler接口的rejectedExecution()方法來處理被拒絕的任務。
線程池提供了四種拒絕策略實現類來拒絕任務，具體如下：
類 | 描述 |
AbortPolicy | 直接拋出一個RejectedExecutionException，這也是JDK默認的拒絕策略 |
DiscardPolicy | 什么也不做，直接丟棄任務 |
DiscardOldestPolicy | 將阻塞隊列中的任務移除出來，然后執行當前任務 |
CallerRunsPolicy | 嘗試直接運行被拒絕的任務，如果線程池已經被關閉了，任務就被丟棄了 |
我們知道 Java 種的線程一共 6 種狀態，其實線程池也有狀態。
因為線程池也是異步執行的，有的任務正在執行，有的任務存儲在任務隊列中，有的線程處于工作狀態，有的線程處于空閑狀態等待回收，為了更加精細化的管理線程池，線程池也設計了 5 中狀態，部分核心源碼如下：
public class ThreadPoolExecutor extends AbstractExecutorService { // 線程池線程數的bit數 private static final int COUNT_BITS = Integer.SIZE - 3; // 線程池狀態 private static final int RUNNING = -1 << COUNT_BITS; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS;}其中的狀態流程，可以用如下圖來描述！
這幾個狀態的轉化關系，可以用如下幾個步驟來概括：
正如文章的開頭所介紹的，使用線程池的方式，通常可以用如下幾個步驟來概括：
// 1.創建固定大小為4的線程數、空閑線程的存活時間為15秒、阻塞任務隊列的上限為1000的線程池完整示例ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 4, 15, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());// 2.提交任務executor.submit(task1);executor.submit(task2);executor.submit(task3);...// 3.使用完畢之后，可以手動關閉線程池executor.shutdown();正如上文所說，其中execute()和submit()方法都可以用來提交任務，稍有不同的是：submit()方法同時還支持獲取任務執行完畢的返回結果。
針對線程池的使用，Java 還提供了Executors工具類，開發者可以通過此工具，快速創建不同類型的線程池。
下面我們一起來看下Executors為用戶提供的幾種創建線程池的方法。
newSingleThreadExecutor()方法表示創建一個單線程的線程池，核心源碼如下：
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));}從構造參數上可以很清晰的看到，線程池中的線程數為 1，不會被線程池自動回收，workQueue 選擇的是無界的LinkedBlockingQueue阻塞隊列，不管來多少任務存入阻塞隊列中，前面一個任務執行完畢，再執行隊列中的剩余任務。
簡單應用示例如下：
public static void main(String[] args) { long startTime = System.currentTimeMillis(); final Random random = new Random(); List<Integer> list = new CopyOnWriteArrayList<>(); // 創建一個單線程線程池 ExecutorService executor = Executors.newSingleThreadExecutor(); for (int i = 0; i < 10; i++) { executor.submit(new Runnable() { @Override public void run() { list.add(random.nextInt(100)); System.out.println("thread name:" + Thread.currentThread().getName()); } }); } while (true){ if(list.size() >= 10){ break; } } System.out.println("執行耗時：" + (System.currentTimeMillis() - startTime) + "ms"); // 關閉線程池 executor.shutdown();}運行結果如下：
thread name:pool-1-thread-1thread name:pool-1-thread-1thread name:pool-1-thread-1thread name:pool-1-thread-1thread name:pool-1-thread-1thread name:pool-1-thread-1thread name:pool-1-thread-1thread name:pool-1-thread-1thread name:pool-1-thread-1thread name:pool-1-thread-1執行耗時：13msnewFixedThreadPool()方法表示創建一個固定大小線程數的線程池，核心源碼如下：
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());}固定大小的線程池和單線程的線程池有異曲同工之處，無非是讓線程池中能運行的線程數量支持手動指定。
簡單應用示例如下：
public static void main(String[] args) { long startTime = System.currentTimeMillis(); final Random random = new Random(); List<Integer> list = new CopyOnWriteArrayList<>(); // 創建固定大小線程數為3的線程池 ExecutorService executor = Executors.newFixedThreadPool(3); for (int i = 0; i < 10; i++) { executor.submit(new Runnable() { @Override public void run() { list.add(random.nextInt(100)); System.out.println("thread name:" + Thread.currentThread().getName()); } }); } while (true){ if(list.size() >= 10){ break; } } System.out.println("執行耗時：" + (System.currentTimeMillis() - startTime) + "ms"); // 關閉線程池 executor.shutdown();}運行結果如下：
thread name:pool-1-thread-2thread name:pool-1-thread-1thread name:pool-1-thread-3thread name:pool-1-thread-3thread name:pool-1-thread-3thread name:pool-1-thread-1thread name:pool-1-thread-3thread name:pool-1-thread-2thread name:pool-1-thread-2thread name:pool-1-thread-1執行耗時：10msnewCachedThreadPool()方法表示創建一個可緩存的無界線程池，核心源碼如下：
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());}從構造參數上可以看出，線程池中的最大線程數為Integer.MAX_VALUE，也就是Integer的最大值，workQueue 選擇的是SynchronousQueue阻塞隊列，這個阻塞隊列不像LinkedBlockingQueue，它沒有容量，只負責做臨時任務緩存，如果有任務進來立刻會被執行。
也就是說，只要添加進去了任務，線程就會立刻去執行，當任務超過線程池的線程數則創建新的線程去執行，線程數量的最大上線為Integer.MAX_VALUE，當線程池中的線程空閑時間超過 60s，則會自動回收該線程。
簡單應用示例如下：
public static void main(String[] args) { long startTime = System.currentTimeMillis(); final Random random = new Random(); List<Integer> list = new CopyOnWriteArrayList<>(); // 創建可緩存的無界線程池 ExecutorService executor = Executors.newCachedThreadPool(); for (int i = 0; i < 10; i++) { executor.submit(new Runnable() { @Override public void run() { list.add(random.nextInt(100)); System.out.println("thread name:" + Thread.currentThread().getName()); } }); } while (true){ if(list.size() >= 10){ break; } } System.out.println("執行耗時：" + (System.currentTimeMillis() - startTime) + "ms"); // 關閉線程池 executor.shutdown();}運行結果如下：
thread name:pool-1-thread-1thread name:pool-1-thread-2thread name:pool-1-thread-3thread name:pool-1-thread-4thread name:pool-1-thread-3thread name:pool-1-thread-2thread name:pool-1-thread-1thread name:pool-1-thread-4thread name:pool-1-thread-4thread name:pool-1-thread-4執行耗時：13msnewScheduledThreadPool()方法表示創建周期性的線程池，可以指定線程池中的核心線程數，支持定時及周期性任務的執行，核心源碼如下：
public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue());}從構造參數上可以看出，線程池支持指定核心線程數，最大線程數為Integer.MAX_VALUE，workQueue 選擇的是DelayedWorkQueue延遲阻塞隊列，這個阻塞隊列支持任務延遲消費，新加入的任務不會立刻被執行，只有時間到期之后才會被取出；當非核心線程處于空閑狀態時，會立刻進行收回。
ScheduledExecutorService支持三種類型的定時調度方法，分別如下：
下面我們一起來看看它們的應用方式。
SimpleDateFormat sdf =new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");// 創建線程數量為2的定時調度線程池ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);System.out.println(sdf.format(new Date()) + " 準備啟動");// 定時執行一次的任務，延遲1s后執行executor.schedule(new Runnable() { @Override public void run() { System.out.println(sdf.format(new Date()) + " thread name:" + Thread.currentThread().getName() + ", schedule"); }}, 1, TimeUnit.SECONDS);輸出結果：
2023-11-17 01:41:12 準備啟動2023-11-17 01:41:13 thread name:pool-1-thread-1, scheduleSimpleDateFormat sdf =new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");// 創建線程數量為2的定時調度線程池ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);System.out.println(sdf.format(new Date()) + " 準備啟動");// 周期性地執行任務，第一個任務延遲1s后執行，之后每隔2s周期性執行任務，需要等待上一次的任務執行完畢才執行下一個executor.scheduleAtFixedRate(new Runnable() { @Override public void run() { System.out.println(sdf.format(new Date()) + " thread name:" + Thread.currentThread().getName() + " begin"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(sdf.format(new Date()) + " thread name:" + Thread.currentThread().getName() + " end"); }}, 1, 2, TimeUnit.SECONDS);輸出結果：
2023-11-17 02:00:44 準備啟動2023-11-17 02:00:45 thread name:pool-1-thread-1 begin2023-11-17 02:00:48 thread name:pool-1-thread-1 end2023-11-17 02:00:48 thread name:pool-1-thread-1 begin2023-11-17 02:00:51 thread name:pool-1-thread-1 end2023-11-17 02:00:51 thread name:pool-1-thread-1 begin2023-11-17 02:00:54 thread name:pool-1-thread-1 endSimpleDateFormat sdf =new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");// 創建線程數量為2的定時調度線程池ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);System.out.println(sdf.format(new Date()) + " 準備啟動");// 周期性地執行任務，第一個任務延遲1s后執行，之后上一個任務執行完畢之后，延遲2秒再執行下一個任務executor.scheduleWithFixedDelay(new Runnable() { @Override public void run() { System.out.println(sdf.format(new Date()) + " thread name:" + Thread.currentThread().getName() + " begin"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(sdf.format(new Date()) + " thread name:" + Thread.currentThread().getName() + " end"); }}, 1, 2, TimeUnit.SECONDS);輸出結果：
2023-11-17 01:53:26 準備啟動2023-11-17 01:53:27 thread name:pool-1-thread-1 begin2023-11-17 01:53:30 thread name:pool-1-thread-1 end2023-11-17 01:53:32 thread name:pool-1-thread-1 begin2023-11-17 01:53:35 thread name:pool-1-thread-1 end2023-11-17 01:53:37 thread name:pool-1-thread-1 begin2023-11-17 01:53:40 thread name:pool-1-thread-1 end從以上的介紹中，我們可以對這四種線程池的參數做一個匯總，內容如下表：
工廠方法 | corePoolSize | maximumPoolSize | keepAliveTime | workQueue |
newSingleThreadExecutor | 1 | 1 | 0 | LinkedBlockingQueue |
newFixedThreadPool | nThreads | nThreads | 0 | LinkedBlockingQueue |
newCachedThreadPool | 0 | Integer.MAX_VALUE | 60s | SynchronousQueue |
newScheduledThreadPool | corePoolSize | Integer.MAX_VALUE | 0 | DelayedWorkQueue |
這四個線程池，主要的區別在于：corePoolSize、maximumPoolSize、keepAliveTime、workQueue 這四個參數，其中線程工廠為默認類DefaultThreadFactory，線程飽和的拒絕策略為默認類AbortPolicy。
結合以上的分析，最后我們再來總結一下。
對于線程池的使用，不太建議采用Executors工具去創建，盡量通過ThreadPoolExecutor的構造方法來創建，原因在于：有利于規避資源耗盡的風險；同時建議開發者手動設定任務隊列的上限，防止服務出現 OOM。
雖然Executors工具提供了四種創建線程池的方法，能幫助開發者省去繁瑣的參數配置，但是newSingleThreadExecutor和newFixedThreadPool方法創建的線程池，任務隊列上限為Integer.MAX_VALUE，這意味著可以無限提交任務，這在高并發的環境下，系統可能會出現 OOM，導致整個線程池不可用；其次newCachedThreadPool方法也存在同樣的問題，無限的創建線程可能會給系統帶來更多的資源消耗。
其次，創建線程池的時候應該盡量給線程定義一個具體的業務名字前綴，方便定位問題，不同類型的業務盡量使用不同的線程池來實現。
例如可以使用guava包，創建自定義的線程工廠。
ThreadFactory threadFactory = new ThreadFactoryBuilder() .setNameFormat(threadNamePrefix + "-%d") .setDaemon(true).build();當然，你也可以自行實現一個線程工廠，需要繼承ThreadFactory接口，案例如下：
import java.util.concurrent.Executors;import java.util.concurrent.ThreadFactory;import java.util.concurrent.atomic.AtomicInteger;/** * 線程工廠，它設置線程名稱，有利于我們定位問題。 */public final class NamingThreadFactory implements ThreadFactory { private final AtomicInteger threadNum = new AtomicInteger(); private final ThreadFactory delegate; private final String name; /** * 創建一個帶名字的線程池生產工廠 */ public NamingThreadFactory(ThreadFactory delegate, String name) { this.delegate = delegate; this.name = name; } @Override public Thread newThread(Runnable r) { Thread t = delegate.newThread(r); t.setName(name + "-" + threadNum.incrementAndGet()); return t; }}創建一個線程名稱以order開頭的線程工廠。
NamingThreadFactory threadFactory = new NamingThreadFactory(Executors.defaultThreadFactory(), "order");最后，再來說說關于線程池中線程數，如何合理設定的問題？
那如何判斷當前是 CPU 密集型任務還是 I/O 密集型任務呢？
最簡單的方法就是：如果當前任務涉及到網絡讀取，文件讀取等，這類都是 IO 密集型任務，除此之外，可以看成是 CPU 密集型任務。
本文篇幅比較長，難免有描述不對的地方，歡迎大家留言指出！
本文鏈接：http://www.www897cc.com/showinfo-26-43325-0.html一文帶你徹底弄懂線程池
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
上一篇： JS小知識，分享一些讓我迷惑的前端面試題
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號