private volatile Singleton singleton = null; public static Singleton getInstance() { if (singleton == null) { synchronized (Singleton.class) { if (singleton == null) { singleton = new Singleton(); } } } return singleton; }

public class Singleton { private static class LazyHolder { private static final Singleton INSTANCE = new Singleton(); } private Singleton (){} public static final Singleton getInstance() { return LazyHolder.INSTANCE; }}

public class Singleton1 { private Singleton1() {} private static final Singleton1 single = new Singleton1(); public static Singleton1 getInstance() { return single; }}

public class FutureDemo1 {
 public static void main(String[] args) throws InterruptedException, ExecutionException { FutureTask<String> future = new FutureTask<String>(new Callable<String>() { @Override public String call() throws Exception { return new RealData().costTime(); } }); ExecutorService service = Executors.newCachedThreadPool(); service.submit(future);
 System.out.println("RealData方法調(diào)用完畢"); // 模擬主函數(shù)中其他耗時(shí)操作 doOtherThing(); // 獲取RealData方法的結(jié)果 System.out.println(future.get()); }
 private static void doOtherThing() throws InterruptedException { Thread.sleep(2000L); }}
class RealData {
 public String costTime() { try { // 模擬RealData耗時(shí)操作 Thread.sleep(1000L); return "result"; } catch (InterruptedException e) { e.printStackTrace(); } return "exception"; }
}

public class FutureDemo2 {
 public static void main(String[] args) throws InterruptedException, ExecutionException { ExecutorService service = Executors.newCachedThreadPool(); Future<String> future = service.submit(new RealData2());
 System.out.println("RealData2方法調(diào)用完畢"); // 模擬主函數(shù)中其他耗時(shí)操作 doOtherThing(); // 獲取RealData2方法的結(jié)果 System.out.println(future.get()); }
 private static void doOtherThing() throws InterruptedException { Thread.sleep(2000L); }}
class RealData2 implements Callable<String>{
 public String costTime() { try { // 模擬RealData耗時(shí)操作 Thread.sleep(1000L); return "result"; } catch (InterruptedException e) { e.printStackTrace(); } return "exception"; }
 @Override public String call() throws Exception { return costTime(); }}

// 取消任務(wù) boolean cancel(boolean mayInterruptIfRunning); // 是否已經(jīng)取消 boolean isCancelled(); // 是否已經(jīng)完成 boolean isDone(); // 取得返回對(duì)象 V get() throws InterruptedException, ExecutionException; // 取得返回對(duì)象, 并可以設(shè)置超時(shí)時(shí)間 V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

while(isRunning) { Thread.sleep(r.nextInt(SLEEP_TIME)); data = new PCData(count.incrementAndGet); // 構(gòu)造任務(wù)數(shù)據(jù) System.out.println(data + " is put into queue"); if (!queue.offer(data, 2, TimeUnit.SECONDS)) { // 將數(shù)據(jù)放入隊(duì)列緩沖區(qū)中 System.out.println("faild to put data : " + data); } }

while (true) { PCData data = queue.take(); // 提取任務(wù) if (data != null) { // 獲取數(shù)據(jù), 執(zhí)行計(jì)算操作 int re = data.getData() * 10; System.out.println("after cal, value is : " + re); Thread.sleep(r.nextInt(SLEEP_TIME));           }       }

public class MasterDemo { // 盛裝任務(wù)的集合 private ConcurrentLinkedQueue<TaskDemo> workQueue = new ConcurrentLinkedQueue<TaskDemo>(); // 所有worker private HashMap<String, Thread> workers = new HashMap<>(); // 每一個(gè)worker并行執(zhí)行任務(wù)的結(jié)果 private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<>();
 public MasterDemo(WorkerDemo worker, int workerCount) { // 每個(gè)worker對(duì)象都需要持有queue的引用, 用于領(lǐng)任務(wù)與提交結(jié)果 worker.setResultMap(resultMap); worker.setWorkQueue(workQueue); for (int i = 0; i < workerCount; i++) { workers.put("子節(jié)點(diǎn): " + i, new Thread(worker)); } }
 // 提交任務(wù) public void submit(TaskDemo task) { workQueue.add(task); }
 // 啟動(dòng)所有的子任務(wù) public void execute(){ for (Map.Entry<String, Thread> entry : workers.entrySet()) { entry.getValue().start(); } }
 // 判斷所有的任務(wù)是否執(zhí)行結(jié)束 public boolean isComplete() { for (Map.Entry<String, Thread> entry : workers.entrySet()) { if (entry.getValue().getState() != Thread.State.TERMINATED) { return false; } }
 return true; }
 // 獲取最終匯總的結(jié)果 public int getResult() { int result = 0; for (Map.Entry<String, Object> entry : resultMap.entrySet()) { result += Integer.parseInt(entry.getValue().toString()); }
 return result; }
}

public class WorkerDemo implements Runnable{
 private ConcurrentLinkedQueue<TaskDemo> workQueue; private ConcurrentHashMap<String, Object> resultMap;
 @Override public void run() { while (true) { TaskDemo input = this.workQueue.poll(); // 所有任務(wù)已經(jīng)執(zhí)行完畢 if (input == null) { break; } // 模擬對(duì)task進(jìn)行處理, 返回結(jié)果 int result = input.getPrice(); this.resultMap.put(input.getId() + "", result); System.out.println("任務(wù)執(zhí)行完畢, 當(dāng)前線程: " + Thread.currentThread().getName()); } }
 public ConcurrentLinkedQueue<TaskDemo> getWorkQueue() { return workQueue; }
 public void setWorkQueue(ConcurrentLinkedQueue<TaskDemo> workQueue) { this.workQueue = workQueue; }
 public ConcurrentHashMap<String, Object> getResultMap() { return resultMap; }
 public void setResultMap(ConcurrentHashMap<String, Object> resultMap) { this.resultMap = resultMap; }}

public class TaskDemo {
 private int id; private String name; private int price;
 public int getId() { return id; }
 public void setId(int id) { this.id = id; }
 public String getName() { return name; }
 public void setName(String name) { this.name = name; }
 public int getPrice() { return price; }
 public void setPrice(int price) { this.price = price; }}

MasterDemo master = new MasterDemo(new WorkerDemo(), 10); for (int i = 0; i < 100; i++) { TaskDemo task = new TaskDemo(); task.setId(i); task.setName("任務(wù)" + i); task.setPrice(new Random().nextInt(10000)); master.submit(task); }
 master.execute();
 while (true) { if (master.isComplete()) { System.out.println("執(zhí)行的結(jié)果為: " + master.getResult()); break; } }

public class CountTask extends RecursiveTask<Long>{ // 任務(wù)分解的閾值 private static final int THRESHOLD = 10000; private long start; private long end;

 public CountTask(long start, long end) { this.start = start; this.end = end; }
 public Long compute() { long sum = 0; boolean canCompute = (end - start) < THRESHOLD; if (canCompute) { for (long i = start; i <= end; i++) { sum += i; } } else { // 分成100個(gè)小任務(wù) long step = (start + end) / 100; ArrayList<CountTask> subTasks = new ArrayList<CountTask>(); long pos = start; for (int i = 0; i < 100; i++) { long lastOne = pos + step; if (lastOne > end) { lastOne = end; } CountTask subTask = new CountTask(pos, lastOne); pos += step + 1; // 將子任務(wù)推向線程池 subTasks.add(subTask); subTask.fork(); }
 for (CountTask task : subTasks) { // 對(duì)結(jié)果進(jìn)行join sum += task.join(); } } return sum; }
 public static void main(String[] args) throws ExecutionException, InterruptedException { ForkJoinPool pool = new ForkJoinPool(); // 累加求和 0 -> 20000000L CountTask task = new CountTask(0, 20000000L); ForkJoinTask<Long> result = pool.submit(task); System.out.println("sum result : " + result.get()); }}