乐趣区

设计模式你精通那高并发中的设计模式呢是不是也精通呢

在开始明天的内容之前,是不是先来温习一下设计模式,毕竟不是所有人对于设计模式都是精通的状态,那这个时候,先温习一下再进行新的步骤,不失为一种很好的状态呀

设计模式

因为图谱开展之后切实是太大了,就不开展了,给大家截取一部分,内容涵盖 23 种设计模式,然而毕竟不是重点

设计准则

对于设计准则,在上方的设计模式图中,就介绍了 6 大设计模式,这一篇中,进行了分类的具体解说,算是附赠图谱吧

须要下面 2 份脑图的,关注 + 转发后,私信“材料”即可查看获取形式

而在回顾了下面的设计模式的相干内容之后,接下来就是本文的重点了,解说几种常见并行模式, 具体目录构造如下图.

单例

单例是最常见的一种设计模式, 个别用于全局对象治理, 比方 xml 配置读写之类的.

个别分为懒汉式, 饿汉式.

懒汉式: 办法上加 synchronized

1 public static synchronized Singleton getInstance() { 2          if (single == null) {3              single = new Singleton(); 4          } 5         return single; 6 }

这种形式, 因为每次获取示例都要获取锁, 不举荐应用, 性能较差

懒汉式: 应用双检锁 + volatile

1     private volatile Singleton singleton = null;  2     public static Singleton getInstance() {  3         if (singleton == null) {4             synchronized (Singleton.class) {5                 if (singleton == null) {6                     singleton = new Singleton();  7                 }  8             }  9         } 10         return singleton; 11     }

本形式是对间接在办法上加锁的一个优化, 益处在于只有第一次初始化获取了锁.

后续调用 getInstance 曾经是无锁状态. 只是写法上略微繁琐点.

至于为什么要 volatile 关键字, 次要波及到 jdk 指令重排, 详见之前的博文: Java 内存模型与指令重排

懒汉式: 应用动态外部类

1 public class Singleton {2     private static class LazyHolder { 3        private static final Singleton INSTANCE = new Singleton(); 4     } 5     private Singleton (){} 6     public static final Singleton getInstance() { 7        return LazyHolder.INSTANCE; 8} 9 }

该形式既解决了同步问题, 也解决了写法繁琐问题. 举荐应用改写法.

毛病在于无奈响应事件来从新初始化 INSTANCE.

饿汉式

1 public class Singleton1 {2     private Singleton1() {}  3     private static final Singleton1 single = new Singleton1();  4     public static Singleton1 getInstance() {  5         return single;  6}  7 }

毛病在于对象在一开始就间接初始化了.

Future 模式

该模式的核心思想是异步调用. 有点相似于异步的 ajax 申请.

当调用某个办法时, 可能该办法耗时较久, 而在主函数中也不急于立即获取后果.

因而能够让调用者立即返回一个凭证, 该办法放到另外线程执行,

后续主函数拿凭证再去获取办法的执行后果即可, 其结构图如下

jdk 中内置了 Future 模式的反对, 其接口如下:

通过 FutureTask 实现

留神其中两个耗时操作.

  • 如果 doOtherThing 耗时 2s, 则整个函数耗时 2s 左右.
  • 如果 doOtherThing 耗时 0.2s, 则整个函数耗时取决于 RealData.costTime, 即 1s 左右完结.
1 public class FutureDemo1 {2  3     public static void main(String[] args) throws InterruptedException, ExecutionException {4         FutureTask<String> future = new FutureTask<String>(new Callable<String>() {5             @Override  6             public String call() throws Exception {7                 return new RealData().costTime();  8}  9         }); 10         ExecutorService service = Executors.newCachedThreadPool(); 11         service.submit(future); 12 13         System.out.println("RealData 办法调用结束"); 14         // 模仿主函数中其余耗时操作 15         doOtherThing(); 16         // 获取 RealData 办法的后果 17         System.out.println(future.get()); 18     } 19 20     private static void doOtherThing() throws InterruptedException { 21         Thread.sleep(2000L); 22     } 23 } 24 25 class RealData {26 27     public String costTime() {28         try { 29             // 模仿 RealData 耗时操作 30             Thread.sleep(1000L); 31             return "result"; 32         } catch (InterruptedException e) {33             e.printStackTrace(); 34         } 35         return "exception"; 36     } 37 38 }

通过 Future 实现

与上述 FutureTask 不同的是, RealData 须要实现 Callable 接口

1 public class FutureDemo2 {2  3     public static void main(String[] args) throws InterruptedException, ExecutionException {4         ExecutorService service = Executors.newCachedThreadPool();  5         Future<String> future = service.submit(new RealData2());  6  7         System.out.println("RealData2 办法调用结束");  8         // 模仿主函数中其余耗时操作  9         doOtherThing(); 10         // 获取 RealData2 办法的后果 11         System.out.println(future.get()); 12     } 13 14     private static void doOtherThing() throws InterruptedException { 15         Thread.sleep(2000L); 16     } 17 } 18 19 class RealData2 implements Callable<String>{20 21     public String costTime() {22         try { 23             // 模仿 RealData 耗时操作 24             Thread.sleep(1000L); 25             return "result"; 26         } catch (InterruptedException e) {27             e.printStackTrace(); 28         } 29         return "exception"; 30     } 31 32     @Override 33     public String call() throws Exception { 34         return costTime(); 35     } 36 }

另外 Future 自身还提供了一些额定的简略管制性能, 其 API 如下

1     // 勾销工作  2     boolean cancel(boolean mayInterruptIfRunning);  3     // 是否曾经勾销  4     boolean isCancelled();  5     // 是否曾经实现  6     boolean isDone();  7     // 获得返回对象  8     V get() throws InterruptedException, ExecutionException;  9     // 获得返回对象, 并能够设置超时工夫 10     V get(long timeout, TimeUnit unit) 11 throws InterruptedException, ExecutionException, TimeoutException;

生产消费者模式

生产者 - 消费者模式是一个经典的多线程设计模式. 它为多线程间的合作提供了良好的解决方案。

在生产者 - 消费者模式中,通常由两类线程,即若干个生产者线程和若干个消费者线程。

生产者线程负责提交用户申请,消费者线程则负责具体解决生产者提交的工作。

生产者和消费者之间则通过共享内存缓冲区进行通信, 其结构图如下

PCData 为咱们须要解决的元数据模型, 生产者构建 PCData, 并放入缓冲队列.

消费者从缓冲队列中获取数据, 并执行计算.

生产者外围代码

1         while(isRunning) {2             Thread.sleep(r.nextInt(SLEEP_TIME));  3             data = new PCData(count.incrementAndGet);  4             // 结构工作数据  5             System.out.println(data + "is put into queue");  6             if (!queue.offer(data, 2, TimeUnit.SECONDS)) {7                 // 将数据放入队列缓冲区中  8                 System.out.println("faild to put data :" + data);  9             } 10         }

消费者外围代码

1         while (true) {2             PCData data = queue.take();  3             // 提取工作  4             if (data != null) {5                 // 获取数据, 执行计算操作  6                 int re = data.getData() * 10;  7                 System.out.println("after cal, value is :" + re);  8                 Thread.sleep(r.nextInt(SLEEP_TIME));  9             } 10         }

生产消费者模式能够无效对数据解耦, 优化系统结构.

升高生产者和消费者线程相互之间的依赖与性能要求.

个别应用 BlockingQueue 作为数据缓冲队列, 他是通过锁和阻塞来实现数据之间的同步,

如果对缓冲队列有性能要求, 则能够应用基于 CAS 无锁设计的 ConcurrentLinkedQueue.

分而治之

严格来讲, 分而治之不算一种模式, 而是一种思维.

它能够将一个大工作拆解为若干个小工作并行执行, 进步零碎吞吐量.

咱们次要讲两个场景, Master-Worker 模式, ForkJoin 线程池.

Master-Worker 模式

该模式核心思想是零碎由两类进行帮助工作: Master 过程, Worker 过程.

Master 负责接管与分配任务, Worker 负责解决工作. 当各个 Worker 解决实现后,

将后果返回给 Master 进行演绎与总结.

假如一个场景, 须要计算 100 个工作, 并对后果求和, Master 持有 10 个子过程.

Master 代码

1 public class MasterDemo {2     // 盛装工作的汇合  3     private ConcurrentLinkedQueue<TaskDemo> workQueue = new ConcurrentLinkedQueue<TaskDemo>();  4     // 所有 worker  5     private HashMap<String, Thread> workers = new HashMap<>();  6     // 每一个 worker 并行执行工作的后果  7     private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<>();  8  9     public MasterDemo(WorkerDemo worker, int workerCount) {10         // 每个 worker 对象都须要持有 queue 的援用, 用于领工作与提交后果 11         worker.setResultMap(resultMap); 12         worker.setWorkQueue(workQueue); 13         for (int i = 0; i < workerCount; i++) {14             workers.put("子节点:" + i, new Thread(worker)); 15         } 16     } 17 18     // 提交工作 19     public void submit(TaskDemo task) {20         workQueue.add(task); 21     } 22 23     // 启动所有的子工作 24     public void execute(){ 25         for (Map.Entry<String, Thread> entry : workers.entrySet()) {26             entry.getValue().start(); 27} 28     } 29 30     // 判断所有的工作是否执行完结 31     public boolean isComplete() { 32         for (Map.Entry<String, Thread> entry : workers.entrySet()) {33             if (entry.getValue().getState() != Thread.State.TERMINATED) {34                 return false; 35} 36         } 37 38         return true; 39     } 40 41     // 获取最终汇总的后果 42     public int getResult() { 43         int result = 0; 44         for (Map.Entry<String, Object> entry : resultMap.entrySet()) {45             result += Integer.parseInt(entry.getValue().toString()); 46         } 47 48         return result; 49     } 50 51 }

Worker 代码

1 public class WorkerDemo implements Runnable{2  3     private ConcurrentLinkedQueue<TaskDemo> workQueue;  4     private ConcurrentHashMap<String, Object> resultMap;  5  6     @Override  7     public void run() {8         while (true) {9             TaskDemo input = this.workQueue.poll(); 10             // 所有工作曾经执行结束 11             if (input == null) {12                 break; 13} 14             // 模仿对 task 进行解决, 返回后果 15             int result = input.getPrice(); 16             this.resultMap.put(input.getId() + "", result); 17             System.out.println(" 工作执行结束, 以后线程: " + Thread.currentThread().getName()); 18         } 19     } 20 21     public ConcurrentLinkedQueue<TaskDemo> getWorkQueue() { 22         return workQueue; 23} 24 25     public void setWorkQueue(ConcurrentLinkedQueue<TaskDemo> workQueue) {26         this.workQueue = workQueue; 27} 28 29     public ConcurrentHashMap<String, Object> getResultMap() { 30         return resultMap; 31} 32 33     public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {34         this.resultMap = resultMap; 35} 36 }
1 public class TaskDemo {2  3     private int id;  4     private String name;  5     private int price;  6  7     public int getId() {8         return id;  9} 10 11     public void setId(int id) {12         this.id = id; 13} 14 15     public String getName() { 16         return name; 17} 18 19     public void setName(String name) {20         this.name = name; 21} 22 23     public int getPrice() { 24         return price; 25} 26 27     public void setPrice(int price) {28         this.price = price; 29} 30 }

主函数测试

1         MasterDemo master = new MasterDemo(new WorkerDemo(), 10);  2         for (int i = 0; i < 100; i++) {3             TaskDemo task = new TaskDemo();  4             task.setId(i);  5             task.setName("工作" + i);  6             task.setPrice(new Random().nextInt(10000));  7             master.submit(task);  8         }  9 10         master.execute(); 11 12         while (true) {13             if (master.isComplete()) {14                 System.out.println("执行的后果为:" + master.getResult()); 15                 break; 16             } 17         }

ForkJoin 线程池

该线程池是 jdk7 之后引入的一个并行执行工作的框架, 其核心思想也是将工作宰割为子工作,

有可能子工作还是很大, 还须要进一步拆解, 最终失去足够小的工作.

将宰割进去的子工作放入双端队列中, 而后几个启动线程从双端队列中获取工作执行.

子工作执行的后果放到一个队列里, 另起线程从队列中获取数据, 合并后果.

假如咱们的场景须要计算从 0 到 20000000L 的累加求和. CountTask 继承自 RecursiveTask, 能够携带返回值.

每次合成大工作, 简略的将工作划分为 100 个等规模的小工作, 并应用 fork() 提交子工作.

在子工作中通过 THRESHOLD 设置子工作合成的阈值, 如果以后须要求和的总数大于 THRESHOLD, 则子工作须要再次合成,

如果子工作能够间接执行, 则进行求和操作, 返回后果. 最终期待所有的子工作执行结束, 对所有后果求和.

1 public class CountTask extends RecursiveTask<Long>{2     // 工作合成的阈值  3     private static final int THRESHOLD = 10000;  4     private long start;  5     private long end;  6  7  8     public CountTask(long start, long end) {9         this.start = start; 10         this.end = end; 11} 12 13     public Long compute() { 14         long sum = 0; 15         boolean canCompute = (end - start) < THRESHOLD; 16         if (canCompute) {17             for (long i = start; i <= end; i++) {18                 sum += i; 19} 20         } else {21             // 分成 100 个小工作 22             long step = (start + end) / 100; 23             ArrayList<CountTask> subTasks = new ArrayList<CountTask>(); 24             long pos = start; 25             for (int i = 0; i < 100; i++) {26                 long lastOne = pos + step; 27                 if (lastOne > end) {28                     lastOne = end; 29} 30                 CountTask subTask = new CountTask(pos, lastOne); 31                 pos += step + 1; 32                 // 将子工作推向线程池 33                 subTasks.add(subTask); 34                 subTask.fork(); 35} 36 37             for (CountTask task : subTasks) {38                 // 对后果进行 join 39                 sum += task.join(); 40             } 41         } 42         return sum; 43     } 44 45     public static void main(String[] args) throws ExecutionException, InterruptedException {46         ForkJoinPool pool = new ForkJoinPool(); 47         // 累加求和 0 -> 20000000L 48         CountTask task = new CountTask(0, 20000000L); 49         ForkJoinTask<Long> result = pool.submit(task); 50         System.out.println("sum result :" + result.get()); 51     } 52 }

ForkJoin 线程池应用一个无锁的栈来治理闲暇线程, 如果一个工作线程临时取不到可用的工作, 则可能被挂起.

挂起的线程将被压入由线程池保护的栈中, 待未来有工作可用时, 再从栈中唤醒这些线程.

好了,明天的内容根本就解说玩了,不晓得对大家有木有帮忙呢,其实这就是编程的魅力,集体认为,能够将所有的知识点进行残缺的贯通,造成一张宏大的常识图谱,相当的有成就感

对于下面设计模式的整顿,我除了工作中用到的之外,更多的是参考上面这份文档

既然说到并发,最初再给分享一些并发的材料吧

须要的敌人,获取形式没变,关注 + 转发后,私信“材料”即可查看获取形式

文件材料已上传公众号:Java 架构师联盟

git 地址:https://github.com/biws-byte/bws

退出移动版