文档章节

线程池工作窃取实例

go4it
 go4it
发布于 2017/09/11 21:32
字数 484
阅读 25
收藏 0

本文主要来展示一下简版的work stealing线程池的实现。

Executors

Executors默认提供了几个工厂方法

/**
     * 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<T> {

    private static final Logger LOGGER = LoggerFactory.getLogger(WorkStealingChannel.class);

    BlockingDeque<T>[] managedQueues;

    AtomicLongMap<Integer> stat = AtomicLongMap.create();

    public WorkStealingChannel() {
        int nCPU = Runtime.getRuntime().availableProcessors();
        int queueCount = nCPU / 2 + 1;
        managedQueues = new LinkedBlockingDeque[queueCount];
        for(int i=0;i<queueCount;i++){
            managedQueues[i] = new LinkedBlockingDeque<T>();
        }
    }

    public void put(T item) throws InterruptedException {
        int targetIndex = Math.abs(item.hashCode() % managedQueues.length);
        BlockingQueue<T> 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<Integer> getStat() {
        return stat;
    }
}

测试实例

public class WorkStealingDemo {

    static final WorkStealingChannel<String> channel = 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}

© 著作权归作者所有

共有 人打赏支持
go4it
粉丝 68
博文 765
码字总数 571053
作品 0
深圳
Java Fork/Join 框架

简介 从JDK1.7开始,Java提供Fork/Join框架用于并行执行任务,它的思想就是讲一个大任务分割成若干小任务,最终汇总每个小任务的结果得到这个大任务的结果。 这种思想和MapReduce很像(input...

Java工程师-Distance
05/23
0
0
Java7中的Fork/Join框架,到底在什么情况下才会采用工作窃取算法?

在学习《Java 7并发编程实战手册.pdf》第五章时,有一段话让我思考很久都很难理解,测试结果也不准确和信服,还请大家指教。 “当你使用同步方法,调用这些方法(比如:invokeAll()方法)的任...

JarvisZhu
2016/11/19
547
1
Java进阶篇:多线程并发实践

关于作者 郭孝星,程序员,吉他手,主要从事Android平台基础架构方面的工作,欢迎交流技术方面的问题,可以去我的Github提issue或者发邮件至guoxiaoxingse@163.com与我交流。 文章目录 一 线...

郭孝星
2017/09/22
0
0
《Java并发编程的艺术》笔记

《Java并发编程的艺术》 方腾飞 魏鹏 程晓明 著 笔记: 第一章:并发编程的挑战 1、上下文切换: 系统的多线程并发执行,就是CPU通过给每个线程分配CPU时间片来实现的。当一个线程的时间用完...

tsmyk0715
2016/08/20
366
0
java 中的Fork/Join框架

什么是Fork/Join框架 Fork/Join框架是一个用于并行执行任务的框架,是一个把大任务分割成若干个小任务,最终汇总每个小任务结果后得到大任务结果的框架。 我们再通过Fork和Join这两个单词来理...

hgfgoodcreate
2016/07/03
88
1

没有更多内容

加载失败,请刷新页面

加载更多

初级开发-编程题

` public static void main(String[] args) { System.out.println(changeStrToUpperCase("user_name_abc")); System.out.println(changeStrToLowerCase(changeStrToUpperCase("user_name_abc......

小池仔
今天
4
0
现场看路演了!

HiBlock
昨天
12
0
Rabbit MQ基本概念介绍

RabbitMQ介绍 • RabbitMQ是一个消息中间件,是一个很好用的消息队列框架。 • ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的s...

寰宇01
昨天
9
0
官方精简版Windows10:微软自己都看不过去了

微软宣布,该公司正在寻求解决方案,以减轻企业客户的Windows 10规模。该公司声称,企业客户下载整个Windows 10文件以更新设备既费钱又费时。 微软宣布,该公司正在寻求解决方案,以减轻企业...

linux-tao
昨天
14
0
TypeScript基础入门之JSX(二)

转发 TypeScript基础入门之JSX(二) 属性类型检查 键入检查属性的第一步是确定元素属性类型。 内在元素和基于价值的元素之间略有不同。 对于内部元素,它是JSX.IntrinsicElements上的属性类型...

durban
昨天
9
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部