GU𝐆𝐫𝐨𝐰 𝐔𝐩
返回文章列表
JUC并发编程
JUC生产者消费者线程池设计模式

16异步模式：生产者消费者与线程池设计

本篇整理两个异步模式：

  • 生产者/消费者：用一个有容量限制的消息队列，把生产结果的线程与消费结果的线程解耦
  • 工作线程（Thread Pool）：让有限的工作线程轮流异步处理无限多的任务，典型实现就是线程池

异步模式之生产者/消费者

定义

与保护性暂停中的 GuardedObject 不同，生产者/消费者模式不需要产生结果和消费结果的线程一一对应。

要点：

  • 消费队列可以用来平衡生产和消费的线程资源
  • 生产者仅负责产生结果数据，不关心数据该如何处理；消费者专心处理结果数据
  • 消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据
  • JDK 中各种阻塞队列，采用的就是这种模式

t1/t2/t3 不断 put 消息到消息队列，t4 从队列 take 消息处理

实现

消息对象 Message 携带一个 id 和消息体：

class Message {
    private int id;
    private Object message;
 
    public Message(int id, Object message) {
        this.id = id;
        this.message = message;
    }
 
    public int getId() {
        return id;
    }
 
    public Object getMessage() {
        return message;
    }
}

MessageQueue 内部用 LinkedList 存放消息：

  • take：队列空时消费者 wait，取出一条消息后 notifyAll 通知生产者
  • put：队列满时生产者 wait，放入一条消息后 notifyAll 通知消费者
class MessageQueue {
    private LinkedList<Message> queue;
    private int capacity;
 
    public MessageQueue(int capacity) {
        this.capacity = capacity;
        queue = new LinkedList<>();
    }
 
    public Message take() {
        synchronized (queue) {
            while (queue.isEmpty()) {
                log.debug("没货了, wait");
                try {
                    queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            Message message = queue.removeFirst();
            queue.notifyAll();
            return message;
        }
    }
 
    public void put(Message message) {
        synchronized (queue) {
            while (queue.size() == capacity) {
                log.debug("库存已达上限, wait");
                try {
                    queue.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(message);
            queue.notifyAll();
        }
    }
}

应用：4 个生产者线程执行下载任务，把结果放入容量为 2 的消息队列；1 个消费者线程不断取消息处理结果。

MessageQueue messageQueue = new MessageQueue(2);
 
// 4 个生产者线程，下载任务
for (int i = 0; i < 4; i++) {
    int id = i;
    new Thread(() -> {
        try {
            log.debug("download...");
            List<String> response = Downloader.download();
            log.debug("try put message({})", id);
            messageQueue.put(new Message(id, response));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }, "生产者" + i).start();
}
 
// 1 个消费者线程，处理结果
new Thread(() -> {
    while (true) {
        Message message = messageQueue.take();
        List<String> response = (List<String>) message.getMessage();
        log.debug("take message({}): [{}] lines", message.getId(), response.size());
    }
}, "消费者").start();

某次运行结果：

10:48:38.070 [生产者3] c.TestProducerConsumer - download...
10:48:38.070 [生产者0] c.TestProducerConsumer - download...
10:48:38.070 [消费者] c.MessageQueue - 没货了, wait
10:48:38.070 [生产者1] c.TestProducerConsumer - download...
10:48:38.070 [生产者2] c.TestProducerConsumer - download...
10:48:41.236 [生产者1] c.TestProducerConsumer - try put message(1)
10:48:41.237 [生产者2] c.TestProducerConsumer - try put message(2)
10:48:41.236 [生产者0] c.TestProducerConsumer - try put message(0)
10:48:41.237 [生产者3] c.TestProducerConsumer - try put message(3)
10:48:41.239 [生产者2] c.MessageQueue - 库存已达上限, wait
10:48:41.240 [生产者1] c.MessageQueue - 库存已达上限, wait
10:48:41.240 [消费者] c.TestProducerConsumer - take message(0): [3] lines
10:48:41.240 [生产者2] c.MessageQueue - 库存已达上限, wait
10:48:41.240 [消费者] c.TestProducerConsumer - take message(3): [3] lines
10:48:41.240 [消费者] c.TestProducerConsumer - take message(1): [3] lines
10:48:41.240 [消费者] c.TestProducerConsumer - take message(2): [3] lines
10:48:41.240 [消费者] c.MessageQueue - 没货了, wait

结果解读：4 个生产者几乎同时完成下载，但队列容量只有 2，后两个生产者在 put 处等待；消费者每取走一条，就唤醒一个等待的生产者继续放入。生产与消费的速度差被队列「削峰填谷」。

异步模式之工作线程

定义

让有限的工作线程（Worker Thread）来轮流异步处理无限多的任务。它也可以被归类为分工模式，典型实现就是线程池，同时也体现了经典设计模式中的享元模式——线程对象被大量任务共享复用。

例如海底捞的服务员（线程）轮流处理每位客人的点餐（任务），如果为每位客人都配一名专属服务员，成本就太高了（对比另一种多线程设计模式：Thread-Per-Message）。

注意：不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率。例如一个餐馆的工人既要招呼客人（任务类型 A）又要到后厨做菜（任务类型 B），效率不高，分成服务员（线程池 A）与厨师（线程池 B）更为合理，当然还可以做更细致的分工。

饥饿

固定大小线程池会有饥饿现象。场景设定：

  • 两个工人是同一个线程池中的两个线程
  • 他们要做的事情是为客人点餐和到后厨做菜，这是两个阶段的工作
  • 客人点餐：必须先点完餐、等菜做好、上菜，处理点餐的工人在此期间必须等待
  • 后厨做菜：直接做即可

工人 A 处理点餐，等工人 B 把菜做好再上菜，俩人配合得蛮好；但同时来了两个客人时，A 和 B 都去处理点餐了，这时没人做饭，饥饿产生。

public class TestDeadLock {
 
    static final List<String> MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅");
    static Random RANDOM = new Random();
 
    static String cooking() {
        return MENU.get(RANDOM.nextInt(MENU.size()));
    }
 
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
 
        executorService.execute(() -> {
            log.debug("处理点餐...");
            Future<String> f = executorService.submit(() -> {
                log.debug("做菜");
                return cooking();
            });
            try {
                log.debug("上菜: {}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
 
        executorService.execute(() -> {
            log.debug("处理点餐...");
            Future<String> f = executorService.submit(() -> {
                log.debug("做菜");
                return cooking();
            });
            try {
                log.debug("上菜: {}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }
}

只有一个点餐任务时输出正常：

17:21:27.883 c.TestDeadLock [pool-1-thread-1] - 处理点餐...
17:21:27.891 c.TestDeadLock [pool-1-thread-2] - 做菜
17:21:27.891 c.TestDeadLock [pool-1-thread-1] - 上菜: 烤鸡翅

当第二个点餐任务也提交后，可能出现两个线程都卡在「处理点餐」、做菜任务永远排不到线程执行的输出：

17:08:41.339 c.TestDeadLock [pool-1-thread-2] - 处理点餐...
17:08:41.339 c.TestDeadLock [pool-1-thread-1] - 处理点餐...

增加线程池大小可以缓解，但不是根本解决方案。根本方案还是前面提到的：不同任务类型采用不同的线程池：

public class TestDeadLock {
 
    static final List<String> MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅");
    static Random RANDOM = new Random();
 
    static String cooking() {
        return MENU.get(RANDOM.nextInt(MENU.size()));
    }
 
    public static void main(String[] args) {
        ExecutorService waiterPool = Executors.newFixedThreadPool(1);
        ExecutorService cookPool = Executors.newFixedThreadPool(1);
 
        waiterPool.execute(() -> {
            log.debug("处理点餐...");
            Future<String> f = cookPool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });
            try {
                log.debug("上菜: {}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
 
        waiterPool.execute(() -> {
            log.debug("处理点餐...");
            Future<String> f = cookPool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });
            try {
                log.debug("上菜: {}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }
}
17:25:14.626 c.TestDeadLock [pool-1-thread-1] - 处理点餐...
17:25:14.630 c.TestDeadLock [pool-2-thread-1] - 做菜
17:25:14.631 c.TestDeadLock [pool-1-thread-1] - 上菜: 地三鲜
17:25:14.632 c.TestDeadLock [pool-1-thread-1] - 处理点餐...
17:25:14.632 c.TestDeadLock [pool-2-thread-1] - 做菜
17:25:14.632 c.TestDeadLock [pool-1-thread-1] - 上菜: 辣子鸡丁

创建多少线程合适

  • 线程数过小：程序不能充分利用系统资源，还容易导致饥饿
  • 线程数过大：导致更多的线程上下文切换，占用更多内存

CPU 密集型运算

通常采用 CPU 核数 + 1 能够实现最优的 CPU 利用率。+1 是保证：当某线程由于页缺失故障（操作系统）或其它原因导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费。

I/O 密集型运算

CPU 不总是处于繁忙状态：执行业务计算时使用 CPU 资源，但执行 I/O 操作、远程 RPC 调用、数据库操作时 CPU 就闲下来了，可以利用多线程提高利用率。

经验公式：

线程数 = 核数 × 期望 CPU 利用率 × 总时间(CPU计算时间 + 等待时间) / CPU计算时间

例如 4 核 CPU，计算时间占 50%，其它等待时间占 50%，期望 CPU 被 100% 利用：

4 × 100% × 100% / 50% = 8

例如 4 核 CPU，计算时间占 10%，等待时间占 90%，期望 CPU 被 100% 利用：

4 × 100% × 100% / 10% = 40

自定义线程池

线程池的核心结构：主线程不断 execute 提交任务，线程 t1、t2、t3 这些工作线程不断从阻塞队列里 poll 任务；核心线程都在忙时，新任务进入 Blocking Queue 暂存。

实现分四步：自定义拒绝策略接口 → 自定义任务队列 → 自定义线程池 → 测试。

步骤 1：自定义拒绝策略接口

队列满时怎么办，交给调用方提供的策略实现决定：

@FunctionalInterface // 拒绝策略
interface RejectPolicy<T> {
    void reject(BlockingQueue<T> queue, T task);
}

可选策略包括：死等、带超时等待、让调用者放弃任务、让调用者抛异常、让调用者自己执行任务。

步骤 2：自定义任务队列

基于 ReentrantLock + 两个 Condition：fullWaitSet 供生产者在队满时等待，emptyWaitSet 供消费者在队空时等待。

class BlockingQueue<T> {
    // 1. 任务队列
    private Deque<T> queue = new ArrayDeque<>();
 
    // 2. 锁
    private ReentrantLock lock = new ReentrantLock();
 
    // 3. 生产者条件变量
    private Condition fullWaitSet = lock.newCondition();
 
    // 4. 消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();
 
    // 5. 容量
    private int capcity;
 
    public BlockingQueue(int capcity) {
        this.capcity = capcity;
    }
 
    // 带超时阻塞获取
    public T poll(long timeout, TimeUnit unit) {
        lock.lock();
        try {
            // 将 timeout 统一转换为纳秒
            long nanos = unit.toNanos(timeout);
            while (queue.isEmpty()) {
                try {
                    // 返回值是剩余时间
                    if (nanos <= 0) {
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }
 
    // 阻塞获取
    public T take() {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }
 
    // 阻塞添加
    public void put(T task) {
        lock.lock();
        try {
            while (queue.size() == capcity) {
                try {
                    log.debug("等待加入任务队列 {} ...", task);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入任务队列 {}", task);
            queue.addLast(task);
            emptyWaitSet.signal();
        } finally {
            lock.unlock();
        }
    }
 
    // 带超时时间阻塞添加
    public boolean offer(T task, long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.size() == capcity) {
                try {
                    if (nanos <= 0) {
                        return false;
                    }
                    log.debug("等待加入任务队列 {} ...", task);
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入任务队列 {}", task);
            queue.addLast(task);
            emptyWaitSet.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }
 
    public int size() {
        lock.lock();
        try {
            return queue.size();
        } finally {
            lock.unlock();
        }
    }
 
    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            // 判断队列是否满
            if (queue.size() == capcity) {
                rejectPolicy.reject(this, task);
            } else { // 有空闲
                log.debug("加入任务队列 {}", task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}

步骤 3：自定义线程池

execute 时若工作线程数小于核心线程数，新建 Worker 直接执行任务；否则把任务交给队列，由拒绝策略决定入队方式。Worker 把首个任务执行完后，继续带超时地从队列取任务，取不到任务（空闲超时）就把自己从线程集合中移除，实现线程回收。

class ThreadPool {
    // 任务队列
    private BlockingQueue<Runnable> taskQueue;
 
    // 线程集合
    private HashSet<Worker> workers = new HashSet<>();
 
    // 核心线程数
    private int coreSize;
 
    // 获取任务时的超时时间
    private long timeout;
 
    private TimeUnit timeUnit;
 
    private RejectPolicy<Runnable> rejectPolicy;
 
    // 执行任务
    public void execute(Runnable task) {
        // 当任务数没有超过 coreSize 时，直接交给 worker 对象执行
        // 如果任务数超过 coreSize 时，加入任务队列暂存
        synchronized (workers) {
            if (workers.size() < coreSize) {
                Worker worker = new Worker(task);
                log.debug("新增 worker{}, {}", worker, task);
                workers.add(worker);
                worker.start();
            } else {
                // taskQueue.put(task);
                // 1) 死等
                // 2) 带超时等待
                // 3) 让调用者放弃任务执行
                // 4) 让调用者抛出异常
                // 5) 让调用者自己执行任务
                taskQueue.tryPut(rejectPolicy, task);
            }
        }
    }
 
    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity,
                      RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapcity);
        this.rejectPolicy = rejectPolicy;
    }
 
    class Worker extends Thread {
        private Runnable task;
 
        public Worker(Runnable task) {
            this.task = task;
        }
 
        @Override
        public void run() {
            // 执行任务
            // 1) 当 task 不为空，执行任务
            // 2) 当 task 执行完毕，再接着从任务队列获取任务并执行
            while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                try {
                    log.debug("正在执行...{}", task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    task = null;
                }
            }
            synchronized (workers) {
                log.debug("worker 被移除{}", this);
                workers.remove(this);
            }
        }
    }
}

步骤 4：测试

核心线程数 1、队列容量 1、空闲超时 1 秒，提交 4 个睡眠 1 秒的任务；拒绝策略选择「让调用者自己执行」：

public static void main(String[] args) {
    ThreadPool threadPool = new ThreadPool(1,
            1000, TimeUnit.MILLISECONDS, 1, (queue, task) -> {
        // 1. 死等
        // queue.put(task);
        // 2) 带超时等待
        // queue.offer(task, 1500, TimeUnit.MILLISECONDS);
        // 3) 让调用者放弃任务执行
        // log.debug("放弃{}", task);
        // 4) 让调用者抛出异常
        // throw new RuntimeException("任务执行失败 " + task);
        // 5) 让调用者自己执行任务
        task.run();
    });
 
    for (int i = 0; i < 4; i++) {
        int j = i;
        threadPool.execute(() -> {
            try {
                Thread.sleep(1000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("{}", j);
        });
    }
}

可以把注释依次切换到其它四种策略观察行为差异。这个自定义实现已经具备了 JDK ThreadPoolExecutor 的雏形：核心线程、阻塞队列、拒绝策略、空闲线程回收（JDK 版本的详细分析见本系列 08线程池）。

小结

  • 生产者/消费者用有容量限制的消息队列解耦两类线程：队空消费者等待、队满生产者等待，队列同时承担「削峰填谷」的平衡作用；JDK 的阻塞队列都是这一模式
  • 工作线程模式让少量线程复用处理大量任务，本质是享元思想
  • 固定大小线程池在「任务之间还有提交依赖」时会发生饥饿：线程都在等待永远排不进来的子任务。根本解法是不同任务类型使用不同线程池
  • 线程数估算：CPU 密集型取「核数 + 1」；I/O 密集型按「核数 × 期望利用率 × 总时间 / CPU 计算时间」估算
  • 自定义线程池四件套：核心线程集合、阻塞队列、拒绝策略、空闲超时回收