摘要:序本文主要来展示一下简版的线程池的实现。默认提供了几个工厂方法思路主要用到的是双端队列,不过这里我们粗糙的实现的话,也可以不用到。测试实例输出从数据来看,还是相对均匀的。
序
本文主要来展示一下简版的work stealing线程池的实现。
ExecutorsExecutors默认提供了几个工厂方法
/** * Creates a thread pool that maintains enough threads to support * the given parallelism level, and may use multiple queues to * reduce contention. The parallelism level corresponds to the * maximum number of threads actively engaged in, or available to * engage in, task processing. The actual number of threads may * grow and shrink dynamically. A work-stealing pool makes no * guarantees about the order in which submitted tasks are * executed. * * @param parallelism the targeted parallelism level * @return the newly created thread pool * @throws IllegalArgumentException if {@code parallelism <= 0} * @since 1.8 */ public static ExecutorService newWorkStealingPool(int parallelism) { return new ForkJoinPool (parallelism, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true); } /** * Creates a work-stealing thread pool using all * {@link Runtime#availableProcessors available processors} * as its target parallelism level. * @return the newly created thread pool * @see #newWorkStealingPool(int) * @since 1.8 */ public static ExecutorService newWorkStealingPool() { return new ForkJoinPool (Runtime.getRuntime().availableProcessors(), ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true); }思路
ForkJoinPool主要用到的是双端队列,不过这里我们粗糙的实现的话,也可以不用到deque。
public class WorkStealingChannel{ private static final Logger LOGGER = LoggerFactory.getLogger(WorkStealingChannel.class); BlockingDeque [] managedQueues; AtomicLongMap stat = AtomicLongMap.create(); public WorkStealingChannel() { int nCPU = Runtime.getRuntime().availableProcessors(); int queueCount = nCPU / 2 + 1; managedQueues = new LinkedBlockingDeque[queueCount]; for(int i=0;i (); } } public void put(T item) throws InterruptedException { int targetIndex = Math.abs(item.hashCode() % managedQueues.length); BlockingQueue targetQueue = managedQueues[targetIndex]; targetQueue.put(item); } public T take() throws InterruptedException { int rdnIdx = ThreadLocalRandom.current().nextInt(managedQueues.length); int idx = rdnIdx; while (true){ idx = idx % managedQueues.length; T item = null; if(idx == rdnIdx){ item = managedQueues[idx].poll(); }else{ item = managedQueues[idx].pollLast(); } if(item != null){ LOGGER.info("take ele from queue {}",idx); stat.addAndGet(idx,1); return item; } idx++; if(idx == rdnIdx){ break; } } //走完一轮没有,则随机取一个等待 LOGGER.info("wait for queue:{}",rdnIdx); stat.addAndGet(rdnIdx,1); return managedQueues[rdnIdx].take(); } public AtomicLongMap getStat() { return stat; } }
这里根据cpu的数量建立了几个deque,然后每次put的时候,根据hashcode取模放到对应的队列。然后获取的时候,先从随机一个队列取,没有的话,再robbin round取其他队列的,还没有的话,则阻塞等待指定队列的元素。
测试实例
public class WorkStealingDemo { static final WorkStealingChannelchannel = new WorkStealingChannel<>(); static volatile boolean running = true; static class Producer extends Thread{ @Override public void run() { while(running){ try { channel.put(UUID.randomUUID().toString()); } catch (InterruptedException e) { e.printStackTrace(); } } } } static class Consumer extends Thread{ @Override public void run() { while(running){ try { String value = channel.take(); System.out.println(value); } catch (InterruptedException e) { e.printStackTrace(); } } } } public static void stop(){ running = false; System.out.println(channel.getStat()); } public static void main(String[] args) throws InterruptedException { int nCPU = Runtime.getRuntime().availableProcessors(); int consumerCount = nCPU / 2 + 1; for (int i = 0; i < nCPU; i++) { new Producer().start(); } for (int i = 0; i < consumerCount; i++) { new Consumer().start(); } Thread.sleep(30*1000); stop(); } }
输出
{0=660972, 1=660613, 2=661537, 3=659846, 4=659918}
从数据来看,还是相对均匀的。
文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。
转载请注明本文地址:https://www.ucloud.cn/yun/70405.html
执行器 在前面的所有示例中,由新的线程(由其Runnable对象定义)和线程本身(由Thread对象定义)完成的任务之间存在紧密的联系,这适用于小型应用程序,但在大型应用程序中,将线程管理和创建与应用程序的其余部分分开是有意义的,封装这些函数的对象称为执行器,以下小节详细描述了执行器。 执行器接口定义三个执行器对象类型。 线程池是最常见的执行器实现类型。 Fork/Join是一个利用多个处理器的...
摘要:大多数待遇丰厚的开发职位都要求开发者精通多线程技术并且有丰富的程序开发调试优化经验,所以线程相关的问题在面试中经常会被提到。掌握了这些技巧,你就可以轻松应对多线程和并发面试了。进入等待通行准许时,所提供的对象。 最近看到网上流传着,各种面试经验及面试题,往往都是一大堆技术题目贴上去,而没有答案。 不管你是新程序员还是老手,你一定在面试中遇到过有关线程的问题。Java语言一个重要的特点就...
摘要:同时,它会通过的方法将自己注册到线程池中。线程池中的每个工作线程都有一个自己的任务队列,工作线程优先处理自身队列中的任务或顺序,由线程池构造时的参数决定,自身队列为空时,以的顺序随机窃取其它队列中的任务。 showImg(https://segmentfault.com/img/bVbizJb?w=1802&h=762); 本文首发于一世流云的专栏:https://segmentfau...
摘要:方法接受对象数组作为参数,目标是对数组进行升序排序。创建一个对象,并调用方法将它提交给线程池。此排序算法不直接返回结果给调用方,因此基于类。 分支/合并框架 说明 重点是那个浮点数数组排序的例子,从主函数展开,根据序号看 1、GitHub代码欢迎star。你们轻轻的一点,对我鼓励特大,我有一个习惯,看完别人的文章是会点赞的。 2、个人认为学习语言最好的方式就是模仿、思考别人为什么这么写...
摘要:并不会为每个任务都创建工作线程,而是根据实际情况构造线程池时的参数确定是唤醒已有空闲工作线程,还是新建工作线程。 showImg(https://segmentfault.com/img/bVbiYSP?w=1071&h=707); 本文首发于一世流云的专栏:https://segmentfault.com/blog... 一、引言 前一章——Fork/Join框架(1) 原理,我们...
阅读 1579·2021-10-18 13:35
阅读 2359·2021-10-09 09:44
阅读 812·2021-10-08 10:05
阅读 2706·2021-09-26 09:47
阅读 3560·2021-09-22 15:22
阅读 427·2019-08-29 12:24
阅读 1992·2019-08-29 11:06
阅读 2851·2019-08-26 12:23