摘要：序本文主要來(lái)展示一下簡(jiǎn)版的線程池的實(shí)現(xiàn)。默認(rèn)提供了幾個(gè)工廠方法思路主要用到的是雙端隊(duì)列，不過(guò)這里我們粗糙的實(shí)現(xiàn)的話，也可以不用到。測(cè)試實(shí)例輸出從數(shù)據(jù)來(lái)看，還是相對(duì)均勻的。

本文主要來(lái)展示一下簡(jiǎn)版的work stealing線程池的實(shí)現(xiàn)。

Executors默認(rèn)提供了幾個(gè)工廠方法

/**
     * 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主要用到的是雙端隊(duì)列，不過(guò)這里我們粗糙的實(shí)現(xiàn)的話，也可以不用到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;
            }
        }

        //走完一輪沒(méi)有,則隨機(jī)取一個(gè)等待
        LOGGER.info("wait for queue:{}",rdnIdx);
        stat.addAndGet(rdnIdx,1);
        return managedQueues[rdnIdx].take();
    }

    public AtomicLongMap getStat() {
        return stat;
    }
}

這里根據(jù)cpu的數(shù)量建立了幾個(gè)deque，然后每次put的時(shí)候，根據(jù)hashcode取模放到對(duì)應(yīng)的隊(duì)列。然后獲取的時(shí)候，先從隨機(jī)一個(gè)隊(duì)列取，沒(méi)有的話，再robbin round取其他隊(duì)列的，還沒(méi)有的話，則阻塞等待指定隊(duì)列的元素。

測(cè)試實(shí)例

public class WorkStealingDemo {

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

從數(shù)據(jù)來(lái)看，還是相對(duì)均勻的。