JDK25 并发

1
2
3
4
5

 高层:     HeadIndex -> Index -> Index -> null
                 |          |
 中间层:     Index ------> Index ------> null
                 |          |
 底层:        Node  -->  Node  -->  Node  --> null

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

public boolean add(E e) {
        synchronized (lock) {
            Object[] es = getArray();
            int len = es.length;
            es = Arrays.copyOf(es, len + 1);
            es[len] = e;
            setArray(es);
            return true;
        }
    }

1

private static final Unsafe U = Unsafe.getUnsafe();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

public class Singleton {
    private static volatile Singleton instance;

    public static Singleton getInstance() {
        if (instance == null) {
            synchronized (Singleton.class) {
                if (instance == null) {
                    instance = new Singleton();
                }
            }
        }
        return instance;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

public class FlagExample {
    private volatile boolean running = true;

    public void start() {
        new Thread(() -> {
            while (running) {
                // do something
            }
        }).start();
    }
  //可以从外部停止线程
    public void stop() {
        running = false;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

public class MyLock extends AbstractQueuedSynchronizer {
    @Override
    protected boolean tryAcquire(int acquires) {
        return compareAndSetState(0, 1);
    }
//解锁是一个非常复杂的操作，这里演示就直接让锁释放成功
    @Override
    protected boolean tryRelease(int releases) {
        setState(0);
        return true;
    }

    public void lock() {
        acquire(1);
    }

    public void unlock() {
        release(1);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

Lock lock = new ReentrantLock();
Condition notEmpty = lock.newCondition();
lock.lock();
try {
    while (queue.isEmpty()) {
        notEmpty.await(); // 消费者等待
    }
    // 处理队列中的元素
} finally {
    lock.unlock();
}

1
2
3
4
5
6
7

Semaphore semaphore = new Semaphore(3); // 允许 3 个线程并发访问
try {
    semaphore.acquire(); // 获取许可
    // 访问共享资源
} finally {
    semaphore.release(); // 释放许可
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {
    private static final CyclicBarrier BARRIER = new CyclicBarrier(4);

    public static void main(String[] args) {
        for (int i = 0; i < 4; i++) {
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + " is waiting at barrier");
                    BARRIER.await(); // 等待所有线程到达屏障点
                    System.out.println(Thread.currentThread().getName() + " has crossed the barrier");
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

 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

import java.util.concurrent.Phaser;

public class PhaserExample {
    public static void main(String[] args) {
        int threads = 3;
      //初始规定三个线程
        Phaser phaser = new Phaser(threads);

        for (int i = 0; i < threads; i++) {
            final int threadNum = i;
            new Thread(() -> {
                System.out.println("Thread " + threadNum + " starting phase " + phaser.getPhase());
                // 模拟工作
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Thread " + threadNum + " arriving at phase " + phaser.getPhase());
                phaser.arriveAndAwaitAdvance(); // 到达同步点并等待

                System.out.println("Thread " + threadNum + " starting phase " + phaser.getPhase());
                // 模拟下一阶段工作
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Thread " + threadNum + " arriving at phase " + phaser.getPhase());
                phaser.arriveAndDeregister(); // 到达同步点并解注册
            }).start();
        }
    }
}

 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
77
78

class DynamicThreadPoolWithPhaser {
    public static void main(String[] args) throws InterruptedException {
        // 创建 Phaser
        Phaser phaser = new Phaser() {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("Advancing to phase " + (phase + 1) + " with " + registeredParties + " parties remaining");
                return registeredParties == 0; // 当没有注册线程时终止
            }
        };

        // 阶段 1：启动 4 个线程
        Thread[] threads = new Thread[6]; // 预分配最大线程数
        System.out.println("Starting Phase 1 with 4 threads");
        for (int i = 0; i < 4; i++) {
            phaser.register(); // 为每个线程注册
            threads[i] = new Thread(new Worker(phaser, i, i < 2 ? 2 : 3), "Worker-" + i);
            threads[i].start();
        }
        phaser.arriveAndAwaitAdvance(); // 主线程等待 Phase 1 完成

        // 阶段 2：增加到 6 个线程
        System.out.println("Starting Phase 2 with 6 threads");
        for (int i = 4; i < 6; i++) {
            phaser.register(); // 为新线程注册
            threads[i] = new Thread(new Worker(phaser, i, 3), "Worker-" + i);
            threads[i].start();
        }
        phaser.arriveAndAwaitAdvance(); // 主线程等待 Phase 2 完成

        // 阶段 3：减少到 2 个线程（由 Worker 内部逻辑控制）
        System.out.println("Starting Phase 3 with 2 threads");
        phaser.arriveAndAwaitAdvance(); // 主线程等待 Phase 3 完成

        // 等待所有线程结束
        for (Thread t : threads) {
            if (t != null) {
                t.join();
            }
        }
        System.out.println("All phases completed. Final phase: " + phaser.getPhase() + ", terminated: " + phaser.isTerminated());
    }
}

class Worker implements Runnable {
    private final Phaser phaser;
    private final int id;
    private final int maxPhase; // 最大参与阶段

    public Worker(Phaser phaser, int id, int maxPhase) {
        this.phaser = phaser;
        this.id = id;
        this.maxPhase = maxPhase;
    }

    @Override
    public void run() {
        try {
            for (int p = 0; p <= maxPhase; p++) {
                if (phaser.isTerminated()) {
                    System.out.println("Worker-" + id + " stopped: Phaser terminated");
                    return;
                }
                System.out.println("Worker-" + id + " starting phase " + p);
                Thread.sleep(ThreadLocalRandom.current().nextInt(1000, 2000)); // 模拟工作
                System.out.println("Worker-" + id + " completed phase " + p);
                if (p == maxPhase) {
                    phaser.arriveAndDeregister(); // 到达最大阶段后解注册
                } else {
                    phaser.arriveAndAwaitAdvance(); // 等待其他线程
                }
            }
        } catch (InterruptedException e) {
            System.out.println("Worker-" + id + " interrupted: " + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }
}

 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

class DynamicThreadPool {
    public static void main(String[] args) throws InterruptedException {
        // 创建 ThreadPoolExecutor，初始核心线程数为 5，最大线程数为 10
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5, // 核心线程数
                10, // 最大线程数
                60L, // 空闲线程存活时间
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>() // 任务队列
        );

        // 打印初始线程池状态
        System.out.println("初始核心线程数: " + executor.getCorePoolSize());
        System.out.println("初始线程池大小: " + executor.getPoolSize());

        // 提交 10 个任务
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.execute(() -> {
                System.out.println("任务 " + taskId + " 开始执行，线程: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000); // 模拟工作
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("任务 " + taskId + " 完成，线程: " + Thread.currentThread().getName());
            });
        }

        // 等待 3 秒，让部分任务运行并排队
        Thread.sleep(3000);

        // 打印当前线程池状态
        System.out.println("3 秒后，线程池大小: " + executor.getPoolSize());
        System.out.println("活跃线程数: " + executor.getActiveCount());
        System.out.println("已完成任务数: " + executor.getCompletedTaskCount());
        System.out.println("排队任务数: " + executor.getQueue().size());

        // 将核心线程数动态调整为 10
        executor.setCorePoolSize(10);
        System.out.println("核心线程数调整为: " + executor.getCorePoolSize());

        // 等待 2 秒，让新线程启动
        Thread.sleep(2000);

        // 打印调整后的线程池状态
        System.out.println("调整核心线程数后，线程池大小: " + executor.getPoolSize());
        System.out.println("活跃线程数: " + executor.getActiveCount());
        System.out.println("已完成任务数: " + executor.getCompletedTaskCount());
        System.out.println("排队任务数: " + executor.getQueue().size());

        // 提交另外 5 个任务
        for (int i = 10; i < 15; i++) {
            final int taskId = i;
            executor.execute(() -> {
                System.out.println("任务 " + taskId + " 开始执行，线程: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000); // 模拟工作
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("任务 " + taskId + " 完成，线程: " + Thread.currentThread().getName());
            });
        }

        // 关闭线程池
        executor.shutdown();
        System.out.println("线程池关闭已启动");

        // 等待所有任务完成
        executor.awaitTermination(1, TimeUnit.HOURS);
        System.out.println("所有任务完成");
    }
}

 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

import java.util.concurrent.Exchanger;

public class ProducerConsumerExample {
    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();
        new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    String data = "Data " + i;
                    System.out.println("Producer sending: " + data);
                    String received = exchanger.exchange(data);
                    System.out.println("Producer received: " + received);
                    Thread.sleep(1000); // 模拟生产时间
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
        new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    String ack = "Ack " + i;
                    System.out.println("Consumer sending: " + ack);
                    String received = exchanger.exchange(ack);
                    System.out.println("Consumer received: " + received);
                    Thread.sleep(1200); // 模拟消费时间
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }
}

 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

    /**
     * 用给定的初始参数创建一个新的ThreadPoolExecutor。
     */
    public ThreadPoolExecutor(int corePoolSize,//线程池的核心线程数量
                              int maximumPoolSize,//线程池的最大线程数
                              long keepAliveTime,//当线程数大于核心线程数时，多余的空闲线程存活的最长时间
                              TimeUnit unit,//时间单位
                              BlockingQueue<Runnable> workQueue,//任务队列，用来储存等待执行任务的队列
                              ThreadFactory threadFactory,//线程工厂，用来创建线程，一般默认即可
                              RejectedExecutionHandler handler//拒绝策略，当提交的任务过多而不能及时处理时，我们可以定制策略来处理任务
                               ) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

 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

   // 存放线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)
   private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    private static int workerCountOf(int c) {
        return c & CAPACITY;
    }
    //任务队列
    private final BlockingQueue<Runnable> workQueue;

    public void execute(Runnable command) {
        // 如果任务为null，则抛出异常。
        if (command == null)
            throw new NullPointerException();
        // ctl 中保存的线程池当前的一些状态信息
        int c = ctl.get();

        //  下面会涉及到 3 步 操作
        // 1.首先判断当前线程池中执行的任务数量是否小于 corePoolSize
        // 如果小于的话，通过addWorker(command, true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 2.如果当前执行的任务数量大于等于 corePoolSize 的时候就会走到这里，表明创建新的线程失败。
        // 通过 isRunning 方法判断线程池状态，线程池处于 RUNNING 状态并且队列可以加入任务，该任务才会被加入进去
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            // 再次获取线程池状态，如果线程池状态不是 RUNNING 状态就需要从任务队列中移除任务，并尝试判断线程是否全部执行完毕。同时执行拒绝策略。
            if (!isRunning(recheck) && remove(command))
                reject(command);
                // 如果当前工作线程数量为0，新创建一个线程并执行。
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //3. 通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
        // 传入 false 代表增加线程时判断当前线程数是否少于 maxPoolSize
        //如果addWorker(command, false)执行失败，则通过reject()执行相应的拒绝策略的内容。
        else if (!addWorker(command, false))
            reject(command);
    }

  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
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

    // 全局锁，并发操作必备
    private final ReentrantLock mainLock = new ReentrantLock();
    // 跟踪线程池的最大大小，只有在持有全局锁mainLock的前提下才能访问此集合
    private int largestPoolSize;
    // 工作线程集合，存放线程池中所有的（活跃的）工作线程，只有在持有全局锁mainLock的前提下才能访问此集合
    private final HashSet<Worker> workers = new HashSet<>();
    //获取线程池状态
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    //判断线程池的状态是否为 Running
    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }


    /**
     * 添加新的工作线程到线程池
     * @param firstTask 要执行
     * @param core参数为true的话表示使用线程池的基本大小，为false使用线程池最大大小
     * @return 添加成功就返回true否则返回false
     */
   private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            //这两句用来获取线程池的状态
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
               //获取线程池中工作的线程的数量
                int wc = workerCountOf(c);
                // core参数为false的话表明队列也满了，线程池大小变为 maximumPoolSize
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
               //原子操作将workcount的数量加1
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                // 如果线程的状态改变了就再次执行上述操作
                c = ctl.get();
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
        // 标记工作线程是否启动成功
        boolean workerStarted = false;
        // 标记工作线程是否创建成功
        boolean workerAdded = false;
        Worker w = null;
        try {

            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
              // 加锁
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                   //获取线程池状态
                    int rs = runStateOf(ctl.get());
                   //rs < SHUTDOWN 如果线程池状态依然为RUNNING,并且线程的状态是存活的话，就会将工作线程添加到工作线程集合中
                  //(rs=SHUTDOWN && firstTask == null)如果线程池状态小于STOP，也就是RUNNING或者SHUTDOWN状态下，同时传入的任务实例firstTask为null，则需要添加到工作线程集合和启动新的Worker
                   // firstTask == null证明只新建线程而不执行任务
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                       //更新当前工作线程的最大容量
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                      // 工作线程是否启动成功
                        workerAdded = true;
                    }
                } finally {
                    // 释放锁
                    mainLock.unlock();
                }
                //// 如果成功添加工作线程，则调用Worker内部的线程实例t的Thread#start()方法启动真实的线程实例
                if (workerAdded) {
                    t.start();
                  /// 标记线程启动成功
                    workerStarted = true;
                }
            }
        } finally {
           // 线程启动失败，需要从工作线程中移除对应的Worker
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

1

最佳线程数 = N（CPU 核心数）∗（1+WT（线程等待时间）/ST（线程计算时间））`，其中 `WT（线程等待时间）=线程运行总时间 - ST（线程计算时间）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

public class InheritableThreadLocalDemo {
    public static void main(String[] args) {
        ThreadLocal<String> ThreadLocal = new ThreadLocal<>();
        ThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>();
        ThreadLocal.set("父类数据:threadLocal");
        inheritableThreadLocal.set("父类数据:inheritableThreadLocal");

        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("子线程获取父类ThreadLocal数据：" + ThreadLocal.get());
                System.out.println("子线程获取父类inheritableThreadLocal数据：" + inheritableThreadLocal.get());
            }
        }).start();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

private void init(ThreadGroup g, Runnable target, String name,
                      long stackSize, AccessControlContext acc,
                      boolean inheritThreadLocals) {
    if (name == null) {
        throw new NullPointerException("name cannot be null");
    }

    if (inheritThreadLocals && parent.inheritableThreadLocals != null)
        this.inheritableThreadLocals =
            ThreadLocal.createInheritedMap(parent.inheritableThreadLocals);
    this.stackSize = stackSize;
    tid = nextThreadID();
}

 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

public class ExecutorTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        用线程池计算从1到10000的累加
        ThreadPoolExecutor threadPoolExecutor= new ThreadPoolExecutor(10,100,1000, TimeUnit.MILLISECONDS,new LinkedBlockingDeque<>());
//        分两个线程计算，计算速度应对是单线程的两倍
        Future<Integer> future1 = threadPoolExecutor.submit(() -> {
            int sum = 0;
            for (int i = 0; i < 5000; i++) {
                sum += i;
            }
            return sum;
        });
        Future<Integer> future2 = threadPoolExecutor.submit(() -> {
            int sum = 0;
            for (int i = 5000; i <= 10000; i++) {
                sum += i;
            }
            return sum;
        });
        Integer s1 = future1.get();
        Integer s2 = future2.get();
        threadPoolExecutor.shutdown();
        System.out.println(s1+s2);
    }
}

 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

public class SumTask extends RecursiveTask<Long> {

    private final int l;
    private final int r;

    public SumTask(int l, int r) {
        this.l = l;
        this.r = r;
    }
    @Override
    protected Long compute() {
        long res=0;
//        判断任务是否拆封
        if(r-l<100){
            for (int i = l; i <=r; i++) {
                res+=i;
            }
            return res;
        }
//        需要被拆封的情况：经典二分查找
        int mid=(l+r)/2;
        SumTask leftTask=new SumTask(l,mid);
        SumTask rightTask=new SumTask(mid+1,r);
        leftTask.fork();
        rightTask.fork();
        long left=leftTask.join();
        long right=rightTask.join();
        return left+right;
    }
}