1. 程式人生 > >《Java併發程式設計實戰》筆記3——執行緒池的使用

《Java併發程式設計實戰》筆記3——執行緒池的使用

1、執行緒飢餓死鎖

線上程池中,如果任務依賴於其他任務,那麼可能發生死鎖。在單執行緒的Executor中,如果一個任務將另一個任務提交到同一個Executor,並且等待這個被提交任務的結果,那麼通常會引發死鎖。

如下面程式碼所示:

public class ThreadDeadlock {
        
	ExecutorService exec = Executors.newSingleThreadExecutor();//使用單執行緒的Executor

	public class LoadFileTask implements Callable<String> {
		private final String fileName;

		public LoadFileTask(String fileName) {
			this.fileName = fileName;
		}

		public String call() throws Exception {
			// Here's where we would actually read the file
			return "";
		}
	}

	public class RenderPageTask implements Callable<String> {
		public String call() throws Exception {
			Future<String> header, footer;
			header = exec.submit(new LoadFileTask("header.html"));
			footer = exec.submit(new LoadFileTask("footer.html"));
			String page = renderBody();
			System.out.println("提交子任務");
			// Will deadlock -- task waiting for result of subtask
			return header.get() + page + footer.get();
		}

		private String renderBody() {
			// Here's where we would actually render the page
			return "";
		}
	}

	public static void main(String[] args) {
		ThreadDeadlock deadLock = new ThreadDeadlock();
		ThreadDeadlock.RenderPageTask t = deadLock.new RenderPageTask();
		Future<String> result = deadLock.exec.submit(t);
		System.out.println("提交主任務");
		try {
			System.out.println("獲取結果");
			result.get();
			System.out.println("===結束===");
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}
	}
}

執行結果:

2、設定執行緒池的大小

如果執行緒池過大,大量的執行緒將在相對較少的CUP和記憶體資源上發生競爭,這不僅會導致更高的記憶體使用量,而且還可能耗盡資源。如果執行緒池過小,那麼將導致許多空閒的處理器無法執行工作,從而降低吞吐率。

3、配置ThreadPoolExecutor

(1)引數簡介

     ThreadPoolExecutor實現了ExecutorService和Executors介面,通常可通過Executors工廠方法來配置引數。較為完整的構造器如下所示。

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit timeUnit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler rejectedExecutionHandler)

corePoolSize:核心執行緒數(執行緒池努力去保持的一個執行緒數量)

maximumPoolSize:最大執行緒數

keepAliveTime:執行緒空閒時間。當執行緒數>corePoolSize,空閒執行緒最多等待keepAliveTime時間,便會銷燬。

timeUnit :keepAliveTime的時間單位

workQueue:任務佇列,儲存等待執行的任務。

threadFactory :用來建立新執行緒的工廠

rejectedExecutionHandler :任務拒絕處理器。當拒絕任務時會觸發

執行順序

①執行緒數<corePoolSize,即使有其他空閒的工作執行緒,也會建立一個新執行緒來處理請求。

②如果執行緒數>=corePoolSize,會將新任務放在佇列中等待

③如果佇列滿了,且執行緒數<maximumPoolSize,才會建立一個新執行緒來處理請求。

④如果佇列滿了,且執行緒數>=maximumPoolSize,拒絕請求

如果將corePoolSize與maximumPoolSize設定成一樣,那麼會的到一個固定大小的執行緒池。

通過設定maximumPoolSize為一個無限大的數值(如Integer.MAX_VALUE),該執行緒池能適應任意數量的併發任務。

(2)管理佇列任務

     ThreadPoolExecutor可以提供一個BlockingQueue來儲存等待執行的任務。基本的任務排隊方法有三種:無界佇列、有界佇列和同步移交。

     只有當任務相互獨立時,為執行緒池或工作佇列設定界限才是合理的。如果任務之間存在依賴性,那麼有界的執行緒池或佇列就可能導致執行緒“飢餓”死鎖問題。此時應該使用無界的執行緒池,如:newCachedThreadPool。

(3)任務拒絕處理策略(RejectedExecutionHandler )

當任務新增到執行緒池中被拒絕時:

①ThreadPoolExecutor.AbortPolicy。會丟擲RejectedExecutionException異常

②ThreadPoolExecutor.CallerRunsPolicy。將某些任務回退到呼叫者,從而降低新任務的流量。它不會線上程池的某個執行緒中執行新提交的任務,而是在一個呼叫了execute的執行緒中執行該任務。

③ThreadPoolExecutor.DiscardPolicy。無法執行的任務會被簡單地拋棄

④ThreadPoolExecutor.DiscardOldestPolicy。如果executor沒有關閉,在任務佇列頭的任務會被拋棄,然後會再次嘗試重新提交新的任務(可能會再次失敗,繼續重複這個操作)

(4)執行緒工廠(ThreadFactory )

執行緒池需要建立新的執行緒時,都是通過執行緒工廠方法來完成的。但有時需要使用定製的執行緒工廠方法。

public interface ThreadFactory {
    Thread newThread(Runnable r);
}

/**
 * 自定義執行緒工廠
 */
public class MyThreadFactory implements ThreadFactory {
    private final String poolName;//執行緒池名

    public MyThreadFactory(String poolName) {
        this.poolName = poolName;
    }
    @Override
    public Thread newThread(Runnable runnable) {
        return new MyAppThread(runnable, poolName);
    }
}

 

public class MyAppThread extends Thread {
    public static final String DEFAULT_NAME = "MyAppThread";
    private static volatile boolean debugLifecycle = false;
    private static final AtomicInteger created = new AtomicInteger();
    private static final AtomicInteger alive = new AtomicInteger();
    private static final Logger log = Logger.getAnonymousLogger();

    public MyAppThread(Runnable r) {
        this(r, DEFAULT_NAME);
    }

    public MyAppThread(Runnable runnable, String name) {
        super(runnable, name + "-" + created.incrementAndGet());
        //設定日誌顯示級別
        log.setLevel(Level.FINE);
        //Set the handler invoked when this thread abruptly terminates due to an uncaught exception
        setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
            public void uncaughtException(Thread t,
                                          Throwable e) {
                log.log(Level.SEVERE,
                        "UNCAUGHT in thread " + t.getName(), e);
            }
        });
    }

	public void run() {
        // Copy debug flag to ensure consistent value throughout.
        boolean debug = debugLifecycle;
        if (debug) log.log(Level.SEVERE, "Created " + getName());
        try {
            alive.incrementAndGet();
            super.run();
        } finally {
            alive.decrementAndGet();
            if (debug) log.log(Level.SEVERE, "Exiting " + getName());
        }
    }

    public static int getThreadsCreated() {
        return created.get();
    }

    public static int getThreadsAlive() {
        return alive.get();
    }

    public static boolean getDebug() {
        return debugLifecycle;
    }

    public static void setDebug(boolean b) {
        debugLifecycle = b;
    }
}

測試:

public class TestMain {
	public static void main(String[] args) {
		MyThreadFactory factory = new MyThreadFactory("Wow");
		MyAppThread.setDebug(true);
		MyAppThread t =(MyAppThread) factory.newThread(new Runnable() {
			@Override
			public void run() {
				int a = 1/0;//沒有捕獲的異常
			}
		});
		t.start();
		//*******************************************************************//
		MyThreadFactory factory2 = new MyThreadFactory("hoo");
		MyAppThread.setDebug(true);
		MyAppThread t2 =(MyAppThread) factory2.newThread(new Runnable() {
			@Override
			public void run() {
				int a = 1/0;//沒有捕獲的異常
			}
		});
		t2.start();
	}
}

測試結果:

 

 

(5)擴充套件ThreadPoolExecutor

可以根據需求擴充套件ThreadPoolExecutor重寫beforeExecute、afterExecute、terminated等方法