第七章 - 线程池应用

2021-05-04 13:29

阅读：671

标签：处理异步排序关闭 normal join() can add exception

线程池

自定义线程池

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import lombok.extern.slf4j.Slf4j;

/**
 * 自定义线程池
 */
@Slf4j(topic = "c.TestMyPool")
public class TestMyPool {
	public static void main(String[] args) {
		MyThreadPool pool = new MyThreadPool(2, 10, 1000, TimeUnit.MILLISECONDS, (queue, task) -> {
			while (!queue.tryPut(task, 1000, TimeUnit.MILLISECONDS)) {
				log.debug("拒绝策略重试添加任务到队列");
			}
		});
		for (int i = 0; i  log.debug("任务执行...{}", finalI));
		}
	}
}

/**
 * 拒绝策略
 *
 * @param 
 */
@FunctionalInterface
interface MyRejectPolicy {
	void reject(MyBlockingQueue queue, T task);
}

/**
 * 线程池
 */
class MyThreadPool {
	// 工作线程池
	private final HashSet workers = new HashSet();
	// 线程核心大小
	private long coreSize;
	// 阻塞队列
	private final MyBlockingQueue taskQueue;
	// 时间
	private long timeout;
	// 时间单位
	private final TimeUnit timeUnit;
	// 拒绝策略
	private final MyRejectPolicy rejectPolicy;
	
	public MyThreadPool(long coreSize, long capcity, long timeout, TimeUnit timeUnit, MyRejectPolicy rejectPolicy) {
		this.coreSize = coreSize;
		this.timeout = timeout;
		this.timeUnit = timeUnit;
		this.rejectPolicy = rejectPolicy;
		taskQueue = new MyBlockingQueue(capcity);
	}
	
	public void execute(Runnable task) {
		synchronized (workers) {
			if (workers.size() >= coreSize) {
				taskQueue.tryPut(this.rejectPolicy, task);
			}
			else {
				Worker worker = new Worker(task);
				workers.add(worker);
				worker.start();
			}
		}
	}
	
	class Worker extends Thread {
		private Runnable task;
		
		public Worker(Runnable task) {
			this.task = task;
		}
		
		@Override
		public void run() {
			// 这里的代码很关键，是个循环，而且第一次执行完毕新的任务后，旧的任务还能够设置到task中，再次运行
			while (null != task || (task = taskQueue.tryTask(timeout, timeUnit)) != null) {
				try {
					task.run();
				} finally {
					task = null;
				}
			}
			synchronized (workers) {
				workers.remove(this);
			}
		}
	}
}

/**
 * 阻塞任务
 */
class MyBlockingQueue {
	/**
	 * 字段:
	 * 1. 需要存访任务的队列
	 * 2. 队列大小
	 * 方法:
	 * 获取任务 删除任务
	 * 添加任务
	 * 获取队列大小
	 */
	/**
	 * 任务的队列
	 */
	private Deque queue = new ArrayDeque();
	
	/**
	 * 队列大小
	 */
	private long capcity;
	
	/**
	 * 锁
	 */
	private final ReentrantLock lock = new ReentrantLock();
	
	/**
	 * 队列空
	 */
	private Condition emptyWwaitSet = lock.newCondition();
	
	/**
	 * 队列满
	 */
	private Condition fullWwaitSet = lock.newCondition();
	
	public MyBlockingQueue(long capcity) {
		this.capcity = capcity;
	}
	
	/**
	 * 获取任务
	 *
	 * @return
	 */
	public T task() {
		try {
			lock.lock();
			while (queue.isEmpty()) {
				// 等待
				try {
					emptyWwaitSet.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			T t = queue.removeFirst();
			fullWwaitSet.signal();
			return t;
		} finally {
			lock.unlock();
		}
	}
	
	/**
	 * 尝试获取任务
	 *
	 * @param timeout
	 * @param unit
	 *
	 * @return
	 */
	public T tryTask(long timeout, TimeUnit unit) {
		long nanos = unit.toNanos(timeout);
		try {
			lock.lock();
			while (queue.isEmpty()) {
				// 等待
				try {
					if (nanos = this.capcity) {
				// 满了, 不需要生产了
				try {
					fullWwaitSet.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			queue.addLast(task);
			emptyWwaitSet.signal();
		} finally {
			lock.unlock();
		}
	}
	
	/**
	 * 带时间的任务添加
	 *
	 * @param task
	 * @param timeout
	 * @param unit
	 *
	 * @return
	 */
	public boolean tryPut(T task, long timeout, TimeUnit unit) {
		long nanos = unit.toNanos(timeout);
		try {
			lock.lock();
			while (queue.size() >= this.capcity) {
				// 满了, 不需要生产了
				try {
					if (nanos  rejectPolicy, T task) {
		try {
			lock.lock();
			if (queue.size() >= capcity) {
				rejectPolicy.reject(this, task);
			}
			else {
				queue.addLast(task);
				emptyWwaitSet.signal();
			}
		} finally {
			lock.unlock();
		}
	}
	
	/**
	 * 获取大小
	 *
	 * @return
	 */
	public long size() {
		try {
			lock.lock();
			return this.queue.size();
		} finally {
			lock.unlock();
		}
	}
	
}

扩展自定义链接池

@Slf4j(topic = "c.ConnectionDemo")
public class ConnectionDemo {
	public static void main(String[] args) {
		Pool.INSTANCE.init(2);
		for (int i = 0; i  {
				Connection connection = null;
				try {
					connection = Pool.INSTANCE.borrow();
				} finally {
					Pool.INSTANCE.free(connection);
				}
			}).start();
		}
	}
}
/**
 * 自定义线程池
 */
@Slf4j(topic = "c.Pool")
enum Pool {
	INSTANCE(10);
	
	private int poolSize;
	private Connection[] connections;
	private AtomicIntegerArray states;
	
	Pool(int poolSize) {
		init(poolSize);
	}
	
	/**
	 * 这个函数都是 new 所以是线程安全的, 它可能被无数次的new出来,但是最后一次new的值才是最终我们用的到的, 前面无数次的new最后都会被gc回收
	 * @param poolSize
	 */
	public void init(int poolSize) {
		this.poolSize = poolSize;
		states = new AtomicIntegerArray(new int[poolSize]);
		connections = new Connection[poolSize];
		for (int i = 0; i  opt = Optional.ofNullable(connection);
		for (int i = 0; i

ThreadPoolExecutor

线程池的组成主要是

核心线程 -- 核心线程是线程池遇到任务默认让核心线程执行(corePoolSize)

急救线程 -- 当核心线程数量不足, 我们救要使用上急救线程(maximumPoolSize - corePoolSize) 但是他存在keepAliveTime 保持时间, 如果急救线程空闲时间超过, 则线程消失

阻塞队列 -- 当任务多于maximumPoolSize , 线程池中的线程全忙 , 多出任务丢到阻塞队列

任务 -- 线程执行的任务

技术图片

线程池状态

ThreadPoolExecutor 使用 int 的高 3 位来表示线程池状态，低 29 位表示线程数量5

状态名	高 3 位	接收新任务	处理阻塞队列任务	说明
RUNNING	111	Y	Y
SHUTDOWN	000	N	Y	不会接收新任务，但会处理阻塞队列剩余任务
STOP	001	N	N	会中断正在执行的任务，并抛弃阻塞队列任务
TIDYING	010	-	-	任务全执行完毕，活动线程为 0 即将进入终结
TERMINATED	011	-	-	终结状态

从数字上比较，TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING
这些信息存储在一个原子变量 ctl 中，目的是将线程池状态与线程个数合二为一，这样就可以用一次 cas 原子操作进行赋值

class zhazha {
	public static void main(String[] args) {
	    // c 为旧值， ctlOf 返回结果为新值
        ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
        // rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们
        private static int ctlOf(int rs, int wc) { return rs | wc; }
	}
}

构造方法

class zhazha {
	public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler);
}

corePoolSize 核心线程数目 (最多保留的线程数)
maximumPoolSize 最大线程数目
keepAliveTime 生存时间 - 针对救急线程
unit 时间单位 - 针对救急线程
workQueue 阻塞队列
threadFactory 线程工厂 - 创建线程工厂类, 可以在类内部主动创建一个自己想要的线程属性, 比如如下几个属性

异常处理, 优先级, 是否守护者线程, 名字, 加载器

技术图片

handler 拒绝策略

工作方式：

线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。
当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue 队列排
队，直到有空闲的线程。
如果队列选择了有界队列，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线
程来救急。
如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它
著名框架也提供了实现

(1) AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略

(2) CallerRunsPolicy 让调用者运行任务

(3) DiscardPolicy 放弃本次任务

(4) DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之

(5) Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方便定位问题
(6) Netty 的实现，是创建一个新线程来执行任务

(7) ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略

(8) PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略
当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit 来控制。

根据这个构造方法，JDK Executors 类中提供了众多工厂方法来创建各种用途的线程池

固定线程池 -- FixedThreadPool

class zhazha {
	public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue());
    }
}

核心线程数 == 最大线程数 ==> 无急救线程 ===> 无超时时间

队列无界, 随意存放

评价

合适明确数量的耗时任务

急救线程池 -- CachedThreadPool

class zhazha {
	public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue());
    }
}

核心数 == 0 ==> 急救线程数无限创建(最大Integer.MAX_VALUE个线程同时存在)

队列无界

空闲保留时间60s

评价

合适任务多, 单任务耗时少

单例线程池-- SingleThreadExecutor

class zhazha {
	public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue()));
    }
}

永远保证线程池中只有一个线程, 如果意外终止了这个线程, 这会抛弃掉这个线程, 再创建一个新的线程

队列无界

使用包装设计模式, FinalizableDelegatedExecutorService, 可以保证这个ThreadPoolExecutor类内部的一些类似set的方法不会被特殊方法直接调用修改, 也无法被强转成ThreadPoolExecutor, 防止内部的方法暴露

任务调度线程池 -- ScheduledThreadPool

在『任务调度线程池』功能加入之前，可以使用 java.util.Timer 来实现定时功能，Timer 的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务。

Timer定时器的缺陷

class zhazha {
	/**
     * Timer的缺点: 一旦出现异常, 这无法再次使用
     */
    private static void func1() {
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                log.debug("task 1");
                Sleeper.sleep(2);
            }
        }, 1000);
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                log.debug("task 2");
            }
        }, 1000);
        log.debug("start ... ");
    }
}

调度器的使用

class zhazha {
	/**
     * 调度器使用
     */
    private static void func2() {
        ScheduledExecutorService service = new ScheduledThreadPoolExecutor(1);
        service.schedule(() -> {
            log.debug("task 1");
            int i = 1 / 0;
        }, 1000, TimeUnit.MILLISECONDS);
        service.schedule(() -> {
            log.debug("task 2");
        }, 1000, TimeUnit.MILLISECONDS);
    }
}

定时器周期执行任务

class zhazha {
	public static void main(String[] args) {
        ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(1);
        /**
         * 相隔 2 秒
         * 程序未执行完毕, 定时间早就开始定时
         *      时间可能按照周期算, 也可以按照程序执行时间算
         * 19:32:08 FixedRate running
         * 19:32:10 FixedRate running
         * 19:32:12 FixedRate running
         */
        threadPool.scheduleAtFixedRate(() -> {
            log.debug("FixedRate running");
            Sleeper.sleep(20_00);
        }, 1, 1, TimeUnit.SECONDS);
    
        /**
         * 相隔 3 秒, 这里相隔 2 + 1 == 3
         *      程序执行完毕, 定时器才开始定时
         * 19:32:52 FixedDelay running
         * 19:32:55 FixedDelay running
         * 19:32:58 FixedDelay running
         */
        // threadPool.scheduleWithFixedDelay(() -> {
        // 	log.debug("FixedDelay running");
        // 	Sleeper.sleep(2000);
        // }, 1, 1, TimeUnit.SECONDS);
    }
}

评价

整个线程池表现为：线程数固定，任务数多于线程数时，会放入无界队列排队。任务执行完毕，这些线程也不会被释放。用来执行延迟或反复执行的任务。

线程池正确处理执行任务异常

直接在代码内部捕捉异常

class zhazha {
	public static void main(String[] args) {
	    threadPool.schedule(() -> {
        try {
            log.debug("task 1");
            int i = 1 / 0;
        } catch (Exception e) {
            e.printStackTrace();
        }
    }, 1000, TimeUnit.MILLISECONDS);
	}
}

使用future的get方法获取异常

此方法适用于submit方法提交任务的方法

class zhazha {
    public static void main(String[] args) {
        	Future future = threadPool.submit(() -> {
            log.debug("task 1");
            int i = 1 / 0;
            return true;
        });
        try {
            if (future.get()) {
                System.out.println("true");
            }
            else {
                System.out.println("false");
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

自定义异常处理方案

此方法适用于线程池execute方法的使用，如果使用submit函数的话，是无效的

详解：submit和execute方法各区别

class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
		@Override
		public void uncaughtException(Thread t, Throwable e) {
			System.out.println("-----------execption-----------");
			System.out.println("线程信息: " + t.toString());
			System.out.println("异常信息: " + e.getMessage());
		}
	}

class Zhazha {
	public static void main(String[] args) {
	    ExecutorService threadPool = Executors.newCachedThreadPool(r -> {
            Thread thread = new Thread(r);
            thread.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
            return thread;
        });
        
        threadPool.execute(() -> {
            log.debug("zhazha");
            int i = 1 / 0; // 代码在这里就会报错
        });
	}
}

-----------execption-----------
线程信息: Thread[Thread-0,5,main]
异常信息: / by zero

不过上面这种方式优缺点，发现线程池默认的 defaultfactory 内部使用了很多的其他功能，不仅仅是setUncaughtExceptionHandler，还有线程名字等等

所以完整代码应该是这样

static class MyThreadFacorty implements ThreadFactory {
    private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    MyThreadFacorty() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" + POOL_NUMBER.getAndIncrement() + "-thread-";
    }

    @Override
    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);
        }
        t.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
        return t;
    }
}


static class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
    @Override
    public void uncaughtException(Thread t, Throwable e) {
        System.out.println("-----------execption-----------");
        System.out.println("线程信息: " + t.toString());
        System.out.println("异常信息: " + e.getMessage());
    }
}

class zhazha {
	public void test02() throws Exception {
        CountDownLatch latch = new CountDownLatch(1); // 这是测试方案 防止主线程直接就退出
        ExecutorService threadPool = Executors.newCachedThreadPool(new MyThreadFacorty());
        threadPool.execute(() -> {
            log.debug("zhazha");
            int i = 1 / 0;
            latch.countDown();
        });
        latch.await();
    }
}

submit和execute方法各区别

Future> submit(Runnable task);

void execute(Runnable command);

他们都是提交任务到线程池的方法，但是其中的区别还是有的，其中最重要的区别在于：

submit方法的异常如果没有使用返回值Future的get方法异常无法被捕获，即使使用前面的

thread.setUncaughtExceptionHandler(handler);

方案想要捕获异常也是无效的，上面这种方法只适用于execute方法当你想要捕获异常的时候使用

其中底层原理很简单

execute方法底层它把异常抛给线程池，只要我们使用thread.setUncaughtExceptionHandler设置自己的异常捕获机制就能够捕获出来

try {
    task.run();
    afterExecute(task, null);
} catch (Throwable ex) {
    afterExecute(task, ex);
    throw ex;
}

但是submit底层原理就不是这样了

try {
    result = c.call();
    ran = true;
} catch (Throwable ex) {
    result = null;
    ran = false;
    setException(ex);
}

他会把异常捕获放到它的outcome成员变量中

而要想知道是否出现异常必须使用Future的get方法体现出来

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s

其中的 report 方法就是把异常抛出的

 @SuppressWarnings("unchecked")
    private V report(int s) throws ExecutionException {
        // 把捕获的异常放入 x变量判断是否是异常， 这个 s 参数就是是否是异常的判断
        Object x = outcome;
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        // 抛出异常
        throw new ExecutionException((Throwable)x);
    }

提交任务

// 执行任务
void execute(Runnable command);
// 提交任务 task，用返回值 Future 获得任务执行结果
 Future submit(Callable task);
// 提交 tasks 中所有任务
 List> invokeAll(Collection extends Callable> tasks)
    throws InterruptedException;
// 提交 tasks 中所有任务，带超时时间
 List> invokeAll(Collection extends Callable> tasks,
                              long timeout, TimeUnit unit)
    throws InterruptedException;
// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消
 T invokeAny(Collection extends Callable> tasks)
    throws InterruptedException, ExecutionException;
// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间
 T invokeAny(Collection extends Callable> tasks,
 long timeout, TimeUnit unit)
 throws InterruptedException, ExecutionException, TimeoutException;

关闭线程池

shutdown

/*
线程池状态变为 SHUTDOWN
- 不会接收新任务
- 但已提交任务会执行完
- 此方法不会阻塞调用线程的执行
*/
void shutdown();

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        // 修改线程池状态
        advanceRunState(SHUTDOWN);
        // 仅会打断空闲线程
        interruptIdleWorkers();
        onShutdown(); // 扩展点 ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    // 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等)
    tryTerminate();
}

shutdownNow

/*
    线程池状态变为 STOP
    - 不会接收新任务
    - 会将队列中的任务返回
    - 并用 interrupt 的方式中断正在执行的任务
*/
List shutdownNow();

public List shutdownNow() {
     List tasks;
     final ReentrantLock mainLock = this.mainLock;
     mainLock.lock();
     try {
         checkShutdownAccess();
         // 修改线程池状态
         advanceRunState(STOP);
         // 打断所有线程
         interruptWorkers();
         // 获取队列中剩余任务
         tasks = drainQueue();
     } finally {
    	 mainLock.unlock();
     }
     // 尝试终结
     tryTerminate();
     return tasks;
}

其它方法

// 不在 RUNNING 状态的线程池，此方法就返回 true
boolean isShutdown();
// 线程池状态是否是 TERMINATED
boolean isTerminated();
// 调用 shutdown 后，由于调用线程并不会等待所有任务运行结束，因此如果它想在线程池 TERMINATED 后做些事
情，可以利用此方法等待
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

异步模式之工作线程

让有限的工作线程（Worker Thread）来轮流异步处理无限多的任务。也可以将其归类为分工模式，它的典型实现就是线程池，也体现了经典设计模式中的享元模式

例如，海底捞的服务员（线程），轮流处理每位客人的点餐（任务），如果为每位客人都配一名专属的服务员，那么成本就太高了（对比另一种多线程设计模式：Thread-Per-Message）
注意，不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率
例如，如果一个餐馆的工人既要招呼客人（任务类型A），又要到后厨做菜（任务类型B）显然效率不咋地，分成服务员（线程池A）与厨师（线程池B）更为合理，当然你能想到更细致的分工

线程池饥饿

固定大小线程池会有饥饿现象

例如: 饭店里只有两个人, 这两个人既能服务客人也能做饭, 而且每个人只能服务一个客人 , 突然店铺来个两个客人, 饭店两个人都去服务各自选中的客人, 现在没有人做饭了, 程序在这里无法再次运行下去了, 也就是饥饿了

@Slf4j(topic = "c.TestHungerDemo")
public class TestHungerDemo {
	
	private static final List MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅");
	private static final Random RANDOM = new Random();
	private static String cooking() {
		return MENU.get(RANDOM.nextInt(MENU.size()));
	}
	
	public static void main(String[] args) {
		ExecutorService pool = Executors.newFixedThreadPool(2);
		pool.execute(() -> {
			log.debug("处理点餐服务....");
			Future future = pool.submit(() -> {
				log.debug("做菜中....");
				return cooking();
			});
			try {
				log.debug("菜做完了. 上菜{}", future.get());
			} catch (InterruptedException | ExecutionException e) {
				e.printStackTrace();
			}
		});
		
		pool.execute(() -> {
			log.debug("处理点餐服务....");
			Future future = pool.submit(() -> {
				log.debug("做菜中....");
				return cooking();
			});
			try {
				log.debug("菜做完了. 上菜{}", future.get());
			} catch (InterruptedException | ExecutionException e) {
				e.printStackTrace();
			}
		});
		
		// 这句话是不可以的, 会报错, 拒绝运行错误, 说白了就是在执行 Future future = pool.submit(() -> {
		// 这一行的时候 pool.shutdown(); 已经执行完毕了, 这个时候再去执行 pool.submit 方法就会出现错误
		// pool.shutdown();
	}
	
}

15:07:17.930 [pool-1-thread-1] DEBUG c.TestHungerDemo - 处理点餐服务....
15:07:17.930 [pool-1-thread-2] DEBUG c.TestHungerDemo - 处理点餐服务....

上面的代码就存在饥饿问题, 没有线程去煮饭了

解决方案

不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率

ExecutorService waiterPool = Executors.newFixedThreadPool(1);
ExecutorService cookPool = Executors.newFixedThreadPool(1);

waiterPool.execute(() -> {
    log.debug("处理点餐服务....");
    Future future = cookPool.submit(() -> {
        log.debug("做菜中....");
        return cooking();
    });
    try {
        log.debug("菜做完了. 上菜{}", future.get());
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
});

waiterPool.execute(() -> {
    log.debug("处理点餐服务....");
    Future future = cookPool.submit(() -> {
        log.debug("做菜中....");
        return cooking();
    });
    try {
        log.debug("菜做完了. 上菜{}", future.get());
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
});

创建多少线程池合适

过小会导致程序不能充分地利用系统资源、容易导致饥饿
过大会导致更多的线程上下文切换，占用更多内存

CPU 密集型运算

通常采用 cpu 核数 + 1 能够实现最优的 CPU 利用率，+1 是保证当线程由于页缺失故障（操作系统）或其它原因导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费

I/O 密集型运算

CPU 不总是处于繁忙状态，例如，当你执行业务计算时，这时候会使用 CPU 资源，但当你执行 I/O 操作时、远程RPC 调用时，包括进行数据库操作时，这时候 CPU 就闲下来了，你可以利用多线程提高它的利用率。
经验公式如下

线程数 = 核数 * 期望 CPU 利用率 * 总时间(CPU计算时间+等待时间) / CPU 计算时间

例如 4 核 CPU 计算时间是 50% ，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式

4 * 100% * 100% / 50% = 8

例如 4 核 CPU 计算时间是 10% ，其它等待时间是 90%，期望 cpu 被 100% 利用，套用公式

4 * 100% * 100% / 10% = 40

例如 4 核 CPU 计算时间是 50% ，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式
4 * 100% * 100% / 50% = 8
例如 4 核 CPU 计算时间是 10% ，其它等待时间是 90%，期望 cpu 被 100% 利用，套用公式
4 * 100% * 100% / 10% = 40

tomcat线程池

技术图片

LimitLatch 用来限流，可以控制最大连接个数，类似 J.U.C 中的 Semaphore 后面再讲
Acceptor 只负责【接收新的 socket 连接】
Poller 只负责监听 socket channel 是否有【可读的 I/O 事件】
一旦可读，封装一个任务对象（socketProcessor），提交给 Executor 线程池处理
Executor 线程池中的工作线程最终负责【处理请求】

Tomcat 线程池扩展了 ThreadPoolExecutor，行为稍有不同

如果总线程数达到 maximumPoolSize

? 这时不会立刻抛 RejectedExecutionException 异常

? 而是再次尝试将任务放入队列，如果还失败，才抛出 RejectedExecutionException 异常

源码 tomcat-7.0.42

public void execute(Runnable command, long timeout, TimeUnit unit) {
    submittedCount.incrementAndGet();
    try {
        super.execute(command);
    } catch (RejectedExecutionException rx) {
        if (super.getQueue() instanceof TaskQueue) {
            final TaskQueue queue = (TaskQueue) super.getQueue();
            try {
                if (!queue.force(command, timeout, unit)) {
                    submittedCount.decrementAndGet();
                    throw new RejectedExecutionException("Queue capacity is full.");
                }
            } catch (InterruptedException x) {
                submittedCount.decrementAndGet();
                Thread.interrupted();
                throw new RejectedExecutionException(x);
            }
        }
        else {
            submittedCount.decrementAndGet();
            throw rx;
        }
    }
}

TaskQueue.java

public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
    if (parent.isShutdown())
        throw new RejectedExecutionException("Executor not running, can‘t force a command into the queue");
    return super.offer(o, timeout, unit); //forces the item onto the queue, to be used if the task 
    is rejected
}

Connector 配置

技术图片

Executor 线程配置

技术图片

jdk线程池源码分析

技术图片
线程池在Exectors中存在几个线程池方案，不过这些线程池方案都存在问题，ali开发手册上不让用
主要原因是：
CacheTheradPool和FiedThreadPool的阻塞方案不对劲，最大数量是Integer.Max_value数量级，容易出现溢出
还有他的拒绝阻塞策略也是不太完美的，需要我们重新设计

我们现在来分析源码

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue());
}

public ThreadPoolExecutor(int corePoolSize,
                      int maximumPoolSize,
                      long keepAliveTime,
                      TimeUnit unit,
                      BlockingQueue workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
     Executors.defaultThreadFactory(), defaultHandler);
}

从上面的源码来看，调用了个构造方法，传入了核心线程数，最大线程数，线程空闲等待时间，时间单位，工作等待队列，线程创建工厂类，队列满拒绝策略

其中创建线程工厂类方法主要就是创建线程，并给线程赋予名字和判断是否daemon线程
然后线程队列拒绝策略就是如果队列满了，还往队列中添加，这抛出异常消息
上面分析到这里就已经完成了，现在我们需要分析的师如何执行任务了

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    // 获取标志位（其中包含线程的几种状态和线程队列的数量）
    int c = ctl.get();
    // 判断数量是否小于核心线程
    if (workerCountOf(c)

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (int c = ctl.get();;) {
        // 检测队列是否为空
        if (runStateAtLeast(c, SHUTDOWN)
            && (runStateAtLeast(c, STOP)
                || firstTask != null
                || workQueue.isEmpty()))
            return false;
        for (;;) {
            // 如果 core 为 true 则 返回核心线程数 corePoolSize 判断工作线程数是否超出核心线程数, 超出，返回false，添加任务失败
            if (workerCountOf(c)
                >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
                return false;
            // 自增线程数量标记
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateAtLeast(c, SHUTDOWN))
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 创建Worker线程
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 上锁
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                int c = ctl.get();
                // 如果线程池正在运行
                if (isRunning(c) ||
                    (runStateLessThan(c, STOP) && firstTask == null)) {
                    // 线程正在运行
                    if (t.isAlive())
                        throw new IllegalThreadStateException();
                    // 添加线程到线程池中
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 直接运行
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // 添加失败，直接返回 false，（通过这里workerStarted一定为false）
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

public boolean offer(E e) {
    // 如果发现任务为空则抛出异常
    if (e == null) throw new NullPointerException();
    // 获取数量
    final AtomicInteger count = this.count;
    // 判断数量是否为极限值
    if (count.get() == capacity)
        return false;
    final int c;
    // 创建节点
    final Node node = new Node(e);
    // 可重入锁
    final ReentrantLock putLock = this.putLock;
    // 上锁
    putLock.lock();
    try {
        // 如果任务数量和总数相同，则无法添加
        if (count.get() == capacity)
            return false;
        // 添加节点
        enqueue(node);
        // 自增数量
        c = count.getAndIncrement();
        // 判断如果数量是否操作了总数
        if (c + 1

分3个步骤进行。

如果小于corePoolSize线程的运行数量，尝试用给定的命令启动一个新线程作为它的第一个任务。对addWorker的调用会原子化地检查runState和workerCount，因此，通过返回false来防止在不应该添加线程的情况下添加线程的错误警报。
如果一个任务可以成功排队，那么我们仍然需要重复检查是否应该添加一个线程（因为上次检查后已有的线程死了），或者是进入这个方法后池子关闭了。所以我们要重新检查状态，如果停止了，我们就重新检查状态，必要时回滚查询，如果没有，就启动一个新的线程。
如果我们无法排队任务，那么我们就尝试添加一个新的线程。如果失败了，我们知道我们已经关机或者饱和了，所以拒绝这个任务。

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 如果task不为空则执行，如果为空则从队列中获取任务再去执行
        while (task != null || (task = getTask()) != null) {
            w.lock();
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                // 执行前处理
                beforeExecute(wt, task);
                try {
                    task.run();
                    // 执行后处理
                    afterExecute(task, null);
                } catch (Throwable ex) {
                    // 错误执行后处理
                    afterExecute(task, ex);
                    throw ex;
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

fork/join任务拆分

Fork/Join 是 JDK 1.7 加入的新的线程池实现，它体现的是一种分治思想，适用于能够进行任务拆分的 cpu 密集型运算
所谓的任务拆分，是将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。跟递归相关的一些计算，如归并排序、斐波那契数列、都可以用分治思想进行求解
Fork/Join 在分治的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程来完成，进一步提升了运算效率
Fork/Join 默认会创建与 cpu 核心数大小相同的线程池

提交给 Fork/Join 线程池的任务需要继承 RecursiveTask（有返回值）或 RecursiveAction（没有返回值），例如下面定义了一个对 1~n 之间的整数求和的任务

@Slf4j(topic = "c.TestForkJoinDemo")
public class TestForkJoinDemo {
	public static void main(String[] args) {
		int num = 1000;
		// 简单的拆分
		forkjoinFunc(num);
		// 拆分优化版
		forkJoinFuncPro(num);
	}
	
	private static void forkJoinFuncPro(int num) {
		ForkJoinPool pool = new ForkJoinPool(4);
		System.out.println("统计：" + pool.invoke(new MyTaskPro(1, num)));
		System.out.println("线程池数量：" + pool.getPoolSize());
	}
	
	private static void forkjoinFunc(int num) {
		ForkJoinPool pool = new ForkJoinPool(4);
		System.out.println("统计：" + pool.invoke(new MyTask(num)));
		System.out.println("线程池数量：" + pool.getPoolSize());
	}
	
	/**
	 * 进阶优化版
	 */
	static class MyTaskPro extends RecursiveTask {
		
		private final int start;
		private final int end;
		
		public MyTaskPro(int start, int end) {
			this.start = start;
			this.end = end;
		}
		
		@Override
		public String toString() {
			return start + "," + end;
		}
		
		@Override
		protected Integer compute() {
			if (start == end) {
				// log.debug("start == end {}", start);
				return start;
			}
			if ((end - start) == 1) {
				// log.debug("end - start == 1");
				return end + start;
			}
			int mid = start + (end - start) / 2;
			MyTaskPro t1 = new MyTaskPro(start, mid);
			MyTaskPro t2 = new MyTaskPro(mid + 1, end);
			t1.fork();
			t2.fork();
			// log.debug("fork() t1 = {} , t2 = {}", t1, t2);
			int res = t1.join() + t2.join();
			// log.debug("join() {} + {} = {}", t1, t2, res);
			return res;
		}
	}
	
	/**
	 * 这个方案容易出现栈溢出
	 */
	static class MyTask extends RecursiveTask {
		
		private final int n;
		
		public MyTask(int n) {
			this.n = n;
		}
		
		@Override
		public String toString() {
			return "{" + n + ‘}‘;
		}
		
		@Override
		protected Integer compute() {
			if (n == 1) {
				return n;
			}
			MyTask myTask = new MyTask(n - 1);
			myTask.fork();
			// log.debug("fork() {} + {}", n, myTask);
			Integer res = n + myTask.join();
			// log.debug("join() res = {}", res);
			return res;
		}
	}
}

容易出现栈溢出的版本，整个过程

技术图片

优化进阶版本

技术图片

第七章 - 线程池应用

标签：处理异步排序关闭 normal join() can add exception

原文地址：https://www.cnblogs.com/bangiao/p/13195693.html

上一篇：第二章 - Java与协程

下一篇：2020/6/11 JavaScript高级程序设计 DOM

文章来自：搜素材网的编程语言模块，转载请注明文章出处。
文章标题：第七章 - 线程池应用
文章链接：http://soscw.com/index.php/essay/82287.html

亲，登录后才可以留言！

第七章 - 线程池应用

线程池

自定义线程池

扩展自定义链接池

ThreadPoolExecutor

线程池状态

构造方法

固定线程池 -- FixedThreadPool

评价

急救线程池 -- CachedThreadPool

评价

单例线程池-- SingleThreadExecutor

任务调度线程池 -- ScheduledThreadPool

Timer定时器的缺陷

调度器的使用

定时器周期执行任务

评价

线程池正确处理执行任务异常

直接在代码内部捕捉异常

使用future的get方法获取异常

自定义异常处理方案

submit和execute方法各区别

提交任务

关闭线程池

shutdown

shutdownNow

其它方法

异步模式之工作线程

线程池饥饿

解决方案

创建多少线程池合适

CPU 密集型运算

I/O 密集型运算

tomcat线程池

jdk线程池源码分析

fork/join任务拆分

评论

热门文章

推荐文章

最新文章

置顶文章