[关闭]
@zsh-o 2018-11-05T15:53:33.000000Z 字数 18417 阅读 1106

从头理解理解future模式

Java

问题

首先给出一个同步问题的定义
未命名文件 (1).png-16.6kB
这里就要提到python的动态类型的tuple的好处，可以实现统一形式的多输入和多输出

接下来从最简单的开始一步一步进行优化

顺序结构

首先从最基本的顺序结构执行，要保证结果的正确性，需要按照上述有向图的一种拓扑排序顺序来组织代码
顺序结构我们只能按照其一种拓扑排序的方法组织其运行，这里按照[A,B,C,D,E,F]顺序运行即可

  1. package com.zsh_o.future;
  3. /**
  4.  * 顺序结构执行
  5.  * */
  7. public class SequentialBase {
  8.     public static void main(String[] args) {
  9.         SequentialBase app = new SequentialBase();
  10.         try {
  11.             app.run();
  12.         } catch (InterruptedException e) {
  13.             e.printStackTrace();
  14.         }
  15.     }
  17.     void run() throws InterruptedException {
  18.         long startTime = System.currentTimeMillis();
  20.         int a = getA();
  21.         int b = getB(a);
  22.         int c = getC();
  23.         int d = getD();
  24.         int e = getE(c, d);
  25.         int f = getF(b, e);
  27.         long endTime = System.currentTimeMillis();
  29.         System.out.printf("Final Result: %d\n", f);
  30.         System.out.printf("Total Time: %d\n", endTime - startTime);
  31.     }
  33.     /**
  34.      * Define Functions A -> F
  35.      * */
  36.     int getA() throws InterruptedException {
  37.         System.out.println("A: Running");
  38.         Thread.sleep(1000);
  39.         System.out.println("A: Returned");
  40.         return 1;
  41.     }
  42.     int getB(int a) throws InterruptedException {
  43.         System.out.println("B: Running");
  44.         Thread.sleep(1000);
  45.         System.out.println("B: Returned");
  46.         return a + 1;
  47.     }
  48.     int getC() throws InterruptedException {
  49.         System.out.println("C: Running");
  50.         Thread.sleep(1000);
  51.         System.out.println("C: Returned");
  52.         return 10;
  53.     }
  54.     int getD() throws InterruptedException {
  55.         System.out.println("D: Running");
  56.         Thread.sleep(1000);
  57.         System.out.println("D: Returned");
  58.         return 20;
  59.     }
  60.     int getE(int c, int d) throws InterruptedException {
  61.         System.out.println("E: Running");
  62.         Thread.sleep(1000);
  63.         System.out.println("E: Returned");
  64.         return c + d;
  65.     }
  66.     int getF(int b, int e) throws InterruptedException {
  67.         System.out.println("F: Running");
  68.         Thread.sleep(1000);
  69.         System.out.println("F: Returned");
  70.         return b * e;
  71.     }
  72. }

执行结果：

  1. A: Running
  2. A: Returned
  3. B: Running
  4. B: Returned
  5. C: Running
  6. C: Returned
  7. D: Running
  8. D: Returned
  9. E: Running
  10. E: Returned
  11. F: Running
  12. F: Returned
  13. Final Result: 60
  14. Total Time: 6003

可以看到执行效果还是非常不错的，加上输入输出也只多了3ms，再复杂的代码都是由最基础的代码一步一步按照场景优化来的，所以先来优化这个顺序结构

先从小功能开始，记录时间的代码太长了，如果要计时的太多会产生很多的计时变量，把其封装一下：

  1. package com.zsh_o.util;
  3. /**
  4.  * 用以记录程序运行时间
  5.  * */
  7. public class CountTimer {
  8.     private long startTime;
  9.     private long endTime;
  10.     private long time;
  12.     public CountTimer() {
  13.         startTime = 0;
  14.         endTime = 0;
  15.         time = 0;
  16.     }
  18.     public void start() {
  19.         startTime = System.currentTimeMillis();
  20.     }
  21.     public void end() {
  22.         endTime = System.currentTimeMillis();
  23.         time = endTime - startTime;
  24.     }
  25.     public long getTime() {
  26.         return time;
  27.     }
  28. }

第二个是，为了查看效果我们强行家了sleep一秒，但sleep是要catch异常的，所以每sleep一次都要处理下异常，在这个里面是完全没有必要的，我们单拿出来并处理下异常

  1. package com.zsh_o.util;
  3. public class Util {
  4.     public static void sleep(long ms) {
  5.         try {
  6.             Thread.sleep(ms);
  7.         } catch (InterruptedException e) {
  8.             e.printStackTrace();
  9.         }
  10.     }
  11. }

接下来我们考虑如何对代码进行复用，首先发现可以抽象出函数的调用，这样用户只需要关心函数体的内容就可以了，所以抽象出函数的调用，增加Callable接口

  1. package com.zsh_o.future;
  2. public interface Callable<T> {
  3.     T call();
  4. }

这个地方可以看到我们定义的接口没有传递参数，是由于我们不知道上层用户想要传递几个参数，也不知道每个参数是什么类型，所以最好的办法就是不写参数。。。
python里面传递参数相当于直接传递tuple和dict，然后在函数里面检验参数正确性，但对于Java这种静态编译语言这种方法太影响性能，直接用Object[]在函数里面类型转换，这样对函数参数的约束太弱，很容易崩溃。Java也用泛型实现Tuple，但无法保证长度，所以这个地方直接不加参数，用闭包实现功能。

  1. package com.zsh_o.future;
  3. import com.zsh_o.util.CountTimer;
  4. import com.zsh_o.util.Util;
  5. import java.util.ArrayList;
  7. public class SequentialCase1 {
  8.     int a, b, c, d, e, f;
  10.     public static void main(String[] args) {
  11.         SequentialCase1 app = new SequentialCase1();
  12.         app.run();
  13.     }
  15.     void run() {
  16.         CountTimer timer = new CountTimer();
  17.         timer.start();
  18.         ArrayList<Callable> array = new ArrayList<>();
  19.         array.add(() -> {System.out.println("A: Running"); Util.sleep(1000); a = 1; System.out.println("A: Returned"); return a;});
  20.         array.add(() -> {System.out.println("B: Running"); Util.sleep(1000); b = a + 1; System.out.println("B: Returned"); return b;});
  21.         array.add(() -> {System.out.println("C: Running"); Util.sleep(1000); c = 10; System.out.println("C: Returned"); return c;});
  22.         array.add(() -> {System.out.println("D: Running"); Util.sleep(1000); d = 20; System.out.println("D: Returned"); return d;});
  23.         array.add(() -> {System.out.println("E: Running"); Util.sleep(1000); e = c + d; System.out.println("E: Returned"); return e;});
  24.         array.add(() -> {System.out.println("F: Running"); Util.sleep(1000); f = b * e; System.out.println("F: Returned"); return f;});
  25.         for(var e : array) {
  26.             e.call();
  27.         }
  28.         System.out.printf("Final Result: %d\n", f);
  29.         timer.end();
  30.         System.out.printf("Total Time: %d\n", timer.getTime());
  31.     }
  32. }

结果如下

  1. A: Running
  2. A: Returned
  3. B: Running
  4. B: Returned
  5. C: Running
  6. C: Returned
  7. D: Running
  8. D: Returned
  9. E: Running
  10. E: Returned
  11. F: Running
  12. F: Returned
  13. Final Result: 60
  14. Total Time: 6046

可以看到ArrayList还是蛮耗时的

接下来我们发现，为了查看方法的执行效果，输出了每个方法的运行状态，但该运行状态在每个方法上都是相似的，包括sleep一秒也是相似的，所以，上面两个部分相当于对想实现功能的结果的附加和扩展，所以这里我们用装饰器模式对其扩展附加功能，首先定义该装饰器

  1. package com.zsh_o.future;
  3. /**
  4.  * 采用装饰器对Call进行装饰，为Call增加附加功能
  5.  * */
  7. public class CallDecorator<T> implements Callable<T> {
  8.     Callable<T> callable;
  9.     String name;
  11.     public CallDecorator(String name, Callable<T> callable) {
  12.         this.callable = callable;
  13.         this.name = name;
  14.     }
  16.     @Override
  17.     public T call() {
  18.         System.out.printf("%s: Running\n", name);
  19.         try {
  20.             Thread.sleep(1000);
  21.         } catch (InterruptedException e) {
  22.             e.printStackTrace();
  23.         }
  24.         T value = callable.call();
  25.         System.out.printf("%s: Returned\n", name);
  26.         return value;
  27.     }
  28. }

这里有一个问题，为什么是装饰器而不是代理？代理强调的是对代理对象的访问和执行控制，而装饰器强调对行为的扩展，这个地方只是加了输出和暂停一秒，所以叫装饰器更为合适

  1. package com.zsh_o.future;
  2. import com.zsh_o.util.CountTimer;
  3. import java.util.ArrayList;
  5. public class SequentialCase2{
  6.     int a, b, c, d, e, f;
  7.     public static void main(String[] args) {
  8.         SequentialCase2 app = new SequentialCase2();
  9.         try {
  10.             app.run();
  11.         } catch (InterruptedException e1) {
  12.             e1.printStackTrace();
  13.         }
  14.     }
  16.     void run() throws InterruptedException {
  17.         CountTimer timer = new CountTimer();
  18.         timer.start();
  19.         ArrayList<CallDecorator> array = new ArrayList<>();
  20.         array.add(new CallDecorator("A", () -> {a = 1; return a;}));
  21.         array.add(new CallDecorator("B", () -> {b = a + 1; return b;}));
  22.         array.add(new CallDecorator("C", () -> {c = 10; return c;}));
  23.         array.add(new CallDecorator("D", () -> {d = 20; return d;}));
  24.         array.add(new CallDecorator("E", () -> {e = c + d; return e;}));
  25.         array.add(new CallDecorator("F", () -> {f = b * e; return f;}));
  26.         for(var e : array) {
  27.             e.call();
  28.         }
  29.         timer.end();
  31.         System.out.printf("Final Result: %d\n", f);
  32.         System.out.printf("Total Time: %d\n", timer.getTime());
  33.     }
  34. }

结果：

  1. A: Running
  2. A: Returned
  3. B: Running
  4. B: Returned
  5. C: Running
  6. C: Returned
  7. D: Running
  8. D: Returned
  9. E: Running
  10. E: Returned
  11. F: Running
  12. F: Returned
  13. Final Result: 60
  14. Total Time: 6055

并发

我们用并发的方式优化该代码，该程序理论上最少的运行时间是3s，每一列的代码可以一起执行。在最一开始我们采用了6把锁来完成这个同步问题

  1. package com.zsh_o.future;
  3. import com.zsh_o.util.CountTimer;
  4. import com.zsh_o.util.Util;
  6. import java.util.ArrayList;
  7. import java.util.concurrent.CountDownLatch;
  8. import java.util.concurrent.locks.ReentrantLock;
  10. /**
  11.  * 多线程解决，可能会出现的问题是还没来得及加锁，其他的线程已经开始运行
  12.  * */
  14. public class ThreadBase1 {
  15.     private int a, b, c, d, e, f;
  16.     private ReentrantLock lockA, lockB, lockC, lockD, lockE, lockF;
  18.     public ThreadBase1() {
  19.         this.lockA = new ReentrantLock();
  20.         this.lockB = new ReentrantLock();
  21.         this.lockC = new ReentrantLock();
  22.         this.lockD = new ReentrantLock();
  23.         this.lockE = new ReentrantLock();
  24.         this.lockF = new ReentrantLock();
  25.     }
  27.     void run() throws InterruptedException {
  28.         CountTimer timer = new CountTimer();
  29.         timer.start();
  30.         // 用CountDownLatch确保主线程在所有子线程执行完再执行
  31.         CountDownLatch latch = new CountDownLatch(6);
  32.         ArrayList<Thread> array = new ArrayList<>();
  33.         array.add(new Thread(() -> {
  34.             lockA.lock();
  35.             System.out.println("A: Running");
  36.             Util.sleep(1000);a = 1;
  37.             System.out.println("A: Returned");
  38.             lockA.unlock();
  39.             latch.countDown();
  40.         }));
  41.         array.add(new Thread(() -> {
  42.             lockB.lock();lockA.lock();
  43.             System.out.println("B: Running");
  44.             Util.sleep(1000);b = a + 1;
  45.             System.out.println("B: Returned");
  46.             lockA.unlock();lockB.unlock();
  47.             latch.countDown();
  48.         }));
  49.         array.add(new Thread(() -> {
  50.             lockC.lock();
  51.             System.out.println("C: Running");
  52.             Util.sleep(1000);c = 10;
  53.             System.out.println("C: Returned");
  54.             lockC.unlock();
  55.             latch.countDown();
  56.         }));
  57.         array.add(new Thread(() -> {
  58.             lockD.lock();
  59.             System.out.println("D: Running");
  60.             Util.sleep(1000);d = 20;
  61.             System.out.println("D: Returned");
  62.             lockD.unlock();
  63.             latch.countDown();
  64.         }));
  65.         array.add(new Thread(() -> {
  66.             lockE.lock();lockC.lock();lockD.lock();
  67.             System.out.println("E: Running");
  68.             Util.sleep(1000);e = c + d;
  69.             System.out.println("E: Returned");
  70.             lockD.unlock();lockC.unlock();lockE.unlock();
  71.             latch.countDown();
  72.         }));
  73.         array.add(new Thread(() -> {
  74.             lockF.lock();lockB.lock();lockE.lock();
  75.             System.out.println("F: Running");
  76.             Util.sleep(1000);f = b * e;
  77.             System.out.println("F: Returned");
  78.             lockE.unlock();lockB.unlock();lockF.unlock();
  79.             latch.countDown();
  80.         }));
  81.         for(var e: array) {
  82.             e.start();
  83.         }
  84.         latch.await();
  85.         timer.end();
  86.         System.out.printf("Final Result: %d\n", f);
  87.         System.out.printf("Total Time: %d\n", timer.getTime());
  88.     }
  90.     public static void main(String[] args) {
  91.         ThreadBase1 app = new ThreadBase1();
  92.         try {
  93.             app.run();
  94.         } catch (InterruptedException e1) {
  95.             e1.printStackTrace();
  96.         }
  97.     }
  99. }

结果：

  1. D: Running
  2. A: Running
  3. C: Running
  4. D: Returned
  5. C: Returned
  6. A: Returned
  7. E: Running
  8. B: Running
  9. E: Returned
  10. B: Returned
  11. F: Running
  12. F: Returned
  13. Final Result: 60
  14. Total Time: 3030

这个可能会出现的问题是，可能会出现还没来得及加锁后面的就运行了，则会出现错误结果，所以我们用CountDownLatch来修改该代码

  1. package com.zsh_o.future;
  4. import com.zsh_o.util.CountTimer;
  5. import com.zsh_o.util.Util;
  7. import java.util.ArrayList;
  9. /**
  10.  * 多线程解决
  11.  * 用CountDownLatch完善
  12.  * */
  14. public class ThreadCase1 {
  15.     private int a, b, c, d, e, f;
  16.     private CountDownLatch latchB, latchE, latchF;
  18.     public ThreadCase1() {
  19.         latchB = new CountDownLatch(1);
  20.         latchE = new CountDownLatch(2);
  21.         latchF = new CountDownLatch(2);
  22.     }
  24.     void run() throws InterruptedException {
  25.         CountTimer timer = new CountTimer();
  26.         timer.start();
  27.         CountDownLatch latchMain = new CountDownLatch(1);
  28.         ArrayList<Thread> array = new ArrayList<>();
  29.         array.add(new Thread(() -> {
  30.             System.out.println("A: Running");
  31.             Util.sleep(1000);a = 1;
  32.             System.out.println("A: Returned");
  33.             latchB.countDown();
  34.         }));
  35.         array.add(new Thread(() -> {
  36.             try {
  37.                 latchB.await();
  38.                 System.out.println("B: Running");
  39.                 Util.sleep(1000);b = a + 1;
  40.                 System.out.println("B: Returned");
  41.                 latchF.countDown();
  42.             } catch (InterruptedException e1) {
  43.                 e1.printStackTrace();
  44.             }
  46.         }));
  47.         array.add(new Thread(() -> {
  48.             System.out.println("C: Running");
  49.             Util.sleep(1000);c = 10;
  50.             System.out.println("C: Returned");
  51.             latchE.countDown();
  52.         }));
  53.         array.add(new Thread(() -> {
  54.             System.out.println("D: Running");
  55.             Util.sleep(1000);d = 20;
  56.             System.out.println("D: Returned");
  57.             latchE.countDown();
  58.         }));
  59.         array.add(new Thread(() -> {
  60.             try {
  61.                 latchE.await();
  62.                 System.out.println("E: Running");
  63.                 Util.sleep(1000);e = c + d;
  64.                 System.out.println("E: Returned");
  65.                 latchF.countDown();
  66.             } catch (InterruptedException e1) {
  67.                 e1.printStackTrace();
  68.             }
  69.         }));
  70.         array.add(new Thread(() -> {
  71.             try {
  72.                 latchF.await();
  73.                 System.out.println("F: Running");
  74.                 Util.sleep(1000);f = b * e;
  75.                 System.out.println("F: Returned");
  76.                 latchMain.countDown();
  77.             } catch (InterruptedException e1) {
  78.                 e1.printStackTrace();
  79.             }
  80.         }));
  81.         for(var e: array) {
  82.             e.start();
  83.         }
  84.         latchMain.await();
  85.         System.out.printf("Final Result: %d\n", f);
  86.         timer.end();
  87.         System.out.printf("Total Time: %d\n", timer.getTime());
  88.     }
  90.     public static void main(String[] args) {
  91.         ThreadCase1 app = new ThreadCase1();
  92.         try {
  93.             app.run();
  94.         } catch (InterruptedException e1) {
  95.             e1.printStackTrace();
  96.         }
  97.     }
  98. }

执行结果与上面相同，可以看到CountownLatch非常好用，而且原理非常简单，那么我们来实现个吧
原理：await()阻断线程，并且MaxCount个线程执行countDown()之后才打开阻断，需要是不同的线程，需要注意的是countDown()函数不截断执行的线程，只截断执行await()的线程，如果想同时截断需要配合使用

  1. package com.zsh_o.future;
  3. import com.zsh_o.util.CountTimer;
  4. import com.zsh_o.util.Util;
  6. import javax.swing.event.MouseInputListener;
  7. import java.util.concurrent.locks.ReentrantLock;
  9. public class CountDownLatch {
  10.     private int count;
  11.     final private Object gobal;
  12.     private ReentrantLock lockCount;
  14.     public CountDownLatch(int count) {
  15.         this.count = count;
  16.         lockCount = new ReentrantLock();
  17.         gobal = new Object();
  18.     }
  19.     public void countDown() {
  20.         lockCount.lock();
  21.         if(count > 0)
  22.             count--;
  23.         else return;
  24.         if(count == 0)
  25.             synchronized (gobal) {
  26.                 gobal.notifyAll();
  27.             }
  28.         lockCount.unlock();
  29.     }
  30.     public void await() throws InterruptedException {
  31.         synchronized (gobal) {
  32.             if (count > 0)
  33.                 gobal.wait();
  34.         }
  35.     }
  36. }

这里用了java的Object的wait和noifyAll()函数，用以任意数量的线程等待和恢复

Future

上面的问题是需要用户自己来考虑同步机制，如何把同步方法也封装到代码里面，让用户只关心代码逻辑即可，这个地方用的是Future模式，顾名思义，Future表示的是未来的一种意愿，表示未来会发生的事情，这个地方表示该实体在未来会得到返回值，调用future.get的表示期望得到给结果，如果结果还没有准备好则等待结果，例如上面的E会执行int e = futureC.get() + futureD.get()则E会等待C和D执行完之后继续执行。这个地方涉及到两个点：一，如何进入等待，二，如何从等待中恢复；这里的原则是等待过程不能占用计算资源，也就是不能用while一直判断的方法实现，那么实现方法就应该是事件驱动，进入等待释放计算资源，从等待恢复再重新可计算。

这个地方需要用代理模式，用代理形成逻辑，最后具体的执行由代理控制
首先定义futurable接口

  1. package com.zsh_o.future;
  3. public interface Futurable<T> {
  4.     T get();
  5. }

下面是Future抽象类

  1. package com.zsh_o.future;
  3. public abstract class Future<T> implements Futurable<T>, Runnable {
  4.     protected Callable<T> callable;
  5.     protected boolean finished;
  7.     public Future() {
  8.         this.callable = null;
  9.         this.finished = false;
  10.     }
  12.     public void register(Callable<T> callable) {
  13.         this.callable = callable;
  14.     }
  15. }

接下来是以线程的方式实现Future，当然也可以实现分布式的方式，核心思想相似

  1. package com.zsh_o.future;
  2. import com.zsh_o.util.CountTimer;
  3. import com.zsh_o.util.Util;
  5. public class ThreadFuture<T> extends Future<T> {
  6.     private CountDownLatch latch;
  7.     private T value;
  8.     private boolean finished;
  9.     private final Object lock;
  10.     public ThreadFuture() {
  11.         latch = new CountDownLatch(1);
  12.         value = null;
  13.         finished = false;
  14.         lock = new Object();
  15.     }
  17.     @Override
  18.     public T get() {
  19.         try {
  20.             if (callable == null)
  21.                 throw new Exception("Unregistered Callable");
  22.             if (!finished){
  23.                 run();
  24.                 latch.await();
  25.             }
  27.         } catch (Exception e) {
  28.             e.printStackTrace();
  29.         }
  30.         return value;
  31.     }
  33.     @Override
  34.     public void run() {
  35.         new Thread(()->{
  36.             synchronized (lock) {
  37.                 if (!finished) {
  38.                     value = callable.call();
  39.                     finished = true;
  40.                     latch.countDown();
  41.                 }
  42.             }
  43.         }).start();
  44.     }
  46.     public void reset() {
  47.         synchronized (lock) {
  48.             finished = false;
  49.         }
  50.     }
  52.     public boolean getState() {
  53.         return finished;
  54.     }
  55. }

实现的功能是，用register注册函数体逻辑Callable，需要注意的是Callable中需要调用代理的future.get()方法，以形成执行逻辑，get()执行await进入等待，等待其他线程执行完run()，并且保证只执行一次

  1. package com.zsh_o.future;
  3. import com.zsh_o.util.CountTimer;
  4. import com.zsh_o.util.Util;
  5. public class ThreadCase2 {
  6.     private ThreadFuture<Integer> futureA, futureB, futureC, futureD, futureE, futureF;
  8.     public ThreadCase2() {
  9.         this.futureA = new ThreadFuture<>();
  10.         this.futureB = new ThreadFuture<>();
  11.         this.futureC = new ThreadFuture<>();
  12.         this.futureD = new ThreadFuture<>();
  13.         this.futureE = new ThreadFuture<>();
  14.         this.futureF = new ThreadFuture<>();
  15.     }
  17.     void run() {
  18.         CountTimer timer = new CountTimer();
  19.         timer.start();
  20.         futureA.register(()->{System.out.println("A: Runing");int a = 1;Util.sleep(1000);System.out.println("A: Over");return a;});
  21.         futureB.register(()->{System.out.println("B: Runing");int b = futureA.get()+1;Util.sleep(1000);System.out.println("B: Over");return b;});
  22.         futureC.register(()->{System.out.println("C: Runing");int c = 10;Util.sleep(1000);System.out.println("C: Over");return c;});
  23.         futureD.register(()->{System.out.println("D: Runing");int d = 20;Util.sleep(1000);System.out.println("D: Over");return d;});
  24.         futureE.register(()->{System.out.println("E: Runing");int e = futureC.get()+futureD.get();Util.sleep(1000);System.out.println("E: Over");return e;});
  25.         futureF.register(()->{System.out.println("F: Runing");int f = futureB.get()*futureE.get();Util.sleep(1000);System.out.println("F: Over");return f;});
  26.         futureA.run();futureB.run();futureC.run();futureD.run();futureE.run();futureF.run();
  27.         System.out.printf("Final Result: %d\n", futureF.get());
  28.         timer.end();
  29.         System.out.printf("Total Time: %d\n", timer.getTime());
  30.     }
  32.     public static void main(String[] args) {
  33.         ThreadCase2 app = new ThreadCase2();
  34.         app.run();
  35.     }
  36. }

结果：

  1. A: Runing
  2. B: Runing
  3. C: Runing
  4. D: Runing
  5. E: Runing
  6. F: Runing
  7. C: Over
  8. D: Over
  9. A: Over
  10. E: Over
  11. B: Over
  12. F: Over
  13. Final Result: 60
  14. Total Time: 3051

最后，如果并发的非常多可能会崩掉，所以我们加了一个限制线程执行状态最大值的方法

  1. package com.zsh_o.future;
  3. import java.util.concurrent.ConcurrentHashMap;
  4. import java.util.concurrent.ConcurrentLinkedDeque;
  6. public class ThreadPoolFuture<T> {
  7.     private class Item<D> extends Future<D> {
  8.         Thread thread;
  9.         D value;
  10.         CountDownLatch latch;
  11.         final Object lock;
  12.         boolean finished;
  14.         public Item() {
  15.             this.value = null;
  16.             this.latch = new CountDownLatch(1);
  17.             this.lock = new Object();
  18.             this.finished = false;
  19.         }
  21.         @Override
  22.         public D get() {
  23.             return value;
  24.         }
  26.         @Override
  27.         public void run() {
  28.             if (!thread.isAlive()) {
  29.                 thread.start();
  30.             }
  31.         }
  32.     }
  34.     private int maxPool;
  35.     private final Object lockObject;
  36.     private ConcurrentHashMap<String, Item<T>> pool;
  37.     private ConcurrentLinkedDeque<Item<T>> poolList;
  38.     private int currentRuning;
  39.     private CountDownLatch lathGobal;
  41.     public ThreadPoolFuture(int maxPool) {
  42.         this.maxPool = maxPool;
  43.         this.pool = new ConcurrentHashMap<>();
  44.         poolList = new ConcurrentLinkedDeque<>();
  45.         this.lockObject = new Object();
  46.         this.currentRuning = 0;
  47.     }
  49.     public void register(String name, Callable<T> callable) {
  50.         Item<T> item = new Item<>();
  51.         item.thread = new Thread(()->{
  52.             synchronized (item.lock) {
  53.                 if (!item.finished) {
  54.                     synchronized (lockObject) {
  55.                         while (currentRuning >= maxPool) {
  56.                             try {
  57.                                 lockObject.wait();
  58.                             } catch (InterruptedException e) {
  59.                                 e.printStackTrace();
  60.                             }
  61.                         }
  62.                         currentRuning++;
  63.                         System.out.println(name + " Run: " + currentRuning);
  64.                     }
  65.                     item.value = callable.call();
  66.                     synchronized (lockObject) {
  67.                         currentRuning--;
  68.                         lockObject.notifyAll();
  70.                     }
  71.                     item.latch.countDown();
  72.                     lathGobal.countDown();
  73.                     item.finished = true;
  74.                 }
  75.             }
  76.         });
  77.         pool.put(name, item);
  78.         poolList.push(item);
  79.     }
  81.     public T get(String name) {
  82.         if (!pool.containsKey(name)) {
  83.             try {
  84.                 throw new Exception("Unregister Callable");
  85.             } catch (Exception e) {
  86.                 e.printStackTrace();
  87.             }
  88.         }
  89.         Item<T> item = pool.get(name);
  90.         if (!item.finished) {
  91.             try {
  92.                 synchronized (lockObject) {
  93.                     lockObject.notifyAll();
  94.                     currentRuning--;
  95.                     System.out.println(name + " Get: " + currentRuning);
  96.                 }
  97.                 item.run();
  98.                 item.latch.await();
  99.                 synchronized (lockObject) {
  100.                     while (currentRuning >= maxPool) {
  101.                         try {
  102.                             lockObject.wait();
  103.                         } catch (InterruptedException e) {
  104.                             e.printStackTrace();
  105.                         }
  106.                     }
  107.                     currentRuning++;
  108.                     System.out.println(name + " Get: " + currentRuning);
  109.                 }
  110.             } catch (InterruptedException e) {
  111.                 e.printStackTrace();
  112.                 return null;
  113.             }
  114.         }
  115.         return item.value;
  116.     }
  118.     public void start() {
  119.         lathGobal = new CountDownLatch(poolList.size());
  120.         for(var e: poolList) {
  121.             e.run();
  122.         }
  123.     }
  124.     public void await() throws InterruptedException {
  125.         if (lathGobal == null)
  126.             lathGobal = new CountDownLatch(poolList.size());
  127.         lathGobal.await();
  128.     }
  129. }

基本思路是，由于上面我们用消息的方法执行等待和恢复执行，所以进入等待和恢复执行对于我们来说是可见的，所以只需要监控当前运行的最大线程即可，只是执行状态的数量不包括阻塞和等待

  1. package com.zsh_o.future;
  3. import com.zsh_o.util.CountTimer;
  4. import com.zsh_o.util.Util;
  7. /**
  8.  * Future模式解决问题,限制最大运行线程数
  9.  * */
  11. public class ThreadCase3 {
  12.     private ThreadPoolFuture<Integer> pool;
  14.     public ThreadCase3() {
  15.         this.pool = new ThreadPoolFuture<>(2);
  16.     }
  18.     void run() throws Exception {
  19.         CountTimer timer = new CountTimer();
  20.         timer.start();
  21.         pool.register("A", ()->{
  22.             System.out.println("A: Running");
  23.             int a = 1;
  24.             Util.sleep(1000);
  25.             System.out.println("A: Over");
  26.             return a;
  27.         });
  28.         pool.register("B", ()->{
  29.             System.out.println("B: Running");
  30.             int b = pool.get("A") + 1;
  31.             Util.sleep(1000);
  32.             System.out.println("B: Over");
  33.             return b;
  34.         });
  35.         pool.register("C", ()->{
  36.             System.out.println("C: Running");
  37.             int c = 10;
  38.             Util.sleep(1000);
  39.             System.out.println("C: Over");
  40.             return c;
  41.         });
  42.         pool.register("D", ()->{
  43.             System.out.println("D: Running");
  44.             Util.sleep(1000);
  45.             System.out.println("D: Over");
  46.             return 20;
  47.         });
  48.         pool.register("E",()->{
  49.             System.out.println("E: Running");
  50.             Util.sleep(1000);
  51.             int e = pool.get("C") + pool.get("D");
  52.             System.out.println("E: Over");
  53.             return e;
  54.         });
  55.         pool.register("F", ()->{
  56.             System.out.println("F: Running");
  57.             int f = pool.get("B") * pool.get("E");
  58.             Util.sleep(1000);
  59.             System.out.println("F: Over");
  60.             return f;
  61.         });
  62.         pool.start();
  63.         pool.await();
  64.         System.out.printf("Final Result: %d\n", pool.get("F"));
  65.         timer.end();
  66.         System.out.printf("Total Time: %d\n", timer.getTime());
  67.     }
  69.     public static void main(String[] args) {
  70.         ThreadCase3 app = new ThreadCase3();
  71.         try {
  72.             app.run();
  73.         } catch (Exception e) {
  74.             e.printStackTrace();
  75.         }
  76.     }
  77. }

结果：

  1. maxCount = 2
  2. F Run: 1
  3. F: Running
  4. B Get: 0
  5. A Run: 1
  6. A: Running
  7. B Run: 2
  8. B: Running
  9. A Get: 1
  10. C Run: 2
  11. C: Running
  12. A: Over
  13. C: Over
  14. E Run: 2
  15. E: Running
  16. D Run: 2
  17. D: Running
  18. D Get: 1
  19. D: Over
  20. A Get: 2
  21. D Get: 2
  22. E: Over
  23. B: Over
  24. B Get: 1
  25. F: Over
  26. Final Result: 60
  27. Total Time: 4072
  28. ------------------------------
  29. maxCount = 1
  30. F Run: 1
  31. F: Running
  32. B Get: 0
  33. A Run: 1
  34. A: Running
  35. A: Over
  36. E Run: 1
  37. E: Running
  38. C Get: 0
  39. B Run: 1
  40. B: Running
  41. B: Over
  42. D Run: 1
  43. D: Running
  44. D: Over
  45. B Get: 1
  46. E Get: 0
  47. C Run: 1
  48. C: Running
  49. C: Over
  50. C Get: 1
  51. E: Over
  52. E Get: 1
  53. F: Over
  54. Final Result: 60
  55. Total Time: 6098
  56. ---------------------
  57. maxCount = 3
  58. F Run: 1
  59. F: Running
  60. A Run: 2
  61. A: Running
  62. B Run: 3
  63. B: Running
  64. A Get: 2
  65. E Run: 3
  66. E: Running
  67. B Get: 2
  68. D Run: 3
  69. D: Running
  70. A: Over
  71. C Run: 3
  72. D: Over
  73. C Get: 2
  74. C: Running
  75. A Get: 3
  76. C: Over
  77. C Get: 2
  78. E: Over
  79. B: Over
  80. B Get: 1
  81. F: Over
  82. Final Result: 60
  83. Total Time: 3088
  84. -------------------------
  85. maxCount = 4
  86. F Run: 1
  87. F: Running
  88. A Run: 2
  89. A: Running
  90. B Run: 3
  91. B: Running
  92. C Run: 4
  93. C: Running
  94. A Get: 3
  95. B Get: 2
  96. E Run: 3
  97. E: Running
  98. D Run: 4
  99. D: Running
  100. A: Over
  101. C: Over
  102. D: Over
  103. C Get: 2
  104. A Get: 2
  105. C Get: 2
  106. E: Over
  107. B: Over
  108. B Get: 1
  109. F: Over
  110. Final Result: 60
  111. Total Time: 3078
添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注