解決線程池中ThreadGroup的坑
線程池中ThreadGroup的坑
在Java中每一個線程都歸屬於某個線程組管理的一員,例如在主函數main()主工作流程中產生一個線程,則產生的線程屬於main這個線程組管理的一員。簡單地說,線程組(ThreadGroup)就是由線程組成的管理線程的類,這個類是java.lang.ThreadGroup類。
定義一個線程組,通過以下代碼可以實現。
ThreadGroup group=new ThreadGroup(“groupName”); Thread thread=new Thread(group,”the first thread of group”);
ThreadGroup類中的某些方法,可以對線程組中的線程產生作用。例如,setMaxPriority()方法可以設定線程組中的所有線程擁有最大的優先權。
所有線程都隸屬於一個線程組。那可以是一個默認線程組(不指定group),亦可是一個創建線程時明確指定的組。在創建之初,線程被限制到一個組裡,而且不能改變到一個不同的組。每個應用都至少有一個線程從屬於系統線程組。若創建多個線程而不指定一個組,它們就會自動歸屬於系統線程組。
線程組也必須從屬於其他線程組。必須在構建器裡指定新線程組從屬於哪個線程組。若在創建一個線程組的時候沒有指定它的歸屬,則同樣會自動成為系統線程組的一名屬下。因此,一個應用程序中的所有線程組最終都會將系統線程組作為自己的“父”。
那麼假如我們需要在線程池中實現一個帶自定義ThreadGroup的線程分組,該怎麼實現呢?
我們在給線程池(ThreadPoolExecutor)提交任務的時候可以通過execute(Runnable command)來將一個線程任務加入到該線程池,那麼我們是否可以通過new一個指定瞭ThreadGroup的Thread實例來加入線程池來達到前面說到的目的呢?
ThreadGroup是否可行
通過new Thread(threadGroup,runnable)實現線程池中任務分組
public static void main(String[] args) { ThreadPoolExecutor pool = (ThreadPoolExecutor) Executors.newCachedThreadPool(); final ThreadGroup group = new ThreadGroup("Main_Test_Group"); for (int i = 0; i < 5; i++) { Thread thread = new Thread(group, new Runnable() { @Override public void run() { int sleep = (int)(Math.random() * 10); try { Thread.sleep(1000 * 3); System.out.println(Thread.currentThread().getName()+"執行完畢"); System.out.println("當前線程組中的運行線程數"+group.activeCount()); } catch (InterruptedException e) { e.printStackTrace(); } } }, group.getName()+" #"+i+""); pool.execute(thread); } }
運行結果
pool-1-thread-3執行完畢
pool-1-thread-1執行完畢
當前線程組中的運行線程數0
pool-1-thread-2執行完畢
當前線程組中的運行線程數0
當前線程組中的運行線程數0
pool-1-thread-4執行完畢
pool-1-thread-5執行完畢
當前線程組中的運行線程數0
當前線程組中的運行線程數0
運行結果中可以看到group中的線程並沒有因為線程池啟動瞭這個線程任務而運行起來.因此通過線程組來對線程池中的線層任務分組不可行.
從java.util.concurrent.ThreadPoolExecutor源碼中可以看到如下構造函數:
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
如果我們在實例化ThreadPoolExecutor時不指定ThreadFactory,那麼將以默認的ThreadFactory來創建Thread.
Executors內部類DefaultThreadFactory
下面的源碼即是默認的Thread工廠
static class DefaultThreadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix; DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } public Thread newThread(Runnable r) { Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) t.setDaemon(false); if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); return t; } }
從唯一的構造函數可以看到DefaultThreadFactory以SecurityManager 實例中的ThreadGroup來指定線程的group,如果SecurityManager 獲取到的ThreadGroup為null才默認以當前線程的group來指定.public Thread newThread(Runnable r) 則以group來new 一個Thead.這樣我們可以在實例化ThreadPoolExecutor對象的時候在其構造函數內傳入自定義的ThreadFactory實例即可達到目的.
public class MyTheadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1); private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix; private ThreadGroup defaultGroup; public MyTheadFactory() { SecurityManager s = System.getSecurityManager(); defaultGroup = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } public MyTheadFactory(ThreadGroup group) { this.defaultGroup = group; namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } public Thread newThread(Runnable r) { Thread t = new Thread(defaultGroup, null, namePrefix + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) t.setDaemon(false); if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); return t; } }
ThreadGroup的使用及手寫線程池
監聽線程異常關閉
以下代碼在window下不方便測試,需在linux 上 測試
// 以下線程如果強制關閉的話,是無法打印`線程被殺掉瞭` // 模擬關閉 kill PID public static void main(String[] args) { Runtime.getRuntime().addShutdownHook(new Thread( () -> { System.out.println("線程被殺掉瞭"); })); while(true){ System.out.println("i am working ..."); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } }
如何拿到Thread線程中異常
public static void main(String[] args) { Thread thread = new Thread(() -> { try { Thread.sleep(1000); int i = 10/0; } catch (InterruptedException e) { e.printStackTrace(); } }); thread.setUncaughtExceptionHandler((t,e)->{ System.out.println("線程的名字"+ t.getName()); System.out.println(e); }); // 通過註入接口的方式 thread.start(); }
ThreadGroup
註意: threadGroup 設置為isDaemon 後,會隨最後一個線程結束而銷毀,如果沒有設置isDaemon ,則需要手動調用 destory()
線程池使用
自己搭建的簡單線程池實現
其中ThreadGroup 的應用沒有寫,但是我們可以觀察線程關閉後,檢查ThreadGroup 中是否還有活躍的線程等,具體參考ThreadGroup API
import java.util.ArrayList; import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.stream.IntStream; /** * @Author: shengjm * @Date: 2020/2/10 9:52 * @Description: */ public class SimpleThreadPool extends Thread{ /** * 線程數量 */ private int size; private final int queueSize; /** * 默認線程隊列數量 */ private final static int DEFAULR_TASK_QUEUE_SIZE = 2000; private static volatile int seq = 0; private final static String THREAD_PREFIX = "SIMPLE_THREAD_POLL_"; private final static ThreadGroup GROUP = new ThreadGroup("Pool_Group"); private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<>(); private final static List<WorkerTask> THREAD_QUEUE = new ArrayList<>(); private final DiscardPolicy discardPolicy; private volatile boolean destory = false; private int min; private int max; private int active; /** * 定義異常策略的實現 */ private final static DiscardPolicy DEFAULT_DISCARD_POLICY = () -> { throw new DiscardException("線程池已經被撐爆瞭,後繼多餘的人將丟失"); }; /** * */ public SimpleThreadPool(){ this(4,8,12,DEFAULR_TASK_QUEUE_SIZE,DEFAULT_DISCARD_POLICY); } /** * */ public SimpleThreadPool(int min , int active , int max , int queueSize,DiscardPolicy discardPolicy) { this.min = min; this.active = active; this.max = max; this.queueSize = queueSize; this.discardPolicy = discardPolicy; init(); } /** * 初始化 */ private void init() { for(int i = 0; i < min; i++){ createWorkTask(); } this.size = min; this.start(); } private void createWorkTask(){ WorkerTask task = new WorkerTask(GROUP,THREAD_PREFIX+(seq++)); task.start(); THREAD_QUEUE.add(task); } /** * 線程池自動擴充 */ @Override public void run() { while(!destory){ System.out.println(this.min +" --- "+this.active+" --- "+this.max + " --- "+ this.size + " --- "+ TASK_QUEUE.size()); try { Thread.sleep(1000); if(TASK_QUEUE.size() > active && size < active){ for (int i = size; i < active;i++){ createWorkTask(); } size = active; }else if(TASK_QUEUE.size() > max && size < max){ for (int i = size; i < max;i++){ createWorkTask(); } size = max; } synchronized (THREAD_QUEUE){ if(TASK_QUEUE.isEmpty() && size > active){ int release = size - active; for (Iterator<WorkerTask> it = THREAD_QUEUE.iterator();it.hasNext();){ if(release <=0){ break; } WorkerTask task = it.next(); task.close(); task.interrupt(); it.remove(); release--; } size = active; } } } catch (InterruptedException e) { break; } } } public void submit(Runnable runnable){ synchronized (TASK_QUEUE){ if(destory){ throw new DiscardException("線程池已經被摧毀瞭..."); } if(TASK_QUEUE.size() > queueSize){ discardPolicy.discard(); } TASK_QUEUE.addLast(runnable); TASK_QUEUE.notifyAll(); } } /** * 關閉 */ public void shutdown(){ while(!TASK_QUEUE.isEmpty()){ try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } } synchronized (THREAD_QUEUE) { int initVal = THREAD_QUEUE.size(); while (initVal > 0) { for (WorkerTask workerTask : THREAD_QUEUE) { if (workerTask.getTaskState() == TaskState.BLOCKED) { workerTask.interrupt(); workerTask.close(); initVal--; } else { try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } } } } this.destory = true; } } public int getSize() { return size; } public int getMin() { return min; } public int getMax() { return max; } public int getActive() { return active; } /** * 線程狀態 */ private enum TaskState{ FREE , RUNNING , BLOCKED , DEAD } /** * 自定義異常類 */ public static class DiscardException extends RuntimeException{ public DiscardException(String message){ super(message); } } /** * 定義異常策略 */ @FunctionalInterface public interface DiscardPolicy{ void discard() throws DiscardException; } private static class WorkerTask extends Thread{ private volatile TaskState taskState = TaskState.FREE; public TaskState getTaskState(){ return this.taskState; } public WorkerTask(ThreadGroup group , String name){ super(group , name); } @Override public void run(){ OUTER: while(this.taskState != TaskState.DEAD){ Runnable runnable; synchronized (TASK_QUEUE){ while(TASK_QUEUE.isEmpty()){ try { taskState = TaskState.BLOCKED; TASK_QUEUE.wait(); } catch (InterruptedException e) { break OUTER; } } runnable = TASK_QUEUE.removeFirst(); } if(runnable != null){ taskState = TaskState.RUNNING; runnable.run(); taskState = TaskState.FREE; } } } public void close(){ this.taskState = TaskState.DEAD; } } /** * 測試 * @param args */ public static void main(String[] args) { SimpleThreadPool simpleThreadPool = new SimpleThreadPool(); // SimpleThreadPool simpleThreadPool = new SimpleThreadPool(6,15,SimpleThreadPool.DEFAULT_DISCARD_POLICY); IntStream.rangeClosed(0,40).forEach(i -> { simpleThreadPool.submit(() -> { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("the runnable " + i + "be servered by " + Thread.currentThread()); }); }); // try { // Thread.sleep(15000); // } catch (InterruptedException e) { // e.printStackTrace(); // } simpleThreadPool.shutdown(); } }
以上為個人經驗,希望能給大傢一個參考,也希望大傢多多支持WalkonNet。