Java 并发核心：单例、生产者消费者、定时器与线程池实战

这里的关键在于 volatile 关键字防止了指令重排序。JVM 对象初始化流程通常分为内存分配、对象初始化和引用赋值三个阶段。如果没有 volatile，可能出现引用赋值早于初始化完成的'半初始化'状态，导致其他线程拿到未完全构建的对象。

懒汉/饿汉对比

生产者/消费者模式

该模式用于协调多个线程或进程之间的任务分配与数据处理。生产者和消费者通过共享缓冲区（队列）解耦，互不直接依赖，从而平衡处理速率差异并削峰填谷。

实现阻塞队列

下面是一个简单的自定义阻塞队列实现，利用 synchronized 和 wait/notify 机制控制线程同步。

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 = (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 = (head + 1) % array.length;
            useSize--;
            this.notify();
        }
        return ret;
    }

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

模拟生产者/消费者

有了阻塞队列，就可以轻松搭建生产消费模型。

public class Producer_Consumer_Blog {
    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();
    }
}

定时器

定时器允许我们在特定时间间隔或指定时间点执行任务，广泛应用于定时调度、延迟操作等场景。

标准库 Timer

Java 标准库提供了 Timer 类，它基于后台线程执行任务，不会阻塞主线程。

// 默认构造方法：创建一个后台线程
public Timer() {
    this("Timer-" + serialNumber());
}

// 指定线程名字
public Timer(String name) {
    thread.setName(name);
    thread.start();
}

// 指定是否为后台线程
public Timer(boolean isDaemon) {
    this("Timer-" + serialNumber(), isDaemon);
}

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

常用调度方法包括：

1. schedule(TimerTask task, Date time)：指定时间执行一次。
2. schedule(TimerTask task, Date firstTime, long period)：首次执行后周期性重复。
3. schedule(TimerTask task, long delay)：延迟指定毫秒后执行一次。
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 后执行
    timer.schedule(timerTask, 3000);
}

模拟实现定时器

为了深入理解原理，我们可以基于优先级队列手动实现一个简单的定时器。

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();
    }
}

线程池

线程池通过预先创建一定数量的线程来复用资源，避免频繁创建销毁线程的开销，同时维护任务队列以缓冲负载。

核心参数

构造线程池时主要关注以下参数：

1. corePoolSize：核心线程数。
2. maximumPoolSize：最大线程数。
3. keepAliveTime：非核心线程的空闲存活时间。
4. unit：时间单位。
5. workQueue：任务队列。
6. threadFactory：线程工厂。
7. handler：拒绝策略。

执行流程

假设线程池配置为：核心线程 2，最大线程 4，等待队列容量 2。

• 任务数 ≤ 2：由核心线程直接执行。
• 2 < 任务数 ≤ 4：核心线程忙不过来，多余任务进入等待队列。
• 4 < 任务数 ≤ 6：等待队列满，线程池创建非核心线程处理新任务。
• 任务数 > 6：所有资源占满，触发拒绝策略。

拒绝策略

当线程池无法处理任务时，会执行拒绝策略：

模拟实现线程池

最后，我们用一个简化的线程池示例来展示基本逻辑。

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;
    }
}