【面试手撕】非常规算法，多线程常见手撕题目

生产者消费者

ReentrantLock实现的生产苹果/消费苹果

仓库类：

• 定义最大苹果数量+当前苹果数量
• 生产/消费方法

无非是，先判断条件是否满足，不满足就调用await等待,满足的话生产/消费苹果，之后调用signal唤醒其他线程。

注意用的是条件变量，full，empty

生产的时候，仓库满了调用full.await()等待，生产后，调用empty.signal()唤醒消费者。消费是时候相反即可。

最后，注意try，catch，finally 在finally中实现解锁。

package Others.JUC;import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;public class ProducerConsumerDemo1 {public static void main(String[] args) {// 1.初始化3个生产者线程Storage storage = new Storage();for(int i = 0;i<3;i++){new Thread(new Producer(storage),"生产者"+i).start();}// 2.初始化2个消费者线程for(int i = 0;i<2;i++){new Thread(new Consumer(storage),"消费者"+i).start();}}
}class Storage{private int capacity = 10 ;// 最多容纳10个苹果private int count = 0; // 当前苹果数量private ReentrantLock lock = new ReentrantLock();private Condition full = lock.newCondition();private Condition empty = lock.newCondition();public void produce() throws InterruptedException {try{lock.lock();// 生产苹果while(count>=capacity){full.await();}count++;System.out.println(Thread.currentThread().getName()+" 生产了一个苹果，当前苹果数量："+count);empty.signal();}catch (InterruptedException e){e.printStackTrace();}finally {lock.unlock();}TimeUnit.SECONDS.sleep(1); // 模拟生产苹果的时间}public void consume() throws InterruptedException {try{lock.lock();// 条件不满足，等待while(count<=0){empty.await();}count--;System.out.println(Thread.currentThread().getName()+" 消费了一个苹果，当前苹果数量："+count);full.signal();}catch (InterruptedException e){e.printStackTrace();}finally {lock.unlock();}TimeUnit.SECONDS.sleep(1); // 模拟消费苹果的时间}
}class Producer implements Runnable{private Storage storage;public Producer(Storage storage){this.storage = storage;}public Producer(){}@Overridepublic void run() {while(true){try{Thread.sleep(1000); // 模拟生产苹果的时间,1秒生产一个苹果storage.produce();}catch (InterruptedException e){e.printStackTrace();}}}
}class Consumer implements Runnable{private Storage storage;public Consumer(Storage storage){this.storage = storage;}public Consumer(){}@Overridepublic void run() {while(true){try{Thread.sleep(1000); // 模拟消费苹果的时间,1秒消费一个苹果storage.consume();}catch (InterruptedException e){e.printStackTrace();}}}
}

synchronized实现的生产消费

核心是用synchronized锁住了，由于synchronized可以锁随便一个Object，所以把缓冲区buffer作为synchronized锁住的对象即可。

这里主要的区别就是Storage的实现不一致。其他的一致。

class Storage{// 缓冲区private LinkedList<Object> buffer = new LinkedList<>();// 缓冲区大小private final int MAX_SIZE = 10;public void produce(){synchronized(buffer){while(buffer.size()+1>MAX_SIZE){System.out.println("缓冲区已满，生产者等待");try{buffer.wait();}catch(InterruptedException e){e.printStackTrace();}}buffer.add(new Object());System.out.println("生产者生产一个产品，当前缓冲区大小为"+buffer.size());buffer.notify();}}public void consume(){synchronized(buffer){while(buffer.size()==0){System.out.println("缓冲区为空，消费者等待");try{buffer.wait();}catch(InterruptedException e){e.printStackTrace();}}buffer.remove();System.out.println("消费者消费一个产品，当前缓冲区大小为"+buffer.size());buffer.notify();}}
}class Producer implements Runnable{private Storage storage;public Producer(Storage storage){this.storage = storage;}public Producer(){}@Overridepublic void run() {while(true){try{Thread.sleep(1000); // 模拟生产苹果的时间,1秒生产一个苹果storage.produce();}catch (InterruptedException e){e.printStackTrace();}}}
}

LinkedBlockingQueue阻塞队列方法实现

这里更简单了，不需要手动加锁。只需要使用put/take方法，会自己阻塞

class Storage{LinkedBlockingQueue<Object> buffer = new LinkedBlockingQueue<>(10);public void produce(){try{buffer.put(new Object());System.out.println(Thread.currentThread().getName()+" 生产了一个苹果，当前苹果数量："+buffer.size());} catch (InterruptedException e) {throw new RuntimeException(e);}}public void consume(){try{buffer.take();System.out.println(Thread.currentThread().getName()+" 消费了一个苹果，当前苹果数量："+buffer.size());} catch (InterruptedException e) {throw new RuntimeException(e);}}
}

多条件资源分配

场景： 某工厂需要组装产品，要求：

• 零件A和零件B必须同时存在才能组装
• 仓库最多存放5个A和5个B
• 2个供应商线程分别生产A/B（随机0.5~1秒生产1个）
• 3个组装线程消耗1A+1B组合（每次组装耗时1秒）

要求： 使用 wait/notifyAll 实现资源协调，避免死锁

public class FactoryAssembly {private int partA = 0;private int partB = 0;private final int MAX_PARTS = 5;private int assembledProducts = 0;public static void main(String[] args) {FactoryAssembly factory = new FactoryAssembly();// 创建供应商线程AThread supplierA = new Thread(() -> {try {while (true) {factory.producePartA();// 随机生产时间 0.5-1秒Thread.sleep(500 + (long)(Math.random() * 500));}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});// 创建供应商线程BThread supplierB = new Thread(() -> {try {while (true) {factory.producePartB();// 随机生产时间 0.5-1秒Thread.sleep(500 + (long)(Math.random() * 500));}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});// 创建3个组装线程for (int i = 1; i <= 3; i++) {final int assemblyLineId = i;Thread assemblyLine = new Thread(() -> {try {while (true) {factory.assembleParts(assemblyLineId);// 组装时间1秒Thread.sleep(1000);}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});assemblyLine.start();}supplierA.start();supplierB.start();}// 生产零件Apublic synchronized void producePartA() throws InterruptedException {// 如果A零件已达到上限，等待while (partA >= MAX_PARTS) {System.out.println("零件A仓库已满，供应商A等待中...");wait();}partA++;System.out.println("供应商A生产了一个零件A，当前库存: A=" + partA + ", B=" + partB);// 通知所有等待的线程notifyAll();}// 生产零件Bpublic synchronized void producePartB() throws InterruptedException {// 如果B零件已达到上限，等待while (partB >= MAX_PARTS) {System.out.println("零件B仓库已满，供应商B等待中...");wait();}partB++;System.out.println("供应商B生产了一个零件B，当前库存: A=" + partA + ", B=" + partB);// 通知所有等待的线程notifyAll();}// 组装产品public synchronized void assembleParts(int lineId) throws InterruptedException {// 如果A或B零件不足，等待while (partA < 1 || partB < 1) {System.out.println("组装线 #" + lineId + " 等待零件，当前库存: A=" + partA + ", B=" + partB);wait();}// 消耗零件A和B各一个partA--;partB--;assembledProducts++;System.out.println("组装线 #" + lineId + " 完成一个产品组装! 总计: " + assembledProducts + "，剩余库存: A=" + partA + ", B=" + partB);// 通知所有等待的线程notifyAll();}
}

分布式任务调度模拟

题目5：分布式任务调度模拟（综合）

需求：

• 主节点（Producer）每隔1秒生成一个带优先级的任务
• 3个工作节点（Consumer）从任务队列获取任务执行
• 高优先级任务（如优先级>5）必须立即抢占执行
• 使用 PriorityBlockingQueue 实现任务队列
• 任务执行时间随机（1~3秒）

import java.util.concurrent.*;public class DistributedTaskScheduler {// 定义任务类static class Task implements Comparable<Task> {private final int id;private final int priority;private final int executionTime;public Task(int id, int priority) {this.id = id;this.priority = priority;// 随机执行时间 1-3秒this.executionTime = 1000 + (int)(Math.random() * 2000);}public int getId() {return id;}public int getPriority() {return priority;}public int getExecutionTime() {return executionTime;}// 优先级比较，高优先级的任务排在前面@Overridepublic int compareTo(Task other) {return other.priority - this.priority;}@Overridepublic String toString() {return "Task #" + id + " (优先级: " + priority + ")";}}public static void main(String[] args) {// 使用优先级队列存储任务PriorityBlockingQueue<Task> taskQueue = new PriorityBlockingQueue<>();// 创建锁对象用于高优先级任务抢占Object preemptionLock = new Object();// 任务计数器final int[] taskCounter = {0};// 主节点线程生成任务Thread producer = new Thread(() -> {try {while (true) {// 生成随机优先级 (1-10)int priority = 1 + (int)(Math.random() * 10);Task task = new Task(++taskCounter[0], priority);taskQueue.put(task);System.out.println("主节点生成: " + task);// 如果是高优先级任务，通知工作节点抢占if (priority > 5) {System.out.println("!!!! 高优先级任务 #" + task.getId() + " (优先级: " + priority + ") 需要抢占执行!");synchronized (preemptionLock) {preemptionLock.notifyAll();}}// 每隔1秒生成一个任务Thread.sleep(1000);}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});// 创建3个工作节点线程for (int i = 1; i <= 3; i++) {final int workerId = i;Thread consumer = new Thread(() -> {try {while (true) {// 从队列获取任务Task task = taskQueue.take();System.out.println("工作节点 #" + workerId + " 开始执行: " + task);// 模拟任务执行Thread.sleep(task.getExecutionTime());System.out.println("工作节点 #" + workerId + " 完成任务: " + task);// 检查是否有高优先级任务需要执行if (!taskQueue.isEmpty() && taskQueue.peek().getPriority() > 5) {System.out.println("工作节点 #" + workerId + " 发现高优先级任务，立即处理!");} else {// 如果没有高优先级任务，稍微休息一下synchronized (preemptionLock) {try {preemptionLock.wait(500);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}}}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});consumer.start();}producer.start();}
}

交替打印

两个线程交替打印1-100之间的数

ReentrantLock 实现

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;public class TwoThreadsReentrantLock {private static volatile int num = 1;private static final int MAX = 100;private static Lock lock = new ReentrantLock();private static Condition cA = lock.newCondition();private static Condition cB = lock.newCondition();public static void main(String[] args) {new Thread(()->{while(num<MAX){try{lock.lock();while(num%2==0){cA.await();}System.out.println(Thread.currentThread().getName()+" "+num++);cB.signal();}catch (InterruptedException e){e.printStackTrace();}finally {lock.unlock();}}},"线程A").start();new Thread(()->{while(num<MAX){try{lock.lock();while(num%2!=0){cB.await();}System.out.println(Thread.currentThread().getName()+" "+num++);cA.signal();}catch (InterruptedException e){e.printStackTrace();}finally {lock.unlock();}}},"线程B").start();}
}

synchronized实现

public class TwoThreadsSynchronized {private static volatile int num = 1;private static final int MAX = 100;private static final Object lock = new Object();public static void main(String[] args) {new Thread(()->{while(num<MAX){synchronized(lock){while(num%2==0){try{lock.wait();}catch (InterruptedException e){e.printStackTrace();}}System.out.println(Thread.currentThread().getName()+" "+num++);lock.notify();}}},"t1").start();new Thread(()->{while(num<MAX){synchronized(lock){while(num%2==1){try{lock.wait();}catch (InterruptedException e){e.printStackTrace();}}System.out.println(Thread.currentThread().getName()+" "+num++);lock.notify();}}},"t2").start();}
}

Semaphore实现

import java.util.concurrent.Semaphore;public class TwoThreadsSemaphore {private static volatile int num = 1;private static final int MAX = 100;private static Semaphore s1 = new Semaphore(1);private static Semaphore s2 = new Semaphore(0);public static void main(String[] args) {new Thread(()->{while(num<MAX){try{s1.acquire();}catch (InterruptedException e){e.printStackTrace();}System.out.println(Thread.currentThread().getName()+" "+num++);s2.release();}},"t1").start();new Thread(()->{while(num<MAX){try{s2.acquire();}catch (InterruptedException e){e.printStackTrace();}System.out.println(Thread.currentThread().getName()+" "+num++);s1.release();}},"t2").start();}
}

三个线程交替打印1-100之间的数

ReentrantLock

public class ThreeThreadReentrantLock {static class PrintTask{private static volatile int num = 1;private static final int MAX = 100;private static Lock lock = new ReentrantLock();private static Condition cA = lock.newCondition();private static Condition cB = lock.newCondition();private static Condition cC = lock.newCondition();public void print1(){while(num<=MAX){try{lock.lock();while(num%3!=1){cA.await();}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);cB.signal();}catch (InterruptedException e){e.printStackTrace();}finally {lock.unlock();}}}public void print2(){while(num<=MAX){try{lock.lock();while(num%3!=2){cB.await();}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);cC.signal();}catch (InterruptedException e){e.printStackTrace();}}}public void print3(){while(num<=MAX){try{lock.lock();while(num%3!=0){cC.await();}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);cA.signal();}catch (InterruptedException e){e.printStackTrace();}}}}public static void main(String[] args) {PrintTask task = new PrintTask();new Thread(task::print1,"线程A").start();new Thread(task::print2,"线程B").start();new Thread(task::print3,"线程C").start();}
}

Synchronized

public class ThreeThreadSynchronized {static class PrintTask{private static volatile int num = 1;private static final int MAX = 100;private static final Object lock = new Object();private boolean isA = true;private boolean isB = false;private boolean isC = false;public void print1(){while(num<=MAX){synchronized(lock){while(!isA){try{lock.wait();}catch (InterruptedException e){e.printStackTrace();}}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);isA = false;isB = true;lock.notifyAll();}}}public void print2(){while(num<=MAX){synchronized(lock){while(!isB){try{lock.wait();}catch (InterruptedException e){e.printStackTrace();}}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);isB = false;isC = true;lock.notifyAll();}}}public void print3(){while(num<=MAX){synchronized(lock){while(!isC){try{lock.wait();}catch (InterruptedException e){e.printStackTrace();}}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);isC = false;isA = true;lock.notifyAll();}}}}public static void main(String[] args) {PrintTask task = new PrintTask();new Thread(task::print1,"线程A").start();new Thread(task::print2,"线程B").start();new Thread(task::print3,"线程C").start();}
}

Semaphore

import java.util.concurrent.Semaphore;public class ThreeThreadSemaphore {static class PrintTask{private static volatile int num = 1;private static final int MAX = 100;private static Semaphore s1 = new Semaphore(1);private static Semaphore s2 = new Semaphore(0);private static Semaphore s3 = new Semaphore(0);public void print1(){while(num<=MAX){try{s1.acquire();}catch (InterruptedException e){e.printStackTrace();}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);s2.release();}}public void print2(){while(num<=MAX){try{s2.acquire();}catch (InterruptedException e){e.printStackTrace();}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);s3.release();}}public void print3(){while(num<=MAX){try{s3.acquire();}catch (InterruptedException e){e.printStackTrace();}if(num>MAX){break;}System.out.println(Thread.currentThread().getName()+" "+num++);s1.release();}}}public static void main(String[] args) {PrintTask task = new PrintTask();new Thread(task::print1,"t1").start();new Thread(task::print2,"t2").start();new Thread(task::print3,"t3").start();}}

线程池

手写线程池

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;public class CustomThreadPool {private final BlockingQueue<Runnable> taskQueue;private final List<WorkerThread> workers;private final AtomicBoolean isShutdown;private final int corePoolSize;private final AtomicInteger threadCount;public CustomThreadPool(int corePoolSize) {this.corePoolSize = corePoolSize;this.taskQueue = new LinkedBlockingQueue<>();this.workers = new ArrayList<>(corePoolSize);this.isShutdown = new AtomicBoolean(false);this.threadCount = new AtomicInteger(0);// 初始化线程池中的工作线程for (int i = 0; i < corePoolSize; i++) {WorkerThread worker = new WorkerThread("Worker-" + i);workers.add(worker);worker.start();}}// 提交任务到线程池public void execute(Runnable task) {if (isShutdown.get()) {throw new IllegalStateException("线程池已关闭，无法接受新任务");}try {taskQueue.put(task);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}// 关闭线程池public void shutdown() {isShutdown.set(true);// 中断所有工作线程for (WorkerThread worker : workers) {if (!worker.isInterrupted()) {worker.interrupt();}}}// 等待所有任务完成public void awaitTermination() throws InterruptedException {for (WorkerThread worker : workers) {worker.join();}}// 获取当前队列中等待的任务数public int getQueueSize() {return taskQueue.size();}// 获取活跃线程数public int getActiveThreadCount() {return threadCount.get();}// 工作线程类private class WorkerThread extends Thread {public WorkerThread(String name) {super(name);}@Overridepublic void run() {while (!isShutdown.get() || !taskQueue.isEmpty()) {try {// 从队列中获取任务Runnable task = taskQueue.poll();if (task != null) {threadCount.incrementAndGet();try {System.out.println(Thread.currentThread().getName() + " 开始执行任务");task.run();System.out.println(Thread.currentThread().getName() + " 完成任务");} finally {threadCount.decrementAndGet();}} else {// 如果队列为空，等待新任务Thread.sleep(100);}} catch (InterruptedException e) {// 线程被中断，结束循环break;} catch (Exception e) {// 任务执行中的异常不应该影响工作线程System.err.println(Thread.currentThread().getName() + " 执行任务时发生异常: " + e.getMessage());}}System.out.println(Thread.currentThread().getName() + " 退出");}}// 测试自定义线程池public static void main(String[] args) throws InterruptedException {// 创建一个有3个线程的线程池CustomThreadPool threadPool = new CustomThreadPool(3);// 提交10个任务for (int i = 0; i < 10; i++) {final int taskId = i;threadPool.execute(() -> {try {System.out.println("执行任务 #" + taskId);// 模拟任务执行时间Thread.sleep(1000);System.out.println("任务 #" + taskId + " 执行完成");} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}// 等待5秒后关闭线程池Thread.sleep(5000);System.out.println("关闭线程池");threadPool.shutdown();// 等待所有任务完成threadPool.awaitTermination();System.out.println("所有任务已完成，程序退出");}
}

使用线程池实现任务

机场安检模拟（现实场景）

场景：

• 3个安检口（线程）同时工作
• 旅客（任务）以随机间隔（0~2秒）到达，加入队列
• 每个安检需要2秒处理时间
• 使用 ExecutorService 线程池实现

扩展要求：

• 显示实时排队人数
• 当排队超过10人时增加1个临时安检口
• 使用 ThreadPoolExecutor 动态调整

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;public class SecurityCheckSimulation {// 旅客计数器private static final AtomicInteger passengerCount = new AtomicInteger(0);// 队列中等待的旅客数量private static final AtomicInteger waitingCount = new AtomicInteger(0);// 安检口初始数量private static int initialCheckpoints = 3;// 安检口最大数量private static int maxCheckpoints = 4;public static void main(String[] args) {// 创建可调整大小的线程池ThreadPoolExecutor executor = new ThreadPoolExecutor(initialCheckpoints, // 核心线程数maxCheckpoints, // 最大线程数60L,TimeUnit.SECONDS, // 线程空闲时间new LinkedBlockingQueue<>(), // 任务队列new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略);// 模拟旅客到达Thread passengerGenerator = new Thread(()->{try{while(true){// 新旅客到达int passengerId = passengerCount.incrementAndGet();waitingCount.incrementAndGet();System.out.println("Passenger " + passengerId + " 到达安检。当前排队人数："+waitingCount.get());// 动态调整安检口数量if(waitingCount.get()>10 && executor.getCorePoolSize()<maxCheckpoints){executor.setCorePoolSize(maxCheckpoints);System.out.println("===== 排队人数过多，增加临时安检口! 当前安检口: "+executor.getCorePoolSize()+"=====");}// 将旅客提交给安检口executor.submit(()->processPassenger(passengerId));// 模拟旅客到达的随机时间间隔,随机等待0-2秒Thread.sleep((long) (Math.random()*2000));}}catch (InterruptedException e){e.printStackTrace();}});passengerGenerator.start();}private static void processPassenger(int passengerId)  {// 模拟安检过程System.out.println("Passenger " + passengerId + " 正在安检...");// 安检耗时2秒try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println("Passenger " + passengerId + " 安检完成。");waitingCount.decrementAndGet();System.out.println("当前排队人数："+waitingCount.get());}
}