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

实现细节：1.通过 synchronized 加锁解决线程安全问题 2.外层 if 判断，减少锁的开销 3.volatile 防止指令重排序 (避免半初始化对象) 特点：只有在真正需要时才创建实例，减少系统启动时的资源占用，资源利用率高 缺点：若未正确使用同步机制 (如 synchronized 或 volatile)，可能导致多线程环境下实例被多次创建，线程安全实现复杂

1.4 懒汉模式半初始化

Java 对象初始化流程

1. 内存分配阶段：当使用 new 关键字创建对象时，JVM 会在堆内存中为该对象分配空间
2. 对象初始化阶段：对象内存分配完成后，开始执行初始化
3. 引用赋值阶段：内存分配完成后，JVM 将分配的内存地址赋值给引用变量 Java 对象初始化流程 (指令重排序后)
4. 内存分配阶段：当使用 new 关键字创建对象时，JVM 会在堆内存中为该对象分配空间
5. 引用赋值阶段：内存分配完成后，JVM 将分配的内存地址赋值给引用变量
6. 对象初始化阶段：对象内存分配完成后，开始执行初始化

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

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

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(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. 线程复用：当线程执行完一个任务时，不会立即销毁，而是等待下一个任务的到来 (当然这种等待是有时间限制的)，这样避免了频繁的创建和销毁线程
2. 动态调整：根据实际环境需要动态调整线程数量，以达到最佳性能
3. 任务队列：线程池会维护一个任务队列，用于存放待执行的任务，当线程空闲时，从队列中取出任务并执行 标准库线程池构造方法
4. int corePoolSize：核心线程数
5. int maximumPoolSize：最大线程数
6. long keepAliveTime：非核心线程的空闲时的最大存活时间
7. TimeUnit unit：时间单位
8. BlockingQueue workQueue：任务队列
9. ThreadFactory threadFactory：线程工厂，用于创建新线程的工厂
10. RejectedExecutionHandler handler：拒绝策略

4.3 线程池的执行流程

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

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

4.4 拒绝策略

1. AbortPolicy:直接抛异常
2. CallerRunsPolicy:由提交该任务的线程来执行
3. DiscardPolicy:丢弃新任务
4. DiscardOldestPolicy:丢弃最老的任务

4.5 模拟实现线程池

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