1、线程池的数量
2、任务独立时，设置线程池的工作队列界限才合理，如果任务之间存在依赖性，则可能导致线程“饥饿死锁”，应使用无界线程池，如newCachedThreadPool
3、单线程的Executor可能发生死锁，newSingleThreadExecutor 对象同时执行父任务与子任务
4、一种带有缓存的计算工具，线程安全的
5、一种This指针逃逸
6、使用Semaphore控制任务提交速度
7、修改标准工厂创建的executor
8、Amdahl定律 串行部分越小加速比越大（单核运行时间/多核运行时间）
9、增加系统的伸缩性（增加计算资源时，吞吐量和处理能力相应增加）
10、一种非阻塞的计数器
1、线程池的数量
U: 目标CPU使用率 [0,1]
W: wait time
C: compute time
2、任务独立时，设置线程池的工作队列界限才合理，如果任务之间存在依赖性，则可能导致线程“饥饿死锁”，应使用无界线程池，如newCachedThreadPool
3、单线程的Executor可能发生死锁，newSingleThreadExecutor 对象同时执行父任务与子任务
/**
* Created with IntelliJ IDEA.
* User: pingansheng
* Date: 2016/6/6
* Time: 15:43
*/
public class SingleThreadExecutorDeadLock {
ExecutorService exec = Executors.newSingleThreadExecutor() ;
class MyCall implements Callable<String> {
@Override public String call () throws Exception {
Future<String> f1 = exec.submit( new MyThread()) ;
Future<String> f2 = exec.submit(new MyThread()) ;
System. out.println(" 任务提交结束，等待两个子任务返回 ");
// 主线程在单线程线程池中，所以 f1与f2 均在等待队列中，永远无法执行。
return f1.get() + f2.get();
}
}
class MyThread implements Callable<String> {
@Override public String call () throws Exception {
System.out.println( "子任务结束") ;
return "RES";
}
}
public static void main(String[] args) throws Exception {
SingleThreadExecutorDeadLock lock = new SingleThreadExecutorDeadLock() ;
Future<String> f3 = lock.exec .submit(lock.new MyCall()) ;
try {
System.out.println(f3.get()) ;
} finally {
lock.exec.shutdown() ;
}
}
}
4、一种带有缓存的计算工具，线程安全的
```java
/**
* 一种带有缓存的计算工具
* Created with IntelliJ IDEA.
* User: pingansheng
* Date: 2016/6/1
* Time: 14:29
*/
public class CachedCompute {
interface Computable<A, V> {
V compute(A args) throws InterruptedException , ExecutionException;
}
class CacheComputer<A, V> implements Computable< A, V > {
// 缓存
private final Map<A, Future<V>> cache = new ConcurrentHashMap<>() ;
private Computable< A, V > computer;
public CacheComputer(Computable< A, V > c) {
this .computer = c ;
}
@Override public V compute (A args) throws InterruptedException , ExecutionException {
Future<V> f = cache.get(args);
if (null == f) {
Callable<V> callable = new Callable< V>() {
@Override public V call() throws Exception {
return computer .compute(args) ;
}
};
FutureTask<V > ft = new FutureTask< V>(callable);
f = cache .putIfAbsent(args, ft) ;
if (null == f) {
System. out.println(" 缓存放置成功 ");
f = ft;
ft.run();
}
}
try {
return f.get() ;
} catch (CancellationException e) {
cache.remove(args , f);
} catch (ExecutionException e) {
throw e ;
}
return null;
}
}
CacheComputer cache = new CacheComputer<String , String>( new Computable<String, String>() {
@Override public String compute (String args)
throws InterruptedException , ExecutionException {
return "计算结果";
}
});
public static void main (String[] args) throws Throwable {
CachedCompute compute = new CachedCompute();
System. out.println(compute.cache .compute("key")) ;
}
}
5、一种This指针逃逸
/**
* Created with IntelliJ IDEA.
* User: pingansheng
* Date: 2016/5/11
* Time: 17:08
*/
public class ThisEscape {
private String name;
public ThisEscape() throws Throwable{
new Thread(new EscapeRunnable()).start() ;
Thread. sleep( 1000);
name ="123";
}
private class EscapeRunnable implements Runnable {
@Override
public void run() {
// 通过ThisEscape.this就可以引用外围类对象 , 但是此时外围类对象可能还没有构造完成 , 即发生了外围类的this引用的逃逸，构造函数未完成之前不应该暴露this指针
System. out.println(ThisEscape.this. name); //可能会出现令人疑惑的错误 name=null
}
}
public static void main(String[] args) throws Throwable{
new ThisEscape();
}
}
6、使用Semaphore控制任务提交速度
/**
* 使用 semaphore控制任务的提交速度
* Created with IntelliJ IDEA.
* User: pingansheng
* Date: 2016/6/7
* Time: 10:24
*/
public class BoundedExecutor {
private final Executor executor;
private final Semaphore semaphore;
public BoundedExecutor(Executor exe , int bound) {
this .executor = exe ;
this. semaphore = new Semaphore(bound);
}
public void submitTask(final Runnable command) throws InterruptedException {
semaphore .acquire();
System. out.println(" 信号量获取成功，当前剩余数： " + semaphore .availablePermits()) ;
try {
executor .execute(new Runnable() {
@Override public void run() {
try {
command.run();
} finally {
semaphore.release();
}
}
});
} catch (RejectedExecutionException e) {
semaphore .release();
}
}
public static void main(String[] args) throws Exception {
ExecutorService es=Executors.newCachedThreadPool() ;
BoundedExecutor exe = new BoundedExecutor(es , 100) ;
for ( int i = 0 ; i < 50 ; i++) {
exe.submitTask(new Runnable() {
@Override public void run() {
try {
System. out.println(" 任务执行 ");
Thread.sleep(1000) ;
} catch (Throwable e) {
}
}
});
}
System.out.println( "提交50 个任务结束 ");
es.shutdown() ;
}
}
7、修改标准工厂创建的executor
/**
* 强制类型转换重新设置 Executor的线程池参数
* newSingleThreadExecutor 除外，不是线程工厂直接创建，而是通过包装类
* public static ExecutorService newSingleThreadExecutor() {
* return new FinalizableDelegatedExecutorService
* (new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,
* new LinkedBlockingQueue<Runnable>()));
* }
* Created with IntelliJ IDEA.
* User: pingansheng
* Date: 2016/6/7
* Time: 10:24
*/
public class ExecutorForceSet {
private static final ExecutorService executor = Executors.newFixedThreadPool( 1);
// 使用此方法包装后可以避免被修改
// private static final ExecutorService executor =Executors.unconfigurableExecutorService(Executors.newFixedThreadPool(1));
public static void main(String[] args) throws Exception {
if (executor instanceof ThreadPoolExecutor) {
((ThreadPoolExecutor) executor ).setCorePoolSize(100) ;
((ThreadPoolExecutor) executor ).setMaximumPoolSize( 100);
} else {
System.out.println( "转换出错，非线程工厂创建 ");
}
for (int i = 0; i < 50; i++) {
//lambda
executor.execute(() -> {
try {
System. out.println(" 任务执行 ");
Thread.sleep (2000 );
} catch (Throwable e) {
}
});
}
System.out.println( "提交50 个任务结束 ");
executor.shutdown() ;
}
}
8、Amdahl定律 串行部分越小加速比越大（单核运行时间/多核运行时间）
F: 必须串行部分的比例
N: CPU个数
9、增加系统的伸缩性（增加计算资源时，吞吐量和处理能力相应增加）
- 缩小锁的范围synchronized方法变为synchronized代码块（锁真正关键的地方）
- 缩小锁的粒度synchronized方法（锁对象，static锁Class对象）变为synchronized代码块（锁变量或对象）
- 锁分段：多个锁保护不同的区域，如10个对象数组保护10个数据片段（每片10个），通过取余获取相应的锁，(key.hashCode % length) % lockSize
10、一种非阻塞的计数器
/**
* CasCounter
* <p/>
* Nonblocking counter using CAS
*
* @author Brian Goetz and Tim Peierls
*/
@ThreadSafe
public class CasCounter {
private SimulatedCAS value;
public int getValue() {
return value.get();
}
public int increment() {
int v;
do {
v = value.get();
} while (v != value.compareAndSwap(v, v + 1));
//非阻塞一般使用底层的并发原语操作
return v + 1;
}
}
使用java.util.concurrent.atomic包中的AtomicInteger实现
public class CasCounter {
private AtomicInteger value=new AtomicInteger(0);
public int getValue(){
return value.get();
}
public int increment(){
return value.incrementAndGet();
}
}
incrementAndGet源码
public final int incrementAndGet() {
return unsafe.getAndAddInt(this, valueOffset, 1) + 1; //getAndAddInt返回的是旧值
}
其中unsafe是sun.misc.Unsafe类，该类实现了CAS原子性操作。
比较并替换，并返回旧的值
public final int getAndAddInt(Object var1, long var2, int var4) {
int var5;
do {
var5 = this.getIntVolatile(var1, var2);
} while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4)); //比较并进行替换，非阻塞式
return var5; //返回旧的值
}
其中compareAndSwapInt是native方法。
