package com.yami.trading.common.executor;

import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/* 
 * 一个自定义的阻塞式线程池.
 * @author caster
 *
 **/ 
public class BlockingThreadPoolExecutor extends ThreadPoolExecutor {
	protected static org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BlockingThreadPoolExecutor.class);
	
	private final ReentrantLock taskLock = new ReentrantLock();
	private final Condition unpaused = taskLock.newCondition();
	private final Condition submitCnd = taskLock.newCondition();
	private final int maxTaskCount;
	private volatile int currentTaskCount; // 正在执行的任务数量
	private volatile int activeTaskCount; // 已提交待执行的任务数量+正在执行的任务数量

	public BlockingThreadPoolExecutor(int fixedThreadPoolSize) {
		this(fixedThreadPoolSize, fixedThreadPoolSize, 60, TimeUnit.SECONDS);
	}
	
	public BlockingThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit) {
		// 经测试，好像是：正在执行的线程数量+等待队列的数量（容量还不能设置为0）=maximumPoolSize
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, new ArrayBlockingQueue<Runnable>(1), new AbortPolicy());
		this.maxTaskCount = maximumPoolSize + 1; //SynchronousQueue
	}

	@Override 
	public Future<?> submit(Runnable task) {
		taskLock.lock(); 
		try { 
			// 提交新任务时，如果发现线程池中所有任务数量（正在处理的+队列中待处理的）超过了允许的最大数量，则堵塞提交
			while (maxTaskCount <= activeTaskCount) {
				try {
					logger.warn("[BlockingThreadPoolExecutor submit] 线程池满了，无法执行当前任务, maxTaskCount:" + maxTaskCount + ", activeTaskCount:" + activeTaskCount);
					submitCnd.await();
				} catch (InterruptedException e) {
					throw new RuntimeException("线程池被中断...");
				} 
			}
		} finally { 
			taskLock.unlock(); 
		}

		// 内部提交请求，并且增加有效任务数量
		Future<?> submitResult = super.submit(task);
		activeTaskCount++;
		logger.info("[BlockingThreadPoolExecutor submit] 线程池添加了一个任务, maxTaskCount:" + maxTaskCount + ", activeTaskCount:" + activeTaskCount);
		return submitResult;
	}

	@Override
	public <T> Future<T> submit(Runnable task, T result) {
		taskLock.lock(); 
		try {
			// 提交新任务时，如果发现线程池中所有任务数量（正在处理的+队列中待处理的）超过了允许的最大数量，则堵塞提交
			while (maxTaskCount <= activeTaskCount) { 
				try {
					logger.warn("[BlockingThreadPoolExecutor submit] 线程池满了，无法执行当前任务, maxTaskCount:" + maxTaskCount + ", activeTaskCount:" + activeTaskCount);
					submitCnd.await();
				} catch (InterruptedException e) {
					throw new RuntimeException("线程池被中断...");
				} 
			}
		} finally { 
			taskLock.unlock(); 
		}
		
		Future<T> submitResult = super.submit(task, result);
		activeTaskCount++;
		logger.info("[BlockingThreadPoolExecutor submit] 线程池添加了一个任务, maxTaskCount:" + maxTaskCount + ", activeTaskCount:" + activeTaskCount);
		return submitResult;
	}
	
	@Override
	public <T> Future<T> submit(Callable<T> task) {
		taskLock.lock(); 
		try {
			// 提交新任务时，如果发现线程池中所有任务数量（正在处理的+队列中待处理的）超过了允许的最大数量，则堵塞提交
			while (maxTaskCount <= activeTaskCount) { 
				try {
					logger.warn("[BlockingThreadPoolExecutor submit] 线程池满了，无法执行当前任务, maxTaskCount:" + maxTaskCount + ", activeTaskCount:" + activeTaskCount);
					submitCnd.await();
				} catch (InterruptedException e) {
					throw new RuntimeException("线程池被中断...");
				} 
			}
		} finally { 
			taskLock.unlock(); 
		}
		
		Future<T> submitResult = super.submit(task);
		activeTaskCount++;
		logger.info("[BlockingThreadPoolExecutor submit] 线程池添加了一个任务, maxTaskCount:" + maxTaskCount + ", activeTaskCount:" + activeTaskCount);
		return submitResult;
	}


	/**
	 * 选中一个线程开始处理当前业务的前置调用.
	 *
	 * @param t
	 * @param r
	 */
	@Override 
	protected void beforeExecute(Thread t, Runnable r) { 
		super.beforeExecute(t, r); 
		taskLock.lock(); 
		try { 
			// 正式处理的任务数量如果达到了允许的活跃线程池上限，则阻塞等待
			// 正常情况下不应该有此种情况出现，监控下日志
			while (maxTaskCount <= currentTaskCount + 1) {
				try {
					logger.warn("[BlockingThreadPoolExecutor beforeExecute] =======> maxTaskCount:{}, currentTaskCount:{}", maxTaskCount, currentTaskCount);
					unpaused.await();
				} catch (InterruptedException e) { 
					t.interrupt(); 
				} 
			} 
			currentTaskCount++; 
			logger.info("[BlockingThreadPoolExecutor beforeExecute] 有一个线程池任务开始正式执行, maxTaskCount:" + maxTaskCount + ", activeTaskCount:" + activeTaskCount + ", currentTaskCount:" + currentTaskCount);
		} finally {
			taskLock.unlock(); 
		} 
	}

	/**
	 * 一个线程完成业务处理后执行.
	 *
	 * @param r
	 * @param t
	 */
	@Override 
	protected void afterExecute(Runnable r, Throwable t) { 
		super.afterExecute(r, t); 
		taskLock.lock(); 
		try {
			// 正在执行的业务处理数减一
			currentTaskCount--;
			// 总的业务处理数量减一
			activeTaskCount--;

			logger.info("[BlockingThreadPoolExecutor afterExecute] 完成了一个线程池任务, currentTaskCount:" + currentTaskCount + ", activeTaskCount:" + activeTaskCount);

			// unpaused 可以考虑是否有必要
			unpaused.signalAll();
			// 有一个线程完成了业务处理，代表有空闲线程可以接受新的提交请求
			submitCnd.signalAll();
		} finally {
			taskLock.unlock(); 
		} 
	}

	/**
	 * 提前当前时刻正在执行中的任务数量(不包括处于等待队列的任务).
	 *
	 * @return
	 */
	public int getRunningTaskCount() {
		return this.currentTaskCount;
	}

	/**
	 * 获取正在执行的任务 + 队列中待执行的任务总数.
	 * 
	 * @return
	 */
	public int getUnFinishedTaskCount() {
		taskLock.lock();
		try {
			int count = this.currentTaskCount;
			count = count + this.getQueue().size();

			return count;
		} finally {
			taskLock.unlock();
		}
	}
}