java手写简易版自定义线程池 V1

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class DdqThreadPool {
    private static final class DdqThread extends Thread {
        private Runnable runnable;
        private final ReentrantLock reentrantLock = new ReentrantLock();
        private final Condition condition = reentrantLock.newCondition();

        public void setRunnable(Runnable runnable) {
            try {
                reentrantLock.lock();
                this.runnable = runnable;
                condition.signalAll();
            } catch (Exception exception) {
                throw new RuntimeException(exception);
            } finally {
                reentrantLock.unlock();
            }
        }

        @Override
        public void run() {
            while (true) {
                try {
                    reentrantLock.lock();
                    while (runnable == null) condition.await();
                    runnable.run();
                    runnable = null;
                } catch (Exception exception) {
                    throw new RuntimeException(exception);
                } finally {
                    reentrantLock.unlock();
                }
            }

        }
    }

    private List<DdqThread> threads;

    private AtomicInteger threadSize;

    private DdqThreadPool(List<DdqThread> ddqThreads) {
        this.threads = ddqThreads;
        this.threadSize = new AtomicInteger(ddqThreads.size());
    }

    public static DdqThreadPool newDdqThreadPool(int maxThreadSize) {
        if (maxThreadSize < 1) throw new IllegalArgumentException("线程数量必须大于0");
        List<DdqThread> ddqThreads = new LinkedList<>();
        for (int idx = 0; idx < maxThreadSize; idx++) {
            DdqThread ddqThread = new DdqThread();
            ddqThread.start();
            ddqThreads.add(ddqThread);
        }
        return new DdqThreadPool(ddqThreads);
    }

    public void execute(Runnable runnable) {
        if (runnable == null) throw new IllegalArgumentException("runnable不能为null");
        int oldThreadSize1 = threadSize.get();
        if (oldThreadSize1 < 1) return;
        if (!threadSize.compareAndSet(oldThreadSize1, oldThreadSize1 - 1)) return;
        DdqThread idleThread = threads.remove(0);
        idleThread.setRunnable(runnable);
        //以下三段代码可能还存在bug，因为idleThread.setRunnable(runnable)后，由于是异步执行，可能会存在着线程没执行完，但是忙碌线程又被添加回了可用线程集合
        int oldThreadSize2;
        while (!threadSize.compareAndSet((oldThreadSize2 = threadSize.get()), oldThreadSize2 + 1)) ;
        threads.add(idleThread);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

V2版本请点击此处