并发编程之常用概念类和框架（一）

public static Unsafe getUnsafe() {
    Class var0 = Reflection.getCallerClass();
    if(!VM.isSystemDomainLoader(var0.getClassLoader())) {
        throw new SecurityException("Unsafe");
    } else {
        return theUnsafe;
    }
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 *  如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange，当两个线程都到达同步点时，
 *  这两个线程就可以交换数据，将本线程生产出来的数据传递给对方
 * @author lzhcode
 *
 */
public class TestExchanger {
  public static void main(String[] args) {
    ExecutorService service = Executors.newCachedThreadPool();
    final Exchanger exchanger = new Exchanger();
    service.execute(new Runnable(){
      public void run() {
        try {       
          String data1 = "zxx";
          System.out.println("线程" + Thread.currentThread().getName() + 
          "正在把数据" + data1 +"换出去");
          Thread.sleep((long)(Math.random()*10000));
          String data2 = (String)exchanger.exchange(data1);
          System.out.println("线程" + Thread.currentThread().getName() + 
          "换回的数据为" + data2);
        }catch(Exception e){
        }
      } 
    });
    service.execute(new Runnable(){
      public void run() {
        try {       
          String data1 = "lhm";
          System.out.println("线程" + Thread.currentThread().getName() + 
          "正在把数据" + data1 +"换出去");
          Thread.sleep((long)(Math.random()*10000));          
          String data2 = (String)exchanger.exchange(data1);
          System.out.println("线程" + Thread.currentThread().getName() + 
          "换回的数据为" + data2);
        }catch(Exception e){
        }       
      } 
    });   
  }
}

// 1. 创建线程池
ExecutorService executorService = new ThreadPoolExecutor(3, 5, 60,
      TimeUnit.SECONDS,
      new LinkedBlockingDeque<Runnable>(10), new DefaultThreadFactory("biz-exec"),
      new ThreadPoolExecutor.CallerRunsPolicy());
// 2. 创建执行任务
List<Future<Object>> futureList = new ArrayList<>();
for(Object arg : list) {
        futureList.add(executorService.submit(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
              // xxx
            }
        }));
}
// 3. 结果获取
for(Future f: futureList) {
    Object obj = f.get();
}

public class CountTask extends RecursiveTask<Integer> {
    private int start;
    private int end;
    private static final int THRED_HOLD = 30;
    public CountTask(int start, int end) {
        this.start = start;
        this.end = end;
    }
    @Override
    protected Integer compute() {
        int sum = 0;
        boolean canCompute = (end - start) <= THRED_HOLD;
        if (canCompute) { // 不需要拆分
            for (int i = start; i <= end; i++) {
                sum += i;
            }
            System.out.println("thread: " + Thread.currentThread() + " start: " + start + " end: " + end);
        } else {
            int mid = (end + start) / 2;
            CountTask left = new CountTask(start, mid);
            CountTask right = new CountTask(mid + 1, end);
            left.fork();
            right.fork();
            sum = left.join() + right.join();
        }
        return sum;
    }
}

@Test
public void testFork() throws ExecutionException, InterruptedException {
    int start = 0;
    int end = 200;
    CountTask task = new CountTask(start, end);
    ForkJoinPool pool = ForkJoinPool.commonPool();
    Future<Integer> ans = pool.submit(task);
    int sum = ans.get();
    System.out.println(sum);
}

public class TestObjWait {
    public static void main(String[] args)throws Exception {
        ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(1000);
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5,5,1000, TimeUnit.SECONDS,queue);
        Future<String> future = poolExecutor.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                TimeUnit.SECONDS.sleep(5);
                return "hello";
            }
        });
        String result = future.get();
        System.out.println(result);
    }
}

@Bean(destroyMethod="shutdown")
    public ThreadPoolTaskExecutor defaultThreadPool() {
        // CPU可用核心数
        int cpuNum = Runtime.getRuntime().availableProcessors();
        //1.IO密集型 估算线程池大小
        //线程数 = CPU可用核心数/(1-阻塞系数)
        //计算密集型任务的阻塞系数为0，而IO密集型任务的阻塞系数则接近于1。一个完全阻塞的任务是注
        //定要挂掉的，所以我们无须担心阻塞系数会达到1。
        //阻塞系数可以采用一些性能分析工具或java.lang.managenment API来确定线程话在系统I/O操作
        //上的时间与CPU密集任务所消耗的时间比值。
        //2.计算密集型估算线程池大小threadNum = cpuNum +1 或则  计算密集型估算线程池大小threadNum = cpuNum * 2;
        int threadNum = cpuNum * 2;
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程数目
        executor.setCorePoolSize(threadNum-1);
        //指定最大线程数
        executor.setMaxPoolSize(threadNum);
        //队列中最大的数目
        executor.setQueueCapacity(300);
        executor.setThreadFactory(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                //如果为守护线程，恢复为常规
                if(t.isDaemon()) {
                    t.setDaemon(false);
                }
                //如果有设置线程优先级，恢复为常规
                if(Thread.NORM_PRIORITY != t.getPriority()) {
                    t.setPriority(Thread.NORM_PRIORITY);
                }
                return t;
            }
        });
        //线程名称前缀
        executor.setThreadNamePrefix("defaultThreadPool_");
        //rejection-policy：当pool已经达到max size的时候，如何处理新任务
        //CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
        //对拒绝task的处理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //线程空闲后的最大存活时间
        executor.setKeepAliveSeconds(60);
        //加载
        executor.initialize();
        return executor;
    }