并发编程专题三-线程的并发工具类
一、Fork-Join框架
1、分而治之
规模为N的问题,N<阈值,直接解决,N>阈值,将N分解为K个小规模子问题,子问题互相对立,与原问题形式相同,将子问题的解合并得到原问题的解
动态规范
2、工作密取 workStealing
就是在任务分割的时候,前面的任务执行可能会比后面的执行速度快,当前面的执行完,后面的还没执行的时候,执行完前面的任务的线程不会停止,而是从后面的任务的尾部取出子任务继续工作。Fork-Join就是实现了这样的机制。
Fork/Join使用的标准范式
代码如下:
import java.util.concurrent.ForkJoinPool; import java.util.concurrent.RecursiveTask; /** * @Auther: BlackKingW * @Date: 2019/4/14 12:09 * @Description: */ public class SumArray { private static class SumTask extends RecursiveTask<Integer>{ private final static int THRESHOLD = MakeArray.ARRAY_LENGTH/10; private int[] src; //表示我们要实际统计的数组 private int fromIndex;//开始统计的下标 private int toIndex;//统计到哪里结束的下标 public SumTask(int[] src, int fromIndex, int toIndex) { this.src = src; this.fromIndex = fromIndex; this.toIndex = toIndex; } @Override protected Integer compute() { if(toIndex-fromIndex < THRESHOLD) { int count = 0; for(int i=fromIndex;i<=toIndex;i++) { //SleepTools.ms(1); count = count + src[i]; } return count; }else { //fromIndex....mid....toIndex //1...................70....100 int mid = (fromIndex+toIndex)/2; SumTask left = new SumTask(src,fromIndex,mid); SumTask right = new SumTask(src,mid+1,toIndex); invokeAll(left,right); return left.join()+right.join(); } } } public static void main(String[] args) { ForkJoinPool pool = new ForkJoinPool(); int[] src = MakeArray.makeArray(); SumTask innerFind = new SumTask(src,0,src.length-1); long start = System.currentTimeMillis(); pool.invoke(innerFind);//同步调用 System.out.println("Task is Running....."); System.out.println("The count is "+innerFind.join() +" spend time:"+(System.currentTimeMillis()-start)+"ms"); } } public class MakeArray { //数组长度 public static final int ARRAY_LENGTH = 100000000; public static int[] makeArray() { //new一个随机数发生器 Random r = new Random(); int[] result = new int[ARRAY_LENGTH]; for(int i=0;i<ARRAY_LENGTH;i++){ //用随机数填充数组 result[i] = r.nextInt(ARRAY_LENGTH*3); } return result; } }
二、常用的并发工具类
1、CountDownLatch
作用:是一组线程等待其他的线程完成工作以后在执行,加强版join
await用来等待,countDown负责计数器的减一
代码示例:
import java.util.concurrent.CountDownLatch; /** * @Auther: BlackKingW * @Date: 2019/4/14 12:09 * @Description: */ public class UseCountDownLatch { static CountDownLatch latch = new CountDownLatch(6); //初始化线程(只有一步,有4个) private static class InitThread implements Runnable{ @Override public void run() { System.out.println("Thread_"+Thread.currentThread().getId() +" ready init work......"); latch.countDown();//初始化线程完成工作了,countDown方法只扣减一次; for(int i =0;i<2;i++) { System.out.println("Thread_"+Thread.currentThread().getId() +" ........continue do its work"); } } } //业务线程 private static class BusiThread implements Runnable{ @Override public void run() { try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } for(int i =0;i<3;i++) { System.out.println("BusiThread_"+Thread.currentThread().getId() +" do business-----"); } } } public static void main(String[] args) throws InterruptedException { //单独的初始化线程,初始化分为2步,需要扣减两次 new Thread(new Runnable() { @Override public void run() { SleepTools.ms(1); //休眠1s System.out.println("Thread_"+Thread.currentThread().getId() +" ready init work step 1st......"); latch.countDown();//每完成一步初始化工作,扣减一次 System.out.println("begin step 2nd......."); SleepTools.ms(1); System.out.println("Thread_"+Thread.currentThread().getId() +" ready init work step 2nd......"); latch.countDown();//每完成一步初始化工作,扣减一次 } }).start(); new Thread(new BusiThread()).start(); for(int i=0;i<=3;i++){ Thread thread = new Thread(new InitThread()); thread.start(); } latch.await(); System.out.println("Main do ites work........"); } }
2、CyclicBarrier
让一组线程达到某个屏障,被阻塞,一直到组内最后一个线程达到屏障时,屏障开放,所有被阻塞的线程会继续运行CyclicBarrier(int parties)
CyclicBarrier(int parties, Runnable barrierAction),屏障开放,barrierAction定义的任务会执行
代码示例:
import java.util.Map; import java.util.Random; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CyclicBarrier; /** * @Auther: BlackKingW * @Date: 2019/4/14 12:09 * @Description: */ public class UseCyclicBarrier { private static CyclicBarrier barrier = new CyclicBarrier(5,new CollectThread()); private static ConcurrentHashMap<String,Long> resultMap = new ConcurrentHashMap<>();//存放子线程工作结果的容器 public static void main(String[] args) { for(int i=0;i<=4;i++){ Thread thread = new Thread(new SubThread()); thread.start(); } } //负责屏障开放以后的工作 private static class CollectThread implements Runnable{ @Override public void run() { StringBuilder result = new StringBuilder(); for(Map.Entry<String,Long> workResult:resultMap.entrySet()){ result.append("["+workResult.getValue()+"]"); } System.out.println(" the result = "+ result); System.out.println("do other business........"); } } //工作线程 private static class SubThread implements Runnable{ @Override public void run() { long id = Thread.currentThread().getId();//线程本身的处理结果 resultMap.put(Thread.currentThread().getId()+"",id); Random r = new Random();//随机决定工作线程的是否睡眠 try { if(r.nextBoolean()) { Thread.sleep(2000+id); System.out.println("Thread_"+id+" ....do something "); } System.out.println(id+"....is await"); barrier.await(); Thread.sleep(1000+id); System.out.println("Thread_"+id+" ....do its business "); } catch (Exception e) { e.printStackTrace(); } } } }
CountDownLatch和CyclicBarrier辨析
1、countdownlatch放行由第三者控制,CyclicBarrier放行由一组线程本身控制
2、countdownlatch放行条件》=线程数,CyclicBarrier放行条件=线程数
3、Semaphore
控制同时访问某个特定资源的线程数量,用在流量控制
import java.sql.Connection; import java.util.LinkedList; import java.util.concurrent.Semaphore; /** * @Auther: BlackKingW * @Date: 2019/4/14 12:09 * @Description: */ public class DBPoolSemaphore { private final static int POOL_SIZE = 10; private final Semaphore useful,useless;//useful表示可用的数据库连接,useless表示已用的数据库连接 public DBPoolSemaphore() { this. useful = new Semaphore(POOL_SIZE); this.useless = new Semaphore(0); } //存放数据库连接的容器 private static LinkedList<Connection> pool = new LinkedList<Connection>(); //初始化池 static { for (int i = 0; i < POOL_SIZE; i++) { pool.addLast(SqlConnectImpl.fetchConnection()); } } /*归还连接*/ public void returnConnect(Connection connection) throws InterruptedException { if(connection!=null) { System.out.println("当前有"+useful.getQueueLength()+"个线程等待数据库连接!!" +"可用连接数:"+useful.availablePermits()); useless.acquire(); synchronized (pool) { pool.addLast(connection); } useful.release(); } } /*从池子拿连接*/ public Connection takeConnect() throws InterruptedException { useful.acquire(); Connection conn; synchronized (pool) { conn = pool.removeFirst(); } useless.release(); return conn; } }
4、Exchange
两个线程间的数据交换, 用的比较少
代码示例:
import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.Exchanger; /** * @Auther: BlackKingW * @Date: 2019/4/14 12:09 * @Description: */ public class UseExchange { private static final Exchanger<Set<String>> exchange = new Exchanger<Set<String>>(); public static void main(String[] args) { //第一个线程 new Thread(new Runnable() { @Override public void run() { Set<String> setA = new HashSet<String>();//存放数据的容器 try { /*添加数据 * set.add(.....) * */ setA = exchange.exchange(setA);//交换set /*处理交换后的数据*/ } catch (InterruptedException e) { } } }).start(); //第二个线程 new Thread(new Runnable() { @Override public void run() { Set<String> setB = new HashSet<String>();//存放数据的容器 try { /*添加数据 * set.add(.....) * set.add(.....) * */ setB = exchange.exchange(setB);//交换set /*处理交换后的数据*/ } catch (InterruptedException e) { } } }).start(); } }
5、Callable、Future和FutureTask
类之间的关系
isDone,结束,正常还是异常结束,或者自己取消,返回true;
isCancelled 任务完成前被取消,返回true;
cancel(boolean):
- 任务还没开始,返回false
- 任务已经启动,cancel(true),中断正在运行的任务,中断成功,返回true,cancel(false),不会去中断已经运行的任务
- 任务已经结束,返回false
代码示例:
import java.util.Random; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.FutureTask; /** * @Auther: BlackKingW * @Date: 2019/4/14 12:09 * @Description: */ public class UseFuture { /*实现Callable接口,允许有返回值*/ private static class UseCallable implements Callable<Integer>{ private int sum; @Override public Integer call() throws Exception { System.out.println("Callable子线程开始计算"); Thread.sleep(2000); for(int i=0;i<5000;i++) { sum = sum+i; } System.out.println("Callable子线程计算完成,结果="+sum); return sum; } } public static void main(String[] args) throws InterruptedException, ExecutionException { UseCallable useCallable = new UseCallable(); FutureTask<Integer> futureTask = new FutureTask<Integer>(useCallable); new Thread(futureTask).start(); Random r = new Random(); SleepTools.second(1); if(r.nextBoolean()) {//随机决定是获得结果还是终止任务 System.out.println("Get UseCallable result = "+futureTask.get()); }else { System.out.println("中断计算"); futureTask.cancel(true); } } }
场景举例:包含图片和文字的文档的处理:图片(云上),可以用future去取图片,主线程继续解析文字。

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
springboot shiro实现权限管理
记得第一次使用shiro是在才入行遇到公司的第一个框架,当时并不知道这是什么,或者说根本就没有安全框架的概念,在慢慢实践中,也对这个有了一定的了解,于是在网上找各种资料学习,了解。记得那时候比较有没的相关博客就是这个了,相信学习shiro的人很多都度过他的博客,内容也比较详细,示例也非常丰富。 开始使用shiro时,是与spring进行整合,可以看这里,当时没有实现太多功能,但是把一些外围的模块都已经实现,而且能够进行多realm匹配。在写这次博客前,也看了下renren-security,同样有很多可以借鉴的地方,所有既然看了这么多,那么自然要把功能做的完善一些了。 当然,此次的系统基础还是上次的springboot jpa搭建开发环境,我会在此基础上进行更新与扩展。 首先还是对shiro有一个初步的认识,当然这些认识都是收集与网络。 首先在学习这个框架时,都基本都会看到的图,那么我们需要了解的无非围绕着shiro里面这些核心的模块,具体是干什么,在什么时候,怎么去用,了解到这些之后,我们就可以按照自己的想法与设计,在适当的场合与环境下正确的使用该框架为我们提供的功能。 Authen...
- 下一篇
死磕 java集合之ConcurrentSkipListMap源码分析——发现个bug
前情提要 点击链接查看“跳表”详细介绍。 拜托,面试别再问我跳表了! 简介 跳表是一个随机化的数据结构,实质就是一种可以进行二分查找的有序链表。 跳表在原有的有序链表上面增加了多级索引,通过索引来实现快速查找。 跳表不仅能提高搜索性能,同时也可以提高插入和删除操作的性能。 存储结构 跳表在原有的有序链表上面增加了多级索引,通过索引来实现快速查找。 源码分析 主要内部类 内部类跟存储结构结合着来看,大概能预测到代码的组织方式。 // 数据节点,典型的单链表结构 static final class Node<K,V> { final K key; // 注意:这里value的类型是Object,而不是V // 在删除元素的时候value会指向当前元素本身 volatile Object value; volatile Node<K,V> next; Node(K key, Object value, Node<K,V> next) { this.key = key; this.value = value; this.next = next; } Nod...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- CentOS7,8上快速安装Gitea,搭建Git服务器
- CentOS8编译安装MySQL8.0.19
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- SpringBoot2整合Thymeleaf,官方推荐html解决方案
- CentOS8,CentOS7,CentOS6编译安装Redis5.0.7
- CentOS7安装Docker,走上虚拟化容器引擎之路
- SpringBoot2全家桶,快速入门学习开发网站教程
- Eclipse初始化配置,告别卡顿、闪退、编译时间过长
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- CentOS7编译安装Gcc9.2.0,解决mysql等软件编译问题