Java 并发核心：单例模式、生产者消费者、定时器及线程池实现

实现细节：

1. 通过 synchronized 加锁解决线程安全问题。
2. 外层 if 判断，减少锁的开销。
3. volatile 防止指令重排序 (避免半初始化对象)。

特点：只有在真正需要时才创建实例，减少系统启动时的资源占用，资源利用率高。

缺点：若未正确使用同步机制 (如 synchronized 或 volatile)，可能导致多线程环境下实例被多次创建，线程安全实现复杂。

1.4 懒汉模式半初始化

Java 对象初始化流程

1. 内存分配阶段：当使用 new 关键字创建对象时，JVM 会在堆内存中为该对象分配空间。
2. 对象初始化阶段：对象内存分配完成后，开始执行初始化。
3. 引用赋值阶段：内存分配完成后，JVM 将分配的内存地址赋值给引用变量。

Java 对象初始化流程 (指令重排序后)

1. 内存分配阶段：当使用 new 关键字创建对象时，JVM 会在堆内存中为该对象分配空间。
2. 引用赋值阶段：内存分配完成后，JVM 将分配的内存地址赋值给引用变量。
3. 对象初始化阶段：对象内存分配完成后，开始执行初始化。

1.5 懒汉/饿汉优缺点对比

2. 生产者/消费者模式

2.1 概述

生产者/消费者模式 (Producer/consumer model)：用于协调多个线程或进程之间的任务分配与数据处理。生产者负责生成数据或任务，消费者负责处理这些数据或任务，二者通过共享的缓冲区 (队列) 进行解耦，避免直接依赖。

核心作用

1. 解耦生产与消费逻辑：生产者仅负责生成数据并放入缓冲区，消费者仅从缓冲区获取数据并处理。两者无需直接交互，降低代码复杂度，提高模块化程度。
2. 平衡处理速率差异：生产者与消费者通常以不同速度运行。缓冲区作为中间层，允许生产者持续写入数据，消费者按自身能力消费，避免互相阻塞。
3. 削峰填谷：通过缓冲队列平滑流量波动，避免系统因瞬时高负载崩溃。当生产者突然产生大量请求时，缓冲区暂时存储这些请求，消费者按照自身处理能力逐步消费；当生产者速度降低时，缓冲区逐步释放积压的请求，保持消费者稳定工作。

2.2 实现阻塞队列

class MyBlockingQueue {
    private int head = 0;
    private int tail = 0;
    private int useSize = 0;
    private final String[] array;

    public MyBlockingQueue(int capacity) {
        array = new String[capacity];
    }

    // 添加
    public synchronized void put(String string) throws InterruptedException {
        if (isFull()) {
            // 队列满了，等待消费者消耗元素
            this.wait();
        }
        array[tail] = string;
        tail = (tail + 1) % array.length;
        useSize++;
        this.notify();
    }

    // 删除
    public String take() throws InterruptedException {
        String ret;
        synchronized (this) {
            if (useSize <= 0) {
                // 队列空了，等待生产者添加元素
                this.wait();
            }
            ret = array[head];
            head = (head + 1) % array.length;
            useSize--;
            this.notify();
        }
        return ret;
    }

    // 判断是否满了
    public boolean isFull() {
        return useSize >= array.length;
    }
}

2.3 实现生产者/消费者模式

public class ProducerConsumerDemo {
    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue(1000);
        Thread thread1 = new Thread(() -> {
            int n = 1;
            while (true) {
                try {
                    queue.put(String.valueOf(n));
                    System.out.println("生产元素 n = " + n);
                    n++;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        Thread thread2 = new Thread(() -> {
            while (true) {
                try {
                    System.out.println("消费元素 n = " + queue.take());
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        thread1.start();
        thread2.start();
    }
}

3. 定时器

3.1 概述

定时器 (Timer)：用于在特定时间间隔或指定时间点执行任务的编程模式，广泛应用于定时任务调度、延迟操作、周期性任务等场景。核心思想是将任务的执行逻辑与时间控制解耦，通过统一的定时器管理多个任务。

核心作用/特点

1. 管理异步任务调度：Timer 允许你安排一个任务在未来的某个时间点执行，或者以固定的间隔重复执行。
2. 后台执行：Timer 可以使用一个后台线程来执行任务，这意味着调度和执行任务不会阻塞主线程 (主线程结束后后台线程跟着结束)。
3. 简单易用：Timer 提供了一个相对简单的方式来处理定时任务，适合用于不需要复杂调度的场景。

标准库 Timer 构造方法

// 1. 默认构造方法
// 创建一个 Timer 对象，是一个后台线程，并使用线程的默认名字
public Timer() {
    this("Timer-" + serialNumber());
}

// 2. 指定线程名字的构造方法
// 创建一个 Timer 对象，是一个后台线程，并使用指定的线程名字
public Timer(String name) {
    thread.setName(name);
    thread.start();
}

// 3. 指定是否为后台线程的构造方法
// 传入 true，是后台线程；传入 false，是前台线程
public Timer(boolean isDaemon) {
    this("Timer-" + serialNumber(), isDaemon);
}

// 4. 指定线程名字和是否为后台线程的构造方法
public Timer(String name, boolean isDaemon) {
    thread.setName(name);
    thread.setDaemon(isDaemon);
    thread.start();
}

标准库 Timer 的 schedule 方法

• 1. schedule(TimerTask task, Date time)：安排任务在指定的时间执行一次

public static void main(String[] args) {
    Timer timer = new Timer();
    TimerTask timerTask = new TimerTask() {
        @Override
        public void run() {
            System.out.println("延迟三秒执行");
        }
    };
    // 使用 Date 对象来指定具体的执行时间
    // new Date(System.currentTimeMillis()+1000) 表示当前时间等待 1000ms
    timer.schedule(timerTask, new Date(System.currentTimeMillis() + 1000));
}

• 2. schedule(TimerTask task, Date firstTime, long period)：安排任务在指定的时间首次执行，然后每隔一段时间重复执行

public static void main(String[] args) {
    Timer timer = new Timer();
    TimerTask timerTask = new TimerTask() {
        @Override
        public void run() {
            System.out.println("延迟三秒执行");
        }
    };
    // 当前时间等待 1000ms 后第一次执行任务
    // 此后每间隔 1000ms 就执行一次任务
    timer.schedule(timerTask, new Date(System.currentTimeMillis() + 1000), 1000);
}

• 3. schedule(TimerTask task, long delay)：安排任务在指定的延迟时间后执行一次 (相对于当前时间)

public static void main(String[] args) {
    Timer timer = new Timer();
    TimerTask timerTask = new TimerTask() {
        @Override
        public void run() {
            System.out.println("延迟三秒执行");
        }
    };
    // 当前时间延迟 3000ms 后执行
    timer.schedule(timerTask, 3000);
}

• 4. schedule(TimerTask task, long delay, long period)：安排任务在指定的延迟时间后首次执行，然后每隔一段时间重复执行

public static void main(String[] args) {
    Timer timer = new Timer();
    TimerTask timerTask = new TimerTask() {
        @Override
        public void run() {
            System.out.println("延迟三秒执行");
        }
    };
    // 当前时间延迟 3000ms 后执行
    // 此后每间隔 3000ms 就执行一次任务
    timer.schedule(timerTask, 3000, 3000);
}

3.2 模拟实现定时器

class MyTask implements Comparable<MyTask> {
    private final Runnable runnable;
    private final long time;

    public MyTask(Runnable runnable, long delay) {
        this.runnable = runnable;
        this.time = System.currentTimeMillis() + delay;
    }

    public long getTime() {
        return this.time;
    }

    public void run() {
        runnable.run();
    }

    @Override
    public int compareTo(MyTask o) {
        return (int) (this.time - o.time);
    }
}

class MyTime {
    private final PriorityQueue<MyTask> queue = new PriorityQueue<>();

    public void schedule(Runnable runnable, long delay) {
        synchronized (this) {
            MyTask myTask = new MyTask(runnable, delay);
            queue.offer(myTask);
            this.notify();
        }
    }

    public MyTime() {
        Thread thread = new Thread(() -> {
            while (true) {
                try {
                    synchronized (this) {
                        while (queue.isEmpty()) {
                            this.wait();
                        }
                        MyTask myTask = queue.peek();
                        long curTime = System.currentTimeMillis();
                        if (curTime >= myTask.getTime()) {
                            myTask.run();
                            queue.poll();
                        } else {
                            this.wait(myTask.getTime() - curTime);
                        }
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        thread.setDaemon(true);
        thread.start();
    }
}

4. 线程池

4.1 概述

线程池：线程池是一种管理和复用线程的编程模式。它预先创建一定数量的线程，在执行任务需要时，将任务分配给这些线程，从而提高运行效率。

核心作用：优化多线程任务的执行效率与管理资源。

特点

• 线程复用：当线程执行完一个任务时，不会立即销毁，而是等待下一个任务的到来 (当然这种等待是有时间限制的)，这样避免了频繁的创建和销毁线程。
• 动态调整：根据实际环境需要动态调整线程数量，以达到最佳性能。
• 任务队列：线程池会维护一个任务队列，用于存放待执行的任务，当线程空闲时，从队列中取出任务并执行。

标准库线程池构造方法参数

1. int corePoolSize：核心线程数。
2. int maximumPoolSize：最大线程数。
3. long keepAliveTime：非核心线程的空闲时的最大存活时间。
4. TimeUnit unit：时间单位。
5. BlockingQueue workQueue：任务队列。
6. ThreadFactory threadFactory：线程工厂，用于创建新线程的工厂。
7. RejectedExecutionHandler handler：拒绝策略。

4.2 线程池的执行流程

假设现在有一个线程池：核心线程数 2，最大线程数 4，等待队列 2。

• 任务数量<=2(A,B) 时，由核心线程执行任务。
• 2<任务数量<=4(A,B,C,D) 时，核心线程无法同时处理所有任务，未被执行的任务 (C,D) 将会进入等待队列中等待核心线程执行。
• 4<任务数量<=6(A,B,C,D,E,F)，此时等待队列也满了，线程池就会开放非核心线程来执行任务，C 和 D 任务继续在等待队列中等待，新添加的 E 和 F 任务由非核心线程来执行任务。
• 任务数量>6，核心线程，等待队列，非核心线程都被任务所占用，仍然无法满足需求，此时就会触发线程池的拒绝策略。

4.3 拒绝策略

• AbortPolicy: 直接抛异常。
• CallerRunsPolicy: 由提交该任务的线程来执行。
• DiscardPolicy: 丢弃新任务。
• DiscardOldestPolicy: 丢弃最老的任务。

4.4 模拟实现线程池

public class MyThreadPoolExecutor {
    private final int capacity = 1000; // 阻塞队列
    private final MyBlockingQueue queue = new MyBlockingQueue(capacity);
    private final List<Thread> list = new ArrayList<>();

    // 创建线程
    public MyThreadPoolExecutor(int n) {
        for (int i = 0; i < n; i++) {
            Thread thread = new Thread(() -> {
                while (true) {
                    try {
                        queue.take().run();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            });
            thread.start();
            list.add(thread);
        }
    }

    // 添加任务
    public void submit(Runnable runnable) throws InterruptedException {
        queue.put(runnable);
    }

    public int getCapacity() {
        return capacity;
    }

    // 获取线程
    public List<Thread> getList() {
        return list;
    }
}