Java并发设计模式详解

1/18/24About 17 min

1. 并发设计模式概述

并发设计模式是针对多线程环境下常见问题的解决方案，它们提供了经过验证的设计思路和实现方式，可以帮助开发者更高效地编写并发程序。

1.1 并发设计模式的重要性

在多核处理器时代，并发编程已成为提高应用性能的关键技术。然而，并发编程也带来了诸多挑战，如线程安全、死锁、竞态条件等。并发设计模式通过封装复杂的并发控制逻辑，提供了简洁、可复用的解决方案，帮助开发者：

提高代码的可读性和可维护性
减少并发编程错误
提高应用性能和吞吐量
增强系统的可扩展性

1.2 并发设计模式的分类

根据解决的问题类型，并发设计模式可以分为以下几类：

类型	主要解决问题	典型模式
同步模式	线程间同步与协作	生产者-消费者、读者-写者、两阶段终止
资源管理模式	共享资源的安全访问	线程安全单例、不可变对象
任务执行模式	任务的并发执行	线程池、并行流水线
并发控制模式	复杂并发流程控制	屏障、门闩、信号量
性能优化模式	并发性能提升	无锁编程、分治算法

1.3 并发设计模式的核心原则

最小化共享状态：减少线程间共享的数据量，降低锁竞争
优先使用不可变对象：不可变对象天然线程安全
合理使用锁：选择适当的锁粒度和类型
避免死锁：遵循锁的获取顺序，使用超时机制
使用并发工具类：优先使用Java并发包提供的工具类

2. 生产者-消费者模式

生产者-消费者模式是最经典的并发设计模式之一，它解决了生产者线程和消费者线程之间的协作问题。

2.1 基本概念与原理

生产者-消费者模式包含以下核心组件：

生产者：生成数据并放入共享缓冲区
消费者：从共享缓冲区取出数据并处理
共享缓冲区：用于存储生产者生成的数据，平衡生产者和消费者的处理速度
同步机制：确保生产者和消费者之间的安全协作

2.2 基于BlockingQueue的实现

Java并发包提供的BlockingQueue是实现生产者-消费者模式的理想选择：

import java.util.concurrent.*;

public class ProducerConsumerExample {
    private static final int BUFFER_SIZE = 10;
    private static final int PRODUCER_COUNT = 2;
    private static final int CONSUMER_COUNT = 3;
    private static final int TASK_COUNT = 20;
    
    private final BlockingQueue<Integer> buffer = new ArrayBlockingQueue<>(BUFFER_SIZE);
    
    // 生产者线程
    private class Producer implements Runnable {
        private final int id;
        
        public Producer(int id) {
            this.id = id;
        }
        
        @Override
        public void run() {
            try {
                for (int i = 0; i < TASK_COUNT; i++) {
                    int data = id * 100 + i;
                    buffer.put(data); // 阻塞直到有空间
                    System.out.println("生产者 " + id + " 生产了: " + data + ", 缓冲区大小: " + buffer.size());
                    Thread.sleep(ThreadLocalRandom.current().nextInt(100, 300));
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
    // 消费者线程
    private class Consumer implements Runnable {
        private final int id;
        
        public Consumer(int id) {
            this.id = id;
        }
        
        @Override
        public void run() {
            try {
                while (true) {
                    Integer data = buffer.take(); // 阻塞直到有数据
                    System.out.println("消费者 " + id + " 消费了: " + data + ", 缓冲区大小: " + buffer.size());
                    processData(data);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        
        private void processData(Integer data) throws InterruptedException {
            // 模拟数据处理
            Thread.sleep(ThreadLocalRandom.current().nextInt(200, 500));
        }
    }
    
    public void start() {
        ExecutorService executor = Executors.newCachedThreadPool();
        
        // 启动生产者线程
        for (int i = 0; i < PRODUCER_COUNT; i++) {
            executor.submit(new Producer(i));
        }
        
        // 启动消费者线程
        for (int i = 0; i < CONSUMER_COUNT; i++) {
            executor.submit(new Consumer(i));
        }
        
        // 关闭线程池
        executor.shutdown();
    }
    
    public static void main(String[] args) {
        ProducerConsumerExample example = new ProducerConsumerExample();
        example.start();
    }
}

2.3 基于Condition的实现

使用ReentrantLock和Condition可以更灵活地实现生产者-消费者模式：

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ProducerConsumerWithCondition {
    private static final int BUFFER_SIZE = 10;
    private final Queue<Integer> buffer = new LinkedList<>();
    private final Lock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    private final Condition notFull = lock.newCondition();
    
    // 生产数据
    public void produce(int data) throws InterruptedException {
        lock.lock();
        try {
            while (buffer.size() == BUFFER_SIZE) {
                // 缓冲区满，等待消费者消费
                notFull.await();
            }
            
            buffer.offer(data);
            System.out.println("生产了: " + data + ", 缓冲区大小: " + buffer.size());
            
            // 通知消费者有新数据
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }
    
    // 消费数据
    public Integer consume() throws InterruptedException {
        lock.lock();
        try {
            while (buffer.isEmpty()) {
                // 缓冲区空，等待生产者生产
                notEmpty.await();
            }
            
            Integer data = buffer.poll();
            System.out.println("消费了: " + data + ", 缓冲区大小: " + buffer.size());
            
            // 通知生产者有空间
            notFull.signal();
            
            return data;
        } finally {
            lock.unlock();
        }
    }
    
    public static void main(String[] args) {
        ProducerConsumerWithCondition pc = new ProducerConsumerWithCondition();
        
        // 生产者线程
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 20; i++) {
                    pc.produce(i);
                    Thread.sleep(ThreadLocalRandom.current().nextInt(100, 300));
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        // 消费者线程
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 20; i++) {
                    pc.consume();
                    Thread.sleep(ThreadLocalRandom.current().nextInt(200, 500));
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        
        producer.start();
        consumer.start();
    }
}

2.4 生产者-消费者模式的变体

单生产者-单消费者：最简单的形式，一个生产者和一个消费者
多生产者-单消费者：多个生产者生成数据，一个消费者处理
单生产者-多消费者：一个生产者生成数据，多个消费者处理
多生产者-多消费者：最复杂的形式，多个生产者和多个消费者

3. 读者-写者模式

读者-写者模式解决了共享资源的读写问题，它允许多个读者同时读取共享资源，但同一时间只能有一个写者修改共享资源。

3.1 基本概念与原理

读者-写者模式的核心思想是：

读操作可以并发：多个读者可以同时读取共享资源
写操作必须独占：写者需要独占共享资源，防止与其他读者或写者冲突
优先级策略：可以根据业务需求设置读者优先或写者优先

3.2 基于ReentrantReadWriteLock的实现

Java并发包提供的ReentrantReadWriteLock是实现读者-写者模式的理想选择：

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteLockExample {
    private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private int sharedData = 0;
    
    // 读操作
    public int readData() throws InterruptedException {
        rwLock.readLock().lock();
        try {
            System.out.println(Thread.currentThread().getName() + " 开始读取数据: " + sharedData);
            Thread.sleep(ThreadLocalRandom.current().nextInt(100, 200)); // 模拟读操作耗时
            System.out.println(Thread.currentThread().getName() + " 完成读取数据: " + sharedData);
            return sharedData;
        } finally {
            rwLock.readLock().unlock();
        }
    }
    
    // 写操作
    public void writeData(int newData) throws InterruptedException {
        rwLock.writeLock().lock();
        try {
            System.out.println(Thread.currentThread().getName() + " 开始写入数据: " + newData);
            Thread.sleep(ThreadLocalRandom.current().nextInt(300, 500)); // 模拟写操作耗时
            sharedData = newData;
            System.out.println(Thread.currentThread().getName() + " 完成写入数据: " + newData);
        } finally {
            rwLock.writeLock().unlock();
        }
    }
    
    public static void main(String[] args) {
        ReadWriteLockExample example = new ReadWriteLockExample();
        
        // 创建多个读者线程
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    for (int j = 0; j < 3; j++) {
                        example.readData();
                        Thread.sleep(ThreadLocalRandom.current().nextInt(100, 300));
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "读者-" + i).start();
        }
        
        // 创建多个写者线程
        for (int i = 0; i < 2; i++) {
            final int writerId = i;
            new Thread(() -> {
                try {
                    for (int j = 0; j < 2; j++) {
                        example.writeData(writerId * 10 + j);
                        Thread.sleep(ThreadLocalRandom.current().nextInt(500, 800));
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "写者-" + i).start();
        }
    }
}

3.3 读者-写者模式的变体

读者优先：当有读者正在读取时，新的读者可以立即开始读取，写者需要等待所有读者完成
写者优先：当有写者请求写入时，新的读者需要等待写者完成，防止写者饥饿
公平策略：按照请求的顺序处理，无论是读者还是写者

4. 线程池模式

线程池模式是一种管理和复用线程的模式，它可以减少线程创建和销毁的开销，提高应用性能。

4.1 基本概念与原理

线程池的核心组件包括：

线程池管理器：负责线程池的创建、销毁和管理
工作线程：执行任务的线程
任务队列：存储等待执行的任务
任务接口：定义任务的执行方法

4.2 Java线程池的实现

Java并发包提供了Executor框架，它是线程池模式的完整实现：

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadPoolExample {
    private static final AtomicInteger taskCounter = new AtomicInteger(0);
    
    public static void main(String[] args) {
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2, // 核心线程数
                5, // 最大线程数
                60, // 线程空闲时间
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(10), // 任务队列
                new ThreadFactory() {
                    private final AtomicInteger threadCounter = new AtomicInteger(0);
                    
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("工作线程-" + threadCounter.incrementAndGet());
                        return thread;
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );
        
        // 提交任务
        for (int i = 0; i < 15; i++) {
            executor.submit(() -> {
                int taskId = taskCounter.incrementAndGet();
                System.out.println(Thread.currentThread().getName() + " 开始执行任务: " + taskId);
                try {
                    Thread.sleep(ThreadLocalRandom.current().nextInt(500, 1000));
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println(Thread.currentThread().getName() + " 完成执行任务: " + taskId);
            });
        }
        
        // 关闭线程池
        executor.shutdown();
    }
}

4.3 线程池的工作原理

线程池的工作流程如下：

当提交一个新任务时，线程池首先检查核心线程数是否已满
如果核心线程数未满，创建新的核心线程执行任务
如果核心线程数已满，将任务放入任务队列
如果任务队列已满，检查最大线程数是否已满
如果最大线程数未满，创建新的非核心线程执行任务
如果最大线程数已满，执行拒绝策略

4.4 线程池的最佳实践

合理设置线程池参数：根据任务类型和系统资源设置合适的核心线程数、最大线程数和任务队列大小
使用合适的拒绝策略：根据业务需求选择合适的拒绝策略
避免任务队列过大：防止系统资源耗尽
监控线程池状态：定期监控线程池的状态，及时调整参数
正确关闭线程池：使用shutdown()或shutdownNow()方法关闭线程池

5. 两阶段终止模式

两阶段终止模式用于安全地终止线程，它确保线程在终止前完成必要的清理工作。

5.1 基本概念与原理

两阶段终止模式的核心思想是：

第一阶段：发送终止请求，设置终止标志或中断线程
第二阶段：线程响应终止请求，完成清理工作后终止

5.2 基于中断的实现

使用线程中断机制可以实现两阶段终止模式：

import java.util.concurrent.TimeUnit;

public class TwoPhaseTerminationExample {
    private Thread workerThread;
    private volatile boolean isTerminated = false;
    private final Object lock = new Object();
    
    // 启动线程
    public void start() {
        workerThread = new Thread(() -> {
            while (!isTerminated) {
                try {
                    // 执行任务
                    doWork();
                    
                    // 等待任务请求
                    synchronized (lock) {
                        lock.wait();
                    }
                } catch (InterruptedException e) {
                    // 响应中断
                    Thread.currentThread().interrupt();
                }
            }
            
            // 执行清理工作
            doCleanup();
        }, "工作线程");
        
        workerThread.start();
    }
    
    // 终止线程
    public void stop() {
        isTerminated = true;
        workerThread.interrupt();
        
        // 唤醒可能在等待的线程
        synchronized (lock) {
            lock.notify();
        }
    }
    
    // 执行任务
    private void doWork() throws InterruptedException {
        System.out.println(Thread.currentThread().getName() + " 执行任务");
        TimeUnit.SECONDS.sleep(1);
    }
    
    // 清理工作
    private void doCleanup() {
        System.out.println(Thread.currentThread().getName() + " 执行清理工作");
    }
    
    public static void main(String[] args) throws InterruptedException {
        TwoPhaseTerminationExample example = new TwoPhaseTerminationExample();
        
        // 启动线程
        example.start();
        
        // 运行一段时间
        TimeUnit.SECONDS.sleep(3);
        
        // 终止线程
        example.stop();
    }
}

5.3 两阶段终止模式的注意事项

处理中断异常：在任务执行过程中，要正确处理InterruptedException
使用volatile变量：确保终止标志的可见性
避免死锁：在清理工作中要避免获取新的锁
确保清理工作完成：确保所有必要的清理工作都能执行完成

6. 线程安全单例模式

线程安全单例模式确保在多线程环境下，类只有一个实例，并且提供全局访问点。

6.1 双重检查锁定模式

双重检查锁定模式是一种常用的线程安全单例实现方式：

public class DoubleCheckedLockingSingleton {
    // 使用volatile确保instance的可见性和有序性
    private volatile static DoubleCheckedLockingSingleton instance;
    
    // 私有构造方法
    private DoubleCheckedLockingSingleton() {
        // 防止反射创建实例
        if (instance != null) {
            throw new RuntimeException("单例类不允许创建多个实例");
        }
    }
    
    // 获取实例
    public static DoubleCheckedLockingSingleton getInstance() {
        if (instance == null) { // 第一次检查
            synchronized (DoubleCheckedLockingSingleton.class) {
                if (instance == null) { // 第二次检查
                    instance = new DoubleCheckedLockingSingleton();
                }
            }
        }
        return instance;
    }
}

6.2 静态内部类模式

静态内部类模式是一种更简洁、更安全的线程安全单例实现方式：

public class StaticInnerClassSingleton {
    // 私有构造方法
    private StaticInnerClassSingleton() {
        // 防止反射创建实例
        if (InnerClass.instance != null) {
            throw new RuntimeException("单例类不允许创建多个实例");
        }
    }
    
    // 静态内部类
    private static class InnerClass {
        private static final StaticInnerClassSingleton instance = new StaticInnerClassSingleton();
    }
    
    // 获取实例
    public static StaticInnerClassSingleton getInstance() {
        return InnerClass.instance;
    }
}

6.3 枚举单例模式

枚举单例模式是最安全的线程安全单例实现方式，它可以防止反射和序列化攻击：

public enum EnumSingleton {
    INSTANCE;
    
    // 单例方法
    public void doSomething() {
        System.out.println("枚举单例执行方法");
    }
}

7. 不可变对象模式

不可变对象模式是指对象创建后，其状态不能被修改的模式。不可变对象天然线程安全，不需要额外的同步措施。

7.1 基本概念与原理

不可变对象的核心特征：

状态不可变：对象创建后，其状态不能被修改
所有字段都是final的：确保字段只能在构造方法中初始化
对象是final的：防止子类修改其行为
防御性拷贝：在返回可变对象时，返回其副本

7.2 不可变对象的实现

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public final class ImmutableObject {
    private final String name;
    private final int age;
    private final List<String> hobbies;
    
    // 构造方法
    public ImmutableObject(String name, int age, List<String> hobbies) {
        this.name = name;
        this.age = age;
        // 防御性拷贝
        this.hobbies = new ArrayList<>(hobbies);
    }
    
    // 获取name
    public String getName() {
        return name;
    }
    
    // 获取age
    public int getAge() {
        return age;
    }
    
    // 获取hobbies，返回不可修改的副本
    public List<String> getHobbies() {
        return Collections.unmodifiableList(hobbies);
    }
    
    // 创建新对象来修改状态
    public ImmutableObject withName(String name) {
        return new ImmutableObject(name, this.age, this.hobbies);
    }
    
    public ImmutableObject withAge(int age) {
        return new ImmutableObject(this.name, age, this.hobbies);
    }
    
    public ImmutableObject withHobbies(List<String> hobbies) {
        return new ImmutableObject(this.name, this.age, hobbies);
    }
}

7.3 不可变对象的优势

线程安全：天然线程安全，不需要额外的同步措施
可共享性：可以安全地在多个线程之间共享
简化并发编程：减少了并发编程的复杂性
提高性能：避免了锁竞争和内存屏障
便于测试：不可变对象的行为更容易预测和测试

8. 并行流水线模式

并行流水线模式是一种将任务分解为多个阶段，每个阶段并行执行的模式，它可以提高任务的处理效率。

8.1 基本概念与原理

并行流水线模式的核心组件包括：

输入阶段：接收输入数据
处理阶段：对数据进行处理，可以并行执行
输出阶段：输出处理结果
流水线管理器：负责协调各个阶段的执行

8.2 基于CompletableFuture的实现

使用CompletableFuture可以实现并行流水线模式：

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ParallelPipelineExample {
    private static final ExecutorService executor = Executors.newFixedThreadPool(10);
    private static final AtomicInteger counter = new AtomicInteger(0);
    
    // 阶段1：数据加载
    private CompletableFuture<String> loadData(String input) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("阶段1-加载数据: " + input);
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "加载后的数据: " + input;
        }, executor);
    }
    
    // 阶段2：数据处理
    private CompletableFuture<String> processData(String data) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("阶段2-处理数据: " + data);
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "处理后的数据: " + data;
        }, executor);
    }
    
    // 阶段3：结果生成
    private CompletableFuture<String> generateResult(String processedData) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("阶段3-生成结果: " + processedData);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "最终结果: " + processedData;
        }, executor);
    }
    
    // 执行流水线
    public CompletableFuture<String> executePipeline(String input) {
        return loadData(input)
                .thenCompose(this::processData)
                .thenCompose(this::generateResult);
    }
    
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ParallelPipelineExample example = new ParallelPipelineExample();
        
        // 提交多个任务到流水线
        for (int i = 0; i < 5; i++) {
            final int taskId = i;
            CompletableFuture<String> future = example.executePipeline("任务-" + taskId);
            
            future.thenAccept(result -> {
                System.out.println("任务 " + taskId + " 完成: " + result);
                counter.incrementAndGet();
            });
        }
        
        // 等待所有任务完成
        while (counter.get() < 5) {
            Thread.sleep(100);
        }
        
        // 关闭线程池
        executor.shutdown();
    }
}