【Java并发】JAVA并发编程实战-读书笔记14

无论何时，线程池需要创建一个线程都要通过一个线程工厂。

public interface ThreadFactory{

Thread newThread(Runnable r);

}

默认的线程工厂创建一个新的非守护的线程，其中的newThread会在创建一个新线程时被调用。

public class MyThreadFactory implements ThreadFactory{

private final String poolName;

public MyThreadFactory(String poolName){

this.poolName=poolName;

}

public Thread newThread(Runnable runnable){

return new MyAppThread(runnable,poolName);

}

}

上面的例子自定义了一个线程工厂，实例化一个新的线程并传入池的名称。

public class MyAppThread extends Thread{

public static final String DEFAULT_NAME=”MyAppThread”;

private static volatile boolean debugLifecycle=false;

private static final AtomicInteger created=new AtomicInteger();

private static final AtomicInteger alive=new AtomicInteger();

private static final Logger log=Logger.getAnonymousLogger();

public MyAppThread(Runnable r){

this(r,DEFAULT_NAME);

}

public MyAppThread(Runnable r,String name){

super(runnable,name+”-”+created.incrementAndGet());

setUncaughtExceptionHandler(

new Thread.UncaughtExceptionHandler(){

public void uncaughtException(Thread t,Throwable e){

log.log(Level.SEVERE,”UNCAUGHT in thread”+t.getName,e);

}

}

);

}

public void run(){

boolean debug=debugLifecycle;

if(debug){

log.log(Level.FINE,”Created”+getName());

}

try{

alive.incrementAndGet();

super.run();

}finally{

alive.decrementAndGet();

if(debug){

log.log(Level.FINE,”Exiting”+getName());

}

}

}

public static int getThreadCreated(){

return created.get();

}

public static int getThreadAlive(){

return alive.get():

}

public static boolean getDebug(){

return debugLifecycle;

}

public static void setDebug(boolean b){

debugLifecycle=b;

}

}

ThreadPoolExecutor的设计是可扩展的，他提供了几个钩子让子类覆盖——beforeExecute、afterExecute和terminated。
执行任务的线程会调用钩子函数beforeExecute和afterExecute。无论任务时正常从run中返回还是抛出一个异常，afterExecute都会被调用。如果任务完成后抛出一个error则afterExecute不会被调用。如果beforeExecute抛出一个RuntimeException，任务将不被执行，afterExecute也不会被调用。

terminated钩子会在线程池完成关闭动作后调用，也就是当所有任务都已完成并且所有工作者线程也已经关闭后会执行terminated。可以用来释放Executor在生命周期里分配到的资源，还可以发出通知、记录日志或者完成统计信息。

public class TimingThreadPool extends ThreadPoolExecutor{

private final ThreadLocal<Long> startTime=new ThreadLocal<Long>();

private final Logger log=Logger.getLogger(“TimingThreadPool”);

private final AtomicLong numTasks=new AtomicLong();

private final AtomicLong totalTime=new AtomicLong();

protected void beforeExecute(Thread t,Runnable t){

super.beforeExecute(t,r);

log.fine(String.formate(“Thread %s:start %s”,t,r));

startTime.set(System.nanoTime());

}

pro
bf92
tected void afterExecute(Runnable r,Throwable t){

try{

long endTime=System.nanoTime();

long taskTime=endTime-startTime.get();

numTasks.incrementAndGet();

totalTime.addAndGet(taskTime);

log.fine(String.formate(“Thread %s:end %s,time=%dns”,t,r,taskTime));

}finally{

super.afterExecute(r,t);

}

}

protected void terminated(){

try{

log.info(String.format(“Terminated:avg time=%dns”,totalTime.get()/numTasks.get()));

}finally{

super.terminated();

}

}

}

把顺序递归转换为并行递归

public<T> void sequentialRecursive(List<Node<T>> nodes,Collection<T> results){

for(Node<T> n:nodes){

results.add(n.compute());

sequentialRecursive(n.getChildren(),results);

}

}

public<T> void parallelRecursive(final Executor exec,List<Node<T>> nodes,final Collection<T> results){

for(final Node<T> n:nodes){

exec.execute(new Runnable(){

public void run(){

results.add(n.compute());

}

});

prarllelRecursive(exec,n.getChildren(),results);

}

}

public<T> Collection<T> getParallelResults(List<Node<T>> nodes) throws InterruptedException{

ExecutorService exec=Executors.newCachedThreadPool();

Queue<T> resultQueue=new ConcurrentLinkedQueue<T>();

parallelRecursive(exec,nodes,resultQueue);

exec.shutdown();

exec.awaitTermination(Long.MAX_VALUE,TimeUnit.SECONDS);

return resultQueue;

}

当parallelRecursive返回的时候，树上的每个节点都已经被访问了，遍历的过程依然是顺序的，只是对compute的调用才是并行执行的。

public interface Puzzle<P,M>{

P initialPosition();

boolean isGoal(P position);

Set<M> legalMoves(P position);

P move(P position,M move)

}

static class Node<P,M>{

final P pos;

final M move;

final Node<P,M> prev;

Node(P pos,M move,Node<P,M> prev){...}

List<M> asMoveList(){

List<M> solution=new LinkedList<M>();

for(Node<P,M> n=this;n.move!=null;n=n.prev){

solution.add(0,n.move);

}

return solution;

}

}

顺序化解决问题

public class SequentialPuzzleSolver<P,M>{

private final Puzzle<P,M> puzzle;

private final Set<P> seen=new HashSet<P>();

public SequentialPuzzleSolver(Puzzle<P,M> puzzle){

this.puzzle=puzzle;

}

public List<M> solve(){

P pos=puzzle.initialPosition();

return search(new Node<P,M>(pos,null,null));

}

private List<M> search(Node<P,M> node){

if(!seen.contains(node.pos)){

seen.add(node.pos);

if(puzzle.isGoal(node.pos)){

return node.asMoveList();

}

for(M move:puzzle.legalMoves(node.pos)){

P pos=puzzle.move(node.pos,move);

Node<P,M> child=new Node<P,M>(pos,move,node);

List<M> result=search(child);

if(result!=null){

return result;

}

}

}

return null;

}

static class Node<P,M>{...}

}

并发版的谜题解决者

public class ConcurrentPuzzleSolver<P,M>{

private final Puzzle<P,M> puzzle;

private final ExecutorService exec;

private final ConcurrentMap<P,Boolean> seen;

final ValueLatch<Node<P,M>> solution=new ValueLatch<Node<P,M>>();

public List<M> solve() throws InterruptedException{

try{

P p=puzzle.initialPosition();

exec.execute(newTask(p,null,null));

Node<P,M> solnNode=solution.getValue();

return (solnNode==null)?null:solnNode.asMoveList();

}finally{

exec.shutdown();

}

}

protected Runnable newTask(P p,M m,Node<P,M> n){

return new SolverTask(p,m,n);

}

class SolverTask extends Node<P,M> implements Runnable{

...

public void run(){

if(solution.isSet()||seen.putIfAbsent(pos,true)!=null){

return;

}

if(puzzle.isGoal(pos)){

solution.setValue(this);

}else{

for(M m:puzzle.legalMoves(pose)){

exec.execute(newTask(puzzle.move(pos,m),m,this));

}

}

}

}

}

使用可携带结果的闭锁

public class ValueLatch<T>{

private T value=null;

private final CountDownLatch done=new CountDownLatch(1);

public boolean isSet(){

return (done.getCount()==0);

}

public synchronized void setValue(T newValue){

if(!isSet()){

value=newValue;

done.countDown();

}

}

public T getValue()throws InterruptedException{

done.await();

synchronized(this){

return value;

}

}

}

并发的Solver还不能很好地处理不存在任何方案的情况。

public class PuzzleSolver<P,M> extends ConcurrentPuzzleSolver<P,M>{

private final AtomicInteger taskCount=new AtomicInteger(0);

protected Runnable newTask(P p,M m,Node<P,M> n){

return new CountingSolverTask(p,m,n);

}

class CountingSolverTask extends SolverTask{

CountingSolverTask(P pos,M move,Node<P,M> prev){

super(pos,move,prev);

taskCount.incrementAndGet();

}

public void run(){

try{

super.run();

}finally{

if(taskCount.decrementAndGet()==0){

solution.setValue(null);

}

}

}

}

}