解決線程池中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。