一、任务和执行策略之间的隐性耦合
Executor可以将任务的提交和任务的执行策略解耦
只有任务是同类型的且执行时间差别不大，才能发挥最大性能，否则，如将一些耗时长的任务和耗时短的任务放在一个线程池，除非线程池很大，否则会造成死锁等问题
1.线程饥饿死锁
类似于：将两个任务提交给一个单线程池，且两个任务之间相互依赖，一个任务等待另一个任务，则会发生死锁；表现为池不够
定义：某个任务必须等待池中其他任务的运行结果，有可能发生饥饿死锁
2.线程池大小
注意：线程池的大小还受其他的限制，如其他资源池：数据库连接池
如果每个任务都是一个连接，那么线程池的大小就受制于数据库连接池的大小
3.配置ThreadPoolExecutor线程池
实例：
1.通过Executors的工厂方法返回默认的一些实现
2.通过实例化ThreadPoolExecutor（.....）自定义实现
线程池的队列
1.无界队列：任务到达，线程池饱满，则任务在队列中等待，如果任务无限达到，则队列会无限扩张
如：单例和固定大小的线程池用的就是此种
2.有界队列：如果新任务到达，队列满则使用饱和策略
3.同步移交：如果线程池很大，将任务放入队列后在移交就会产生延时，如果任务生产者很快也会导致任务排队
SynchronousQueue直接将任务移交给工作线程
机制：将一个任务放入，必须有一个线程等待接受，如果没有，则新增线程，如果线程饱和，则拒绝任务
如：CacheThreadPool就是使用的这种策略
饱和策略：
setRejectedExecutionHandler来修改饱和策略
1.终止Abort（默认）：抛出异常由调用者处理
2.抛弃Discard
3.抛弃DiscardOldest：抛弃最旧的任务，注意：如果是优先级队列将抛弃优先级最高的任务
4.CallerRuns：回退任务，有调用者线程自行处理
4.线程工厂ThreadFactoy
每当创建线程时：其实是调用了线程工厂来完成
自定义线程工厂：implements ThreadFactory
可以定制该线程工厂的行为：如UncaughtExceptionHandler等
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
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 runnable, 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() { // Copy debug flag to ensure consistent value throughout. 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 getThreadsCreated() { return created.get(); } public static int getThreadsAlive() { return alive.get(); } public static boolean getDebug() { return debugLifecycle; } public static void setDebug(boolean b) { debugLifecycle = b; }} |
5.扩展ThreadPoolExecutor
可以被自定义子类覆盖的方法：
1.afterExecute：结束后，如果抛出RuntimeException则方法不会执行
2.beforeExecute：开始前，如果抛出RuntimeException则任务不会执行
3.terminated：在线程池关闭时，可以用来释放资源等
二、递归算法的并行化
1.循环
在循环中，每次循环操作都是独立的
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
//串行化 void processSequentially(List<Element> elements) { for (Element e : elements) process(e); } //并行化 void processInParallel(Executor exec, List<Element> elements) { for (final Element e : elements) exec.execute(new Runnable() { public void run() { process(e); } }); } |
2.迭代
如果每个迭代操作是彼此独立的，则可以串行执行
如：深度优先搜索算法；注意：递归还是串行的，但是，每个节点的计算是并行的
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
//串行 计算compute 和串行迭代 public <T> void sequentialRecursive(List<Node<T>> nodes, Collection<T> results) { for (Node<T> n : nodes) { results.add(n.compute()); sequentialRecursive(n.getChildren(), results); } } //并行 计算compute 和串行迭代 public <T> void parallelRecursive(final Executor exec, List<Node<T>> nodes, final Collection<T> results) { for (final Node<T> n : nodes) { exec.execute(() -> results.add(n.compute())); parallelRecursive(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; } |
实例：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
|
public class ConcurrentPuzzleSolver <P, M> { private final Puzzle<P, M> puzzle; private final ExecutorService exec; private final ConcurrentMap<P, Boolean> seen; protected final ValueLatch<PuzzleNode<P, M>> solution = new ValueLatch<PuzzleNode<P, M>>(); public ConcurrentPuzzleSolver(Puzzle<P, M> puzzle) { this.puzzle = puzzle; this.exec = initThreadPool(); this.seen = new ConcurrentHashMap<P, Boolean>(); if (exec instanceof ThreadPoolExecutor) { ThreadPoolExecutor tpe = (ThreadPoolExecutor) exec; tpe.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy()); } } private ExecutorService initThreadPool() { return Executors.newCachedThreadPool(); } public List<M> solve() throws InterruptedException { try { P p = puzzle.initialPosition(); exec.execute(newTask(p, null, null)); // 等待ValueLatch中闭锁解开，则表示已经找到答案 PuzzleNode<P, M> solnPuzzleNode = solution.getValue(); return (solnPuzzleNode == null) ? null : solnPuzzleNode.asMoveList(); } finally { exec.shutdown();//最终主线程关闭线程池 } } protected Runnable newTask(P p, M m, PuzzleNode<P, M> n) { return new SolverTask(p, m, n); } protected class SolverTask extends PuzzleNode<P, M> implements Runnable { SolverTask(P pos, M move, PuzzleNode<P, M> prev) { super(pos, move, prev); } public void run() { //如果有一个线程找到了答案，则return，通过ValueLatch中isSet CountDownlatch闭锁实现； //为类避免死锁，将已经扫描的节点放入set集合中，避免继续扫描产生死循环 if (solution.isSet() || seen.putIfAbsent(pos, true) != null){ return; // already solved or seen this position } if (puzzle.isGoal(pos)) { solution.setValue(this); } else { for (M m : puzzle.legalMoves(pos)) exec.execute(newTask(puzzle.move(pos, m), m, this)); } } }} |
以上这篇java并发编程_线程池的使用方法(详解)就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持服务器之家。
