Java Concurrency in Practice

文章目录

1. 1、线程池的数量
2. 2、任务独立与工作队列界限
3. 3、单线程的Executor可能发生死锁
4. 4、一种线程安全的缓存计算工具
5. 5、一种This指针逃逸
6. 6、使用Semaphore控制任务提交速度
7. 7、修改标准工厂创建的executor
8. 8、Amdahl定律
9. 9、增加系统的伸缩性
10. 10、一种非阻塞的计数器

1、线程池的数量

$$N_{threads}=N_{cpu}*U_{cpu}*(1+\frac{W}{C})$$

U: 目标CPU使用率 [0,1]
W: wait time
C: compute time

2、任务独立与工作队列界限

• 任务独立时，设置线程池的工作队列界限才合理，如果任务之间存在依赖性，则可能导致线程“饥饿死锁”，应使用无界线程池，如newCachedThreadPool

  3、单线程的Executor可能发生死锁

• 单线程的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、一种线程安全的缓存计算工具

/**
* 一种带有缓存的计算工具
 * 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定律

• 串行部分越小加速比越大（单核运行时间/多核运行时间）$$Speedup=\frac{1}{F+\frac{(1-F)}{N}}$$

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;
    }
}