vlambda博客
学习文章列表

class MyThreadPool{  //利用阻塞队列实现生产者-消费者模式  BlockingQueue<Runnable> workQueue;  //保存内部工作线程  List<WorkerThread> threads = new ArrayList<>(); // 构造方法 MyThreadPool(int poolSize, BlockingQueue<Runnable> workQueue) { this.workQueue = workQueue; // 创建工作线程 for(int idx=0; idx<poolSize; idx++){ WorkerThread work = new WorkerThread(); work.start(); threads.add(work); }  } // 提交任务 void execute(Runnable command){ workQueue.put(command); } // 工作线程负责消费任务，并执行任务 class WorkerThread extends Thread{ public void run() { //循环取任务并执行 while(true){ Runnable task = workQueue.take(); task.run(); } } }}/** 下面是使用示例 **/// 创建有界阻塞队列BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(2);// 创建线程池MyThreadPool pool = new MyThreadPool(10, workQueue);// 提交任务pool.execute(()->{ System.out.println("hello");});

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) { }

 public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); /** * 线程数少于corePoolSize, 直接新建 */ if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } /** * 线程数>=corePoolSize，尝试加入阻塞队列 */ if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } /** * 阻塞队列满, 尝试新建非核心线程,失败(线程数已达最大线程数)则执行拒绝策略 */ else if (!addWorker(command, false)) reject(command); }

private boolean addWorker(Runnable firstTask, boolean core) { ... Worker w = null; try { /** 包装成Worker */ w = new Worker(firstTask); /** Thread t 是任务所绑定的系统线程 */ final Thread t = w.thread; if (t != null) { /** 对任务添加前需加锁 */ final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //  int rs = runStateOf(ctl.get());
 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive())  throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { /** 启动任务 */ t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } ...}

 /** Worker类实现了Runnable接口, 具体执行的方法体 */ final void runWorker(Worker w) { /** 当前线程wt: w所绑定的线程 */ Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { w.lock(); // 检查线程池状态，如果关闭，则中断wt if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { /** 任务具体的执行内容 */ task.run(); } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }

// 提交Runnable任务Future submit(Runnable task);// 提交Callable任务 Future submit(Callable task);// 提交Runnable任务及结果引用 Future submit(Runnable task, T result);

// 取消任务boolean cancel(boolean mayInterruptIfRunning);// 判断任务是否已取消 boolean isCancelled();// 判断任务是否已结束boolean isDone();// 获得任务执行结果V get();// 获得任务执行结果，支持超时V get(long timeout, TimeUnit unit);

/ 创建FutureTaskFutureTask futureTask = new FutureTask<>(()-> 1+2);// 创建线程池ExecutorService es = Executors.newCachedThreadPool();// 提交FutureTask es.submit(futureTask);// 获取计算结果Integer result = futureTask.get();

import java.util.concurrent.Callable;import java.util.concurrent.FutureTask;import java.util.concurrent.TimeUnit;public class ThreadPoolCooking { // T1Task需要执行的任务： // 洗电饭锅，洗米，烧饭，吃饭 static class T1Task implements Callable {
 FutureTask ft2; // T1任务需要T2任务的FutureTask T1Task(FutureTask<String> ft2) { this.ft2 = ft2; } public String call() throws Exception { System.out.println("T1:洗电饭锅..."); TimeUnit.SECONDS.sleep(1); System.out.println("T1:洗米..."); TimeUnit.SECONDS.sleep(1); System.out.println("T1:烧饭..."); TimeUnit.SECONDS.sleep(20); // 获取T2线程的炒菜 String tf = (String)ft2.get(); System.out.println("T1:获取炒菜:" + tf); System.out.println("T1:吃饭..."); return "吃饭:" + tf; } } // T2Task需要执行的任务: // 洗炒锅，洗菜，切肉，炒菜 static class T2Task implements Callable<String> {
 public String call() throws Exception { System.out.println("T2:洗炒锅..."); TimeUnit.SECONDS.sleep(2);
 System.out.println("T2:洗菜..."); TimeUnit.SECONDS.sleep(3);
 System.out.println("T2:切肉..."); TimeUnit.SECONDS.sleep(5);
 System.out.println("T2:炒菜..."); TimeUnit.SECONDS.sleep(10); return "青椒炒肉丝"; } }
 public static void main(String[] args) throws Exception { // 创建任务T2的FutureTask FutureTask ft2 = new FutureTask<String>(new T2Task()); // 创建任务T1的FutureTask FutureTask ft1 = new FutureTask<String>(new T1Task(ft2)); // 线程T1执行任务ft1 Thread T1 = new Thread(ft1); T1.start(); // 线程T2执行任务ft2 Thread T2 = new Thread(ft2); T2.start(); // 等待线程T1执行结果 System.out.println(ft1.get()); }}

import java.util.concurrent.ForkJoinPool;import java.util.concurrent.RecursiveTask;
public class ForkJoinCalculate { public static void main(String[] args) { ForkJoinPool forkJoinPool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
 System.out.println(forkJoinPool.invoke(new N(1, 100))); }
 static class N extends RecursiveTask<Integer> { final int start, end; N(int start, int end) { this.start = start; this.end = end; } protected Integer compute() { if (start == end) { return start; }
 int m = (start + end) / 2; N n1 = null; if (m+1 <= end) { n1 = new N(m + 1, end); n1.fork(); } N n2 = new N(start, m); return n2.compute() + (n1 == null ? 0 : n1.join()); } }}

public interface CompletionService<V> { Future submit(Callable task); Future submit(Runnable task, V result); /** 阻塞方式获取计算结果 */ Future take() throws InterruptedException; /** 非阻塞方式获取计算结果 */ Future poll(); /** 添加了超时机制的poll */ Future poll(long timeout, TimeUnit unit) throws InterruptedException;}

// ExecutorCompletionService实现类public class ExecutorCompletionService<V> implements CompletionService<V> { private final Executor executor; private final AbstractExecutorService aes; private final BlockingQueue<Future<V>> completionQueue;
 ... // 构造函数1 public ExecutorCompletionService(Executor executor) { ... } /** 构造函数2 * 如果不指定 completionQueue，那么默认会使用无界的 LinkedBlockingQueue。 * 任务执行结果的 Future 对象就是加入到 completionQueue 中。 */ public ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue) { ... }}

// 创建线程池ExecutorService executor = Executors.newFixedThreadPool(8);// 创建CompletionServiceCompletionService cs = new ExecutorCompletionService<>(executor);// 用于保存Future对象List<Future<Integer>> futures = new ArrayList<>(10);//提交异步任务，并保存future到futures for (int i = 0; i < remain; ++i) { futures.add( cs.submit(()->downloadImage(i)));}// 累计剩下的Integer remain = 10;try { // 对于返回结果的任务，立即处理，无需等待所有的都完成 for (int i = 0; i < remain; ++i) { int res = cs.poll(); //简单地通过判空来检查是否成功返回 if (res != null) { remain--; // 处理成功的返回，省略。 }} finally {  //取消所有任务  for(Future f : futures) f.cancel(true);}// 返回结果return r;