1.场景线程池使用discardoldestpolicy拒绝策略，阻塞队列使用arrayblockingqueue,发现在某些情形下对于得到的future，调用get()方法当前线程会一直阻塞。
为了便于理解，将实际情景抽象为下面的代码：
threadpoolexecutor threadpoolexecutor = new threadpoolexecutor( 1, 1, 1, timeunit.seconds, new arrayblockingqueue<>(1), executors.defaultthreadfactory(), new threadpoolexecutor.discardoldestpolicy());//新建线程池时核心线程数及最大线程数都设置为1，阻塞队列使用arrayblockingqueue，拒绝策略为discardoldestpolicypublic void dobusiness(){ task task1 = new task(); task task2 = new task(); task task3 = new task(); future<boolean> future1 = threadpoolexecutor.submit(task1);//当前工作线程为0，会新建一个worker作为工作线程，并执行task1 future<boolean> future2 = threadpoolexecutor.submit(task2);//当前核心线程数已满，会将任务放入阻塞队列 future<boolean> future3 = threadpoolexecutor.submit(task3); /*当前核心线程已满并且阻塞队列已满，execute()时会调用threadpoolexecutord的addworker(command,false)，由 于目前task1还没执行完，则工作线程数量为1，已经达到了最大线程数，则addworker(command,false)返回false， 触发对应的拒绝策略，会从阻塞队列中移除task2对应的任务(阻塞队列中并不是直接放的task2，而是以task2为入 参构造的一个futuretask，参见abstarctexecutorservice的submit(callable<t> task)方法*/ try{ boolean result = future2.get(); system.out.println(result); } catch (executionexception e) { e.printstacktrace(); } catch (interruptedexception e) { e.printstacktrace(); }}@testpublic void test_dobusiness(){ dobusiness();//入口}private class task implements callable<boolean>{ @override public boolean call() throws exception { try { thread.sleep(1000);//模拟业务执行 return true; }catch(exception e){ e.printstacktrace(); } return true; }}
2. 原因分析通过上面代码我们明白了阻塞队列会将task2对应的任务移除，那么为何移除之后调用get()方法线程会一直阻塞呢？
其实future future2= threadpoolexecutor.submit(task2)实际会调用abstractexecutorservice的submit(callable task)方法，并且最终返回的future2实际是一个futuretask类型。
public <t> future<t> submit(callable<t> task) { if (task == null) throw new nullpointerexception(); runnablefuture<t> ftask = newtaskfor(task); execute(ftask); return ftask;}
protected <t> runnablefuture<t> newtaskfor(callable<t> callable) { return new futuretask<t>(callable);}
因此，我们直接看futuretask的get()方法
public v get() throws interruptedexception, executionexception { int s = state; if (s <= completing) s = awaitdone(false, 0l); return report(s);}
由于future2已经从阻塞队列中移除，并且从始至终都没有工作线程执行它，即futuretask的状态一直都为new状态，其会进入awaitdone(false,0l)中，接下列我们追踪该方法。
private int awaitdone(boolean timed, long nanos) throws interruptedexception { final long deadline = timed ? system.nanotime() + nanos : 0l; waitnode q = null; boolean queued = false; for (;;) { if (thread.interrupted()) { removewaiter(q); throw new interruptedexception(); } int s = state; if (s > completing) { if (q != null) q.thread = null; return s; } else if (s == completing) // cannot time out yet thread.yield(); else if (q == null)//第一次进for循环时q==null，进入到该分支 q = new waitnode(); else if (!queued)//第二次进for循环时queue为false，则使用cas将q置为waiters的头结点 queued = unsafe.compareandswapobject(this, waitersoffset, q.next = waiters, q); else if (timed) { nanos = deadline - system.nanotime(); if (nanos <= 0l) { removewaiter(q); return state; } locksupport.parknanos(this, nanos); } else//将q置为头结点后，最终会进入这里调用park()方法，阻塞当前线程 locksupport.park(this); }
从上面的代码可以看出调用future2.get()后会一直阻塞在park()方法处，这便是本次问题出现的原因，
3.总结本次问题出现主要是同时满足了以下几点：
1)使用了有界的阻塞队列arrayblockingqueue
2)工作线程达到了线程池配置的最大线程数
3)拒绝策略使用了discardoldestpolicy(使用discardpolicy也会出现这个问题)
4.思考我们日常使用线程池提交任务后，如果在任务执行完成之前调用future的get()方法，当前线程会进入阻塞状态，当任务执行完成后，才会将当前线程唤醒，如何从代码上分析该流程？
首先当任务提交到线程池，如果任务当前在阻塞队列中，则futuretask的状态依然像上面的情况一样，是处于new状态，调用get()方法依然会到达locksupport.park(this)处，将当前线程阻塞。什么时候才会将当前线程唤醒了？
那就是当存在工作线程worker目前分配的任务执行完成后，其会去调用worker类的gettask()方法从阻塞队列中拿到该任务，并执行该任务的run()方法，下面是futuretask的run()方法
public void run() { if (state != new || !unsafe.compareandswapobject(this, runneroffset, null, thread.currentthread())) return; try { callable<v> c = callable; if (c != null && state == new) { v result; boolean ran; try { result = c.call(); ran = true; } catch (throwable ex) { result = null; ran = false; setexception(ex); } if (ran) set(result);//如果任务执行成功，则调用set(v result)方法 } } finally { // runner must be non-null until state is settled to // prevent concurrent calls to run() runner = null; // state must be re-read after nulling runner to prevent // leaked interrupts int s = state; if (s >= interrupting) handlepossiblecancellationinterrupt(s); }}
其会在执行成功后，调用set(v result)方法
protected void set(v v) { if (unsafe.compareandswapint(this, stateoffset, new, completing)) { outcome = v; unsafe.putorderedint(this, stateoffset, normal); // final state finishcompletion();// }}
然后将futuretask状态置为normal(futuretask的状态要和threadpoolexecutor的状态区分开),接着调用finishcompletion()方法
private void finishcompletion() { // assert state > completing; for (waitnode q; (q = waiters) != null;) { if (unsafe.compareandswapobject(this, waitersoffset, q, null)) { for (;;) { thread t = q.thread;//q在await()方法中设置的，其值为调用get()方法的线程 if (t != null) { q.thread = null; locksupport.unpark(t);//唤醒该线程 } waitnode next = q.next; if (next == null) break; q.next = null; // unlink to help gc q = next; } break; } } done();//熟悉的钩子方法 callable = null; // to reduce footprint}
在finishcompletion中唤起因get()而阻塞的线程。
以上就是java threadpoolexecutor线程池拒绝策略实例分析的详细内容。