利用ExecutorService實現線程池

線程池表明一組執行任務的工做線程，每一個線程能夠屢次重用。若是在全部線程都處於活動狀態時提交了新任務，則它們將在隊列中等待，直到線程可用。 線程池實如今內部使用LinkedBlockingQueue向隊列添加和刪除任務。咱們一般想要的是一個工做隊列與一組固定的工做線程相結合，它使用wait（）和notify（）來表示新工做已到達的等待線程。 介紹完線程池原理，咱們來介紹如何用ExecutorService來實現線程池。

ReentrantLock

ReentrantLock是一種獨佔鎖，只能有一個線程獲取鎖，而且一個線程能夠屢次lock。ReentrantLock提供公平鎖與非公平鎖，默認非公平鎖，

ReentrantLock是Lock接口的默認實現。 lock接口定義

public interface Lock {
	//上鎖(不響應Thread.interrupt()直到獲取鎖)
    void lock();
	//上鎖(響應Thread.interrupt())
    void lockInterruptibly() throws InterruptedException;
	//嘗試獲取鎖(以nonFair方式獲取鎖)
    boolean tryLock();
  	//在指定時間內嘗試獲取鎖(響應Thread.interrupt(),支持公平/二階段非公平)
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
	//解鎖
    void unlock();
	//獲取Condition
    Condition newCondition();
}
複製代碼

代碼實現

import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


public class CountableThreadPool {

    private Logger logger = LoggerFactory.getLogger(getClass());

    private int threadNum;


    private int threadNum;  //固定線城池大小

    private AtomicInteger threadAlive = new AtomicInteger();  //活動線程計數

    private Condition condition = reentrantLock.newCondition();

    public CountableThreadPool(int threadNum) {
        this.threadNum = threadNum;
        this.executorService = Executors.newFixedThreadPool(threadNum);
    }

    public CountableThreadPool(int threadNum, ExecutorService executorService) {
        this.threadNum = threadNum;
        this.executorService = executorService;
    }

    public void setExecutorService(ExecutorService executorService) {
        this.executorService = executorService;
    }

    public int getThreadAlive() {
        return threadAlive.get();
    }

    public int getThreadNum() {
        return threadNum;
    }

    private ExecutorService executorService;

    public void execute(final Runnable runnable) {


        if (threadAlive.get() >= threadNum) {
            try {
                reentrantLock.lock();
                while (threadAlive.get() >= threadNum) {
                    try {
                        condition.await();
                    } catch (InterruptedException e) {
                        logger.error(Throwables.getStackTraceAsString(e));
                    }
                }
            } finally {
                reentrantLock.unlock();
            }
        }
        threadAlive.incrementAndGet();
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    runnable.run();
                } finally {
                    try {
                        reentrantLock.lock();
                        threadAlive.decrementAndGet();
                        condition.signal();
                    } finally {
                        reentrantLock.unlock();
                    }
                }
            }
        });
    }

    public boolean isShutdown() {
        return executorService.isShutdown();
    }

    public void shutdown() {
        executorService.shutdown();
    }

    public static void main(String[] args) {
        CountableThreadPool threadPool = new CountableThreadPool(10);
        for(int i = 0;i < 10; i++) {
            threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    //TODO
                }
            });
        }
    }

}
複製代碼