[Java知识库]多线程调用工具类

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.function.Supplier;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 提高方法调用的成功率和获取执行最快方法的结果的工具类
 *
 * @author
 * @version V1.00
 * @createDate 2019-05-05
 */
public class MethodTimeoutUtils {
	private static final Logger LOGGER = LoggerFactory.getLogger(MethodTimeoutUtils.class);
	
    private static final boolean useCommonPool =
        (ForkJoinPool.getCommonPoolParallelism() > 1);

    /**
     * Default executor -- ForkJoinPool.commonPool() unless it cannot
     * support parallelism.
     */
    private static final Executor asyncPool = useCommonPool ?
        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

    /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */
    static final class ThreadPerTaskExecutor implements Executor {
        public void execute(Runnable r) { new Thread(r).start(); }
    }

    /**
     * Null-checks user executor argument, and translates uses of
     * commonPool to asyncPool in case parallelism disabled.
     */
    static Executor screenExecutor(Executor e) {
        if (!useCommonPool && e == ForkJoinPool.commonPool()) {
            return asyncPool;
        }
        if (e == null) {
        	throw new NullPointerException();
        }
        return e;
    }
    
    public static <T> T simpleInvoke(Callable<T> callable, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
    	return simpleInvoke(asyncPool, callable, timeout, unit);
    }
    
    /**
     * 通过线程池获取操作返回的结果
     * @param executor
     * @param callable
     * @param timeout
     * @param unit
     * @return 返回的结果
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
    public static <T> T simpleInvoke(Executor executor, Callable<T> callable, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
    	T result = null;
    	
    	List<Thread> ts = new CopyOnWriteArrayList<Thread>();
    	
    	Callable<T> cb = () -> {
			Thread thread = Thread.currentThread();
    		ts.add(thread);
			T t = null;
    		try {
				t = callable.call();
				ts.remove(thread);
			} catch (Exception e) {
				ts.remove(thread);
				throw e;
			}
			return t;
    	};
    	
		FutureTask<T> futureTask = new FutureTask<T>(cb);
		executor.execute(futureTask);
		try {
			result = futureTask.get(timeout, unit);
		} catch (ExecutionException e) {
			// e.printStackTrace();
			throw e;
		} catch (InterruptedException | TimeoutException e) {
			// e.printStackTrace();
			if (ts.size() > 0) {
				Thread t = ts.get(0);
				if (t.isAlive()) {
					t.interrupt();
				}
			}
			futureTask.cancel(true);
			throw e;
		}
		
		return result;
    }
    
    public static <T> T simpleInvoke(Callable<T> callable, int times, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
    	return simpleInvoke(asyncPool, callable, times, timeout, unit);
    }
    
    /**
     * 通过线程池循环多次获取操作返回的结果，只返回其中一次返回的结果
     * @param executor
     * @param callable
     * @param times
     * @param timeout
     * @param unit
     * @return 第一次返回的结果
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
	public static <T> T simpleInvoke(Executor executor, Callable<T> callable, int times, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
		T result = null;
		
		int i = 1;
    	List<Thread> ts = new CopyOnWriteArrayList<Thread>();
    	
    	Callable<T> cb = () -> {
			Thread thread = Thread.currentThread();
    		ts.add(thread);
			T t = null;
    		try {
				t = callable.call();
				ts.remove(thread);
			} catch (Exception e) {
				ts.remove(thread);
				throw e;
			}
			return t;
    	};
    	
		for (; i <= times; i++) {
			if (i != 1) {
				if (Thread.interrupted()) {
					throw new InterruptedException();
				}
			}
			
			FutureTask<T> futureTask = new FutureTask<T>(cb);
			executor.execute(futureTask);
			try {
				result = futureTask.get(timeout, unit);
				break;
			} catch (ExecutionException e) {
				// e.printStackTrace();
				if (i == times) {
					throw e;
				}
			} catch (InterruptedException e) {
				// e.printStackTrace();
				if (ts.size() > 0) {
					Thread t = ts.get(0);
					if (t.isAlive()) {
						t.interrupt();
					}
				}
				futureTask.cancel(true);
				throw e;
			} catch (TimeoutException e) {
				// e.printStackTrace();
				if (ts.size() > 0) {
					Thread t = ts.get(0);
					if (t.isAlive()) {
						t.interrupt();
					}
				}
				futureTask.cancel(true);
				if (i == times) {
					throw e;
				}
			}
		}
		
		return result;
	}
	
	public static <T> T multipleInvoke(Callable<T> callable, int times, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
		return multipleInvoke(asyncPool, callable, times, timeout, unit, null, null);
	}
	
	public static <T> T multipleInvoke(Callable<T> callable, int times, long timeout, TimeUnit unit, Long lastTimeout) throws InterruptedException, ExecutionException, TimeoutException {
		return multipleInvoke(asyncPool, callable, times, timeout, unit, null, lastTimeout);
	}
	
	public static <T> T multipleInvoke(Callable<T> callable, int times, long timeout, TimeUnit unit, Integer lastTimes) throws InterruptedException, ExecutionException, TimeoutException {
		return multipleInvoke(asyncPool, callable, times, timeout, unit, lastTimes, null);
	}
	
	public static <T> T multipleInvoke(Callable<T> callable, int times, long timeout, TimeUnit unit, Integer lastTimes, Long lastTimeout) throws InterruptedException, ExecutionException, TimeoutException {
		return multipleInvoke(asyncPool, callable, times, timeout, unit, lastTimes, lastTimeout);
	}
	
	public static <T> T multipleInvoke(Executor executor, Callable<T> callable, int times, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
		return multipleInvoke(screenExecutor(executor), callable, times, timeout, unit, null, null);
	}
	
	public static <T> T multipleInvoke(Executor executor, Callable<T> callable, int times, long timeout, TimeUnit unit, Long lastTimeout) throws InterruptedException, ExecutionException, TimeoutException {
		return multipleInvoke(screenExecutor(executor), callable, times, timeout, unit, null, lastTimeout);
	}
	
	public static <T> T multipleInvoke(Executor executor, Callable<T> callable, int times, long timeout, TimeUnit unit, Integer lastTimes) throws InterruptedException, ExecutionException, TimeoutException {
		return multipleInvoke(screenExecutor(executor), callable, times, timeout, unit, lastTimes, null);
	}
	
	/**
	 * 通过线程池循环多次获取操作返回的结果，只返回其中一次返回的结果
	 * @param executor
	 * @param callable
	 * @param times
	 * @param timeout
	 * @param unit
	 * @param lastTimes
	 * @param lastTimeout
	 * @return 第一次返回的结果
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public static <T> T multipleInvoke(Executor executor, Callable<T> callable, int times, long timeout, TimeUnit unit, Integer lastTimes, Long lastTimeout) throws InterruptedException, ExecutionException, TimeoutException {
		T result = null;
		
		int i = 1;
		List<Thread> ts = new CopyOnWriteArrayList<Thread>();
		
		Callable<T> cb = () -> {
			Thread thread = Thread.currentThread();
    		ts.add(thread);
			T t = null;
    		try {
				t = callable.call();
				ts.remove(thread);
			} catch (Exception e) {
				ts.remove(thread);
				throw e;
			}
			return t;
    	};
    	
		for (; i <= times; i++) {
			if (i < times) {
				if (i != 1) {
					if (Thread.interrupted()) {
						throw new InterruptedException();
					}
				}
				
				FutureTask<T> futureTask = new FutureTask<T>(cb);
				executor.execute(futureTask);
				try {
					result = futureTask.get(timeout, unit);
					break;
				} catch (ExecutionException e) {
					// e.printStackTrace();
					if (i == times) {
						throw e;
					}
				} catch (InterruptedException e) {
					// e.printStackTrace();
					if (ts.size() > 0) {
						Thread t = ts.get(0);
						if (t.isAlive()) {
							t.interrupt();
						}
					}
					futureTask.cancel(true);
					throw e;
				} catch (TimeoutException e) {
					// e.printStackTrace();
					if (ts.size() > 0) {
						Thread t = ts.get(0);
						if (t.isAlive()) {
							t.interrupt();
						}
					}
					futureTask.cancel(true);
					if (i == times) {
						throw e;
					}
				}
			} else {
				if (Thread.interrupted()) {
					throw new InterruptedException();
				}
				
				if (lastTimeout != null && lastTimeout > 0) {
					timeout = lastTimeout;
				}
				if (lastTimes == null || lastTimes == 0) {
					FutureTask<T> futureTask = new FutureTask<T>(cb);
					executor.execute(futureTask);
					try {
						result = futureTask.get(timeout, unit);
						break;
					} catch (ExecutionException e) {
						// e.printStackTrace();
						if (i == times) {
							throw e;
						}
					} catch (InterruptedException e) {
						// e.printStackTrace();
						if (ts.size() > 0) {
							Thread t = ts.get(0);
							if (t.isAlive()) {
								t.interrupt();
							}
						}
						futureTask.cancel(true);
						throw e;
					} catch (TimeoutException e) {
						// e.printStackTrace();
						if (ts.size() > 0) {
							Thread t = ts.get(0);
							if (t.isAlive()) {
								t.interrupt();
							}
						}
						futureTask.cancel(true);
						if (i == times) {
							throw e;
						}
					}
				}
				
				// 最后一次，同时发出多个请求
				Callable<T>[] callables = (Callable<T>[])new Callable<?>[lastTimes];
				for (int j = 0; j < lastTimes; j++) {
					callables[j] = callable;
				}
				result = invokeMultiple(executor, timeout, unit, callables);
			}
		}
		//System.out.println("current times:" + i);
		
		return result;
	}
	
	public static <T> T invokeMultiple(long timeout, TimeUnit unit, Function<? super Object,? extends T>... functions) throws InterruptedException, ExecutionException, TimeoutException {
		return invokeMultiple(asyncPool, timeout, unit, functions);
	}
	
	/**
	 * 通过线程池同时获取多个不同操作返回的结果，只返回第一个返回的结果
	 * @param executor
	 * @param timeout
	 * @param unit
	 * @param functions
	 * @return 第一次返回的结果
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public static <T> T invokeMultiple(Executor executor, long timeout, TimeUnit unit, Function<? super Object,? extends T>... functions) throws InterruptedException, ExecutionException, TimeoutException {
		T result = null;
		
		int lastTimes = functions.length;
		List<Thread> ts = new CopyOnWriteArrayList<Thread>();
		CompletableFuture<?>[] futures = new CompletableFuture<?>[lastTimes];
		
		for (int j = 0; j < lastTimes; j++) {
			Function<? super Object, ? extends T> ifn = functions[j];
			Function<? super Object, ? extends T> fn = x -> {
				Thread thread = Thread.currentThread();
				ts.add(thread);
				T t = ifn.apply(null);
				ts.remove(thread);
				return t;
			};
			CompletableFuture<T> future = CompletableFuture.completedFuture(null).thenApplyAsync(fn, executor);
			futures[j] = future;
		}
		CompletableFuture<?> anyOf = CompletableFuture.anyOf(futures)
			.whenCompleteAsync((res, th) -> {
			/*if (th == null) {
				System.out.println("res:" + res);
			} else {
				System.out.println("err:" + th.getMessage());
			}*/
		}, executor);
		
		try {
			result = (T)anyOf.get(timeout, unit);
		} catch (InterruptedException | ExecutionException | TimeoutException e) {
			// e.printStackTrace();
			if (ts.size() > 0) {
				for (Thread t : ts) {
					if (t.isAlive()) {
						t.interrupt();
					}
				}
			}
			for (CompletableFuture<?> completableFuture : futures) {
				if(!completableFuture.isDone()) {
					completableFuture.cancel(true);
				}
			}
			throw e;
		}
		
		if (ts.size() > 0) {
			for (Thread t : ts) {
				if (t.isAlive()) {
					t.interrupt();
				}
			}
		}
		for (CompletableFuture<?> completableFuture : futures) {
			if(!completableFuture.isDone()) {
				completableFuture.cancel(true);
			}
		}
		
		return result;
	}
	
	public static <T> T supplyInvokeMultiple(long timeout, TimeUnit unit, Supplier<T>... suppliers) throws InterruptedException, ExecutionException, TimeoutException {
		return supplyInvokeMultiple(asyncPool, timeout, unit, suppliers);
	}
	
	/**
	 * 通过线程池同时获取多个不同操作返回的结果，只返回第一个返回的结果
	 * @param executor
	 * @param timeout
	 * @param unit
	 * @param suppliers
	 * @return 第一次返回的结果
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public static <T> T supplyInvokeMultiple(Executor executor, long timeout, TimeUnit unit, Supplier<T>... suppliers) throws InterruptedException, ExecutionException, TimeoutException {
		T result = null;
		
		int lastTimes = suppliers.length;
		List<Thread> ts = new CopyOnWriteArrayList<Thread>();
		CompletableFuture<?>[] futures = new CompletableFuture<?>[lastTimes];
		
		for (int j = 0; j < lastTimes; j++) {
			Supplier<T> isp = suppliers[j];
			Supplier<T> sp = () -> {
				Thread thread = Thread.currentThread();
				ts.add(thread);
				T t = isp.get();
				ts.remove(thread);
				return t;
			};
			CompletableFuture<T> future = CompletableFuture.supplyAsync(sp, executor);
			futures[j] = future;
		}
		CompletableFuture<?> anyOf = CompletableFuture.anyOf(futures)
			.whenCompleteAsync((res, th) -> {
			/*if (th == null) {
				System.out.println("res:" + res);
			} else {
				System.out.println("err:" + th.getMessage());
			}*/
		}, executor);
		
		try {
			result = (T)anyOf.get(timeout, unit);
		} catch (InterruptedException | ExecutionException | TimeoutException e) {
			// e.printStackTrace();
			if (ts.size() > 0) {
				for (Thread t : ts) {
					if (t.isAlive()) {
						t.interrupt();
					}
				}
			}
			for (CompletableFuture<?> completableFuture : futures) {
				if(!completableFuture.isDone()) {
					completableFuture.cancel(true);
				}
			}
			throw e;
		}
		
		if (ts.size() > 0) {
			for (Thread t : ts) {
				if (t.isAlive()) {
					t.interrupt();
				}
			}
		}
		for (CompletableFuture<?> completableFuture : futures) {
			if(!completableFuture.isDone()) {
				completableFuture.cancel(true);
			}
		}
		
		return result;
	}
	
	public static <T> T invokeMultiple(long timeout, TimeUnit unit, Callable<T>... callables) throws InterruptedException, ExecutionException, TimeoutException {
		return invokeMultiple(asyncPool, timeout, unit, callables);
	}
	
	/**
	 * 通过线程池同时获取多个不同操作返回的结果，只返回第一个返回的结果
	 * @param executor
	 * @param timeout
	 * @param unit
	 * @param callables
	 * @return T 第一次返回的结果
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public static <T> T invokeMultiple(Executor executor, long timeout, TimeUnit unit, Callable<T>... callables) throws InterruptedException, ExecutionException, TimeoutException {
		T result = null;
		
		int lastTimes = callables.length;
		List<Exception> es = new CopyOnWriteArrayList<Exception>();
		List<Thread> ts = new CopyOnWriteArrayList<Thread>();
		CompletableFuture<?>[] futures = new CompletableFuture<?>[lastTimes];
		
		for (int j = 0; j < lastTimes; j++) {
			Callable<T> callable = callables[j];
			Supplier<T> sp = () -> {
				Thread thread = Thread.currentThread();
				ts.add(thread);
				T t = null;
				try {
					t = callable.call();
					ts.remove(thread);
					es.clear();
				} catch (InterruptedException e) {
					//e.printStackTrace();
					if (ts.contains(thread)) {
						es.add(e);
						if (es.size() < lastTimes) {
							long millis = unit.toMillis(timeout);
							try {
								Thread.interrupted();
								thread.join(millis);
								thread.interrupt();
							} catch (InterruptedException e1) {
								//e1.printStackTrace();
								ts.remove(thread);
							}
						} else {
							ts.remove(thread);
						}
					}
				} catch (Exception e) {
					//e.printStackTrace();
					if (ts.contains(thread)) {
						es.add(e);
						if (es.size() < lastTimes) {
							long millis = unit.toMillis(timeout);
							try {
								if (thread.isInterrupted()) {
									Thread.interrupted();
								}
								thread.join(millis);
								thread.interrupt();
							} catch (InterruptedException e1) {
								//e1.printStackTrace();
								ts.remove(thread);
							}
						} else {
							ts.remove(thread);
						}
					}
				}
				return t;
			};
			//勿用，会导致多个任务不能同时并行执行
			//CompletableFuture<T> future = CompletableFuture.supplyAsync(sp, executor);
			CompletableFuture<T> future = CompletableFuture.supplyAsync(sp);
			futures[j] = future;
		}
		CompletableFuture<?> anyOf = CompletableFuture.anyOf(futures)
			.whenCompleteAsync((res, th) -> {
			/*if (th == null) {
				System.out.println("res:" + res);
			} else {
				System.out.println("err:" + th.getMessage());
			}*/
		}, executor);
		//});
		
		try {
			result = (T)anyOf.get(timeout, unit);
		} catch (InterruptedException | ExecutionException | TimeoutException e) {
			// e.printStackTrace();
			if (ts.size() > 0) {
				Iterator<Thread> tsi = ts.iterator();
				while (tsi.hasNext()) {
					Thread t = tsi.next();
					ts.remove(t);
					if (t.isAlive()) {
						t.interrupt();
					}
				}
			}
			for (CompletableFuture<?> completableFuture : futures) {
				if(!completableFuture.isDone()) {
					completableFuture.cancel(true);
				}
			}
			throw e;
		}
		
		if (ts.size() > 0) {
			Iterator<Thread> tsi = ts.iterator();
			while (tsi.hasNext()) {
				Thread t = tsi.next();
				ts.remove(t);
				if (t.isAlive()) {
					t.interrupt();
				}
			}
		}
		for (CompletableFuture<?> completableFuture : futures) {
			if(!completableFuture.isDone()) {
				completableFuture.cancel(true);
			}
		}
		if (result == null && es.size() > 0) {
			Exception e = es.get(es.size() - 1);
			if (e instanceof InterruptedException) {
				throw (InterruptedException)e;
			} else if (e instanceof TimeoutException) {
				throw (TimeoutException)e;
			} else {
				throw new ExecutionException(e);
			}
		}
		
		return result;
	}
	
	public static List<Object> invokeMultipleList(long timeout, TimeUnit unit, Callable<Object>... callables) throws InterruptedException, ExecutionException, TimeoutException {
		return invokeMultipleList(asyncPool, timeout, unit, true, null, callables);
	}
	
	public static List<Object> invokeMultipleList(Executor executor, long timeout, TimeUnit unit, Callable<Object>... callables) throws InterruptedException, ExecutionException, TimeoutException {
		return invokeMultipleList(executor, timeout, unit, true, null, callables);
	}
	
	public static List<Object> invokeMultipleList(long timeout, TimeUnit unit, boolean allowEmpty, Function<Object, Boolean> emptyFilter, Callable<Object>... callables) throws InterruptedException, ExecutionException, TimeoutException {
		return invokeMultipleList(asyncPool, timeout, unit, allowEmpty, emptyFilter, callables);
	}
	
	/**
	   * 通过线程池同时获取多个不同操作返回的结果
	   * 如果allowEmpty为true，则等待所有操作完成，返回所有操作的结果
	   * 如果allowEmpty为false，则判断emptyFilter是否包含所有完成操作的任一结果，如果都不包含，则返回所有操作的结果，否则抛出异常。
	 * @param executor
	 * @param timeout
	 * @param unit
	 * @param allowEmpty
	 * @param emptyFilter
	 * @param callables
	 * @return List<Object> 返回的结果集合
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public static List<Object> invokeMultipleList(Executor executor, long timeout, TimeUnit unit, boolean allowEmpty, Function<Object, Boolean> emptyFilter, Callable<Object>... callables) throws InterruptedException, ExecutionException, TimeoutException {
		long startTime = System.currentTimeMillis();
		
		int lastTimes = callables.length;
		List<Object> list = new ArrayList<Object>();
		List<Thread> ts = new CopyOnWriteArrayList<Thread>();
		List<FutureTask<Object>> futureTaskList = new ArrayList<FutureTask<Object>>(lastTimes);
		
		for (Callable<Object> callable : callables) {
			Callable<Object> cb = () -> {
				Thread thread = Thread.currentThread();
				ts.add(thread);
				Object t = null;
				try {
					t = callable.call();
					ts.remove(thread);
				} catch (Exception e) {
					ts.remove(thread);
					throw e;
				}
				return t;
	    	};
			FutureTask<Object> futureTask = new FutureTask<Object>(cb);
			executor.execute(futureTask);
			futureTaskList.add(futureTask);
		}
		
		FutureTask<Object> futureTask;
		long ntimeout;
		Object o;
		for (int i = 0; i < futureTaskList.size(); i++) {
			futureTask = futureTaskList.get(i);
			
			if (i == 0) {
				ntimeout = timeout;
			} else {
				if (Thread.interrupted()) {
					if (ts.size() > 0) {
						for (Thread t : ts) {
							if (t.isAlive()) {
								t.interrupt();
							}
						}
					}
					for (i = i + 1; i < futureTaskList.size(); i++) {
						futureTask = futureTaskList.get(i);
						if (!futureTask.isDone()) {
							futureTask.cancel(true);
						}
					}
					throw new InterruptedException();
				}
				
				long stopTime = System.currentTimeMillis();
				long time = stopTime - startTime;
				time = unit.convert(time, TimeUnit.MILLISECONDS);
				if (time >= timeout) {
					if (ts.size() > 0) {
						for (Thread t : ts) {
							if (t.isAlive()) {
								t.interrupt();
							}
						}
					}
					for (i = i + 1; i < futureTaskList.size(); i++) {
						futureTask = futureTaskList.get(i);
						if (!futureTask.isDone()) {
							futureTask.cancel(true);
						}
					}
					throw new TimeoutException();
				}
				
				ntimeout = timeout - time;
			}
			
			try {
				o = futureTask.get(ntimeout, unit);
			} catch (InterruptedException | ExecutionException | TimeoutException e) {
				// e.printStackTrace();
				if (ts.size() > 0) {
					for (Thread t : ts) {
						if (t.isAlive()) {
							t.interrupt();
						}
					}
				}
				futureTask.cancel(true);
				for (i = i + 1; i < futureTaskList.size(); i++) {
					futureTask = futureTaskList.get(i);
					if (!futureTask.isDone()) {
						futureTask.cancel(true);
					}
				}
				throw e;
			}
			
			if (allowEmpty || !emptyFilter.apply(o)) {
				list.add(o);
			} else {
				if (ts.size() > 0) {
					for (Thread t : ts) {
						if (t.isAlive()) {
							t.interrupt();
						}
					}
				}
				for (i = i + 1; i < futureTaskList.size(); i++) {
					futureTask = futureTaskList.get(i);
					if (!futureTask.isDone()) {
						futureTask.cancel(true);
					}
				}
				throw new ExecutionException(new RuntimeException("filter empty value failed, value:" + o));
			}
		}
		
		return list;
	}
	
	public static List<Object> invokeMultipleListLast(long timeout, TimeUnit unit, 
		Callable<Object> callable1, Callable<Object> callable2, Callable<Object> callable3) throws InterruptedException, ExecutionException, TimeoutException {
		return invokeMultipleListLast(asyncPool, timeout, unit, true, null, callable1, callable2, callable3);
	}
	
	public static List<Object> invokeMultipleListLast(Executor executor, long timeout, TimeUnit unit, 
		Callable<Object> callable1, Callable<Object> callable2, Callable<Object> callable3) throws InterruptedException, ExecutionException, TimeoutException {
		return invokeMultipleListLast(executor, timeout, unit, true, null, callable1, callable2, callable3);
	}
	
	public static List<Object> invokeMultipleListLast(long timeout, TimeUnit unit, boolean allowEmpty, Function<Object, Boolean> emptyFilter, 
		Callable<Object> callable1, Callable<Object> callable2, Callable<Object> callable3) throws InterruptedException, ExecutionException, TimeoutException {
		return invokeMultipleListLast(asyncPool, timeout, unit, allowEmpty, emptyFilter, callable1, callable2, callable3);
	}
	
	/**
	   * 通过线程池同时获取多个不同操作返回的结果
	   * 如果callable1操作成功，并且callable2或者callable3至少有一个操作成功，则返回callable1操作和callable2或者callable3其中一个操作成功的结果，否则抛出异常。
	 * @param executor
	 * @param timeout
	 * @param unit
	 * @param allowEmpty
	 * @param emptyFilter
	 * @param callable1
	 * @param callable2
	 * @param callable3
	 * @return
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TimeoutException
	 */
	public static List<Object> invokeMultipleListLast(Executor executor, long timeout, TimeUnit unit, boolean allowEmpty, Function<Object, Boolean> emptyFilter, 
		Callable<Object> callable1, Callable<Object> callable2, Callable<Object> callable3) throws InterruptedException, ExecutionException, TimeoutException {
		long startTime = System.currentTimeMillis();
		List<Object> list = new ArrayList<Object>();
		Object o1 = null;
		Object o2 = null;
		
    	List<Thread> ts = new CopyOnWriteArrayList<Thread>();
		
		Callable<Object> cb1 = () -> {
    		Thread thread = Thread.currentThread();
    		ts.add(thread);
    		Object t = null;
    		try {
				t = callable1.call();
				ts.remove(thread);
			} catch (Exception e) {
				ts.remove(thread);
				throw e;
			}
			return t;
    	};
    	Callable<Object> cb2 = () -> {
    		Thread thread = Thread.currentThread();
    		ts.add(thread);
    		Object t = null;
    		try {
				t = callable2.call();
				ts.remove(thread);
			} catch (Exception e) {
				ts.remove(thread);
				throw e;
			}
			return t;
    	};
    	Callable<Object> cb3 = () -> {
    		Thread thread = Thread.currentThread();
    		ts.add(thread);
    		Object t = null;
    		try {
				t = callable3.call();
				ts.remove(thread);
			} catch (Exception e) {
				ts.remove(thread);
				throw e;
			}
			return t;
    	};
    	
		FutureTask<Object> futureTask1 = new FutureTask<Object>(cb1);
		FutureTask<Object> futureTask2 = new FutureTask<Object>(cb2);
		executor.execute(futureTask1);
		executor.execute(futureTask2);
		try {
			o1 = futureTask1.get(timeout, unit);
		} catch (InterruptedException | ExecutionException | TimeoutException e) {
			// e.printStackTrace();
			if (ts.size() > 0) {
				for (Thread t : ts) {
					if (t.isAlive()) {
						t.interrupt();
					}
				}
			}
			futureTask1.cancel(true);
			if (!futureTask2.isDone()) {
				futureTask2.cancel(true);
			}
			throw e;
		}
		
		if (allowEmpty || !emptyFilter.apply(o1)) {
			if (Thread.interrupted()) {
				if (ts.size() > 0) {
					for (Thread t : ts) {
						if (t.isAlive()) {
							t.interrupt();
						}
					}
				}
				if (!futureTask2.isDone()) {
					futureTask2.cancel(true);
				}
				throw new InterruptedException();
			}
			
			long stopTime = System.currentTimeMillis();
			long time = stopTime - startTime;
			time = unit.convert(time, TimeUnit.MILLISECONDS);
			if (time >= timeout) {
				if (ts.size() > 0) {
					for (Thread t : ts) {
						if (t.isAlive()) {
							t.interrupt();
						}
					}
				}
				if (!futureTask2.isDone()) {
					futureTask2.cancel(true);
				}
				throw new TimeoutException();
			}
			
			if (futureTask2.isDone()) {
				try {
					o2 = futureTask2.get();
				} catch (InterruptedException | CancellationException e) {
					//e.printStackTrace();
					if (ts.size() > 0) {
						for (Thread t : ts) {
							if (t.isAlive()) {
								t.interrupt();
							}
						}
					}
					if (!futureTask2.isDone()) {
						futureTask2.cancel(true);
					}
					throw e;
				} catch (ExecutionException e) {
					//e.printStackTrace();
					if (ts.size() > 0) {
						for (Thread t : ts) {
							if (t.isAlive()) {
								t.interrupt();
							}
						}
					}
					if (!futureTask2.isDone()) {
						futureTask2.cancel(true);
					}
					
					long ntimeout = timeout - time;
					FutureTask<Object> futureTask3 = new FutureTask<Object>(cb3);
					executor.execute(futureTask3);
					try {
						o2 = futureTask3.get(ntimeout, unit);
					} catch (InterruptedException | ExecutionException | TimeoutException e1) {
						// e.printStackTrace();
						if (ts.size() > 0) {
							for (Thread t : ts) {
								if (t.isAlive()) {
									t.interrupt();
								}
							}
						}
						if (!futureTask3.isDone()) {
							futureTask3.cancel(true);
						}
						throw e1;
					}
				}
			} else {
				try {
					long ntimeout = timeout - time;
					o2 = invokeMultiple(executor, ntimeout, unit, 
						() -> futureTask2.get(ntimeout, unit), 
						cb3
					);
				} catch (InterruptedException | ExecutionException | TimeoutException e) {
					//e.printStackTrace();
					if (ts.size() > 0) {
						for (Thread t : ts) {
							if (t.isAlive()) {
								t.interrupt();
							}
						}
					}
					if (!futureTask2.isDone()) {
						futureTask2.cancel(true);
					}
					throw e;
				}
			}
    	} else {
			if (!futureTask2.isDone()) {
				futureTask2.cancel(true);
			}
			throw new ExecutionException(new RuntimeException("filter empty value failed, value:" + o1));
    	}
		if (ts.size() > 0) {
			for (Thread t : ts) {
				if (t.isAlive()) {
					t.interrupt();
				}
			}
		}
		list.add(o1);
		list.add(o2);
    	
		return list;
	}
	
	/**
	 * 判断参数是否为空
	 * @param o
	 * @return
	 */
	public static boolean isEmpty(Object o) {
		if (o == null) {
			return true;
		}

		if (o instanceof String) {
			return "".equals(o);
		} else if (o instanceof Byte) {
			return ((Byte)o) <= 0;
		} else if (o instanceof Short) {
			return ((Short)o) <= 0;
		} else if (o instanceof Integer) {
			return ((Integer)o) <= 0;
		} else if (o instanceof Long) {
			return ((Long)o) <= 0;
		} else if (o instanceof Float) {
			return ((Float)o) <= 0;
		} else if (o instanceof Double) {
			return ((Double)o) <= 0;
		} else if (o instanceof BigDecimal) {
			return ((BigDecimal)o).compareTo(BigDecimal.ZERO) <= 0;
		} else if (o instanceof String) {
			return "".equals(o);
		} else if (o instanceof Collection) {
			return ((Collection)o).size() == 0;
		} else if (o instanceof Map) {
			return ((Map)o).size() == 0;
		}
		return false;
	}
}