并发编程的核心模式包括10种关键方法，每种都有特定用途和适用场景。1.互斥锁（mutex）通过reentrantlock确保线程对共享资源的独占访问；2.读写锁（readwritelock）允许多个线程同时读取、单个写入，提高读多写少场景效率；3.信号量（semaphore）控制并发访问数量，限制资源使用上限；4.countdownlatch用于等待多个线程任务完成后再继续执行；5.cyclicbarrier使一组线程互相等待到达共同屏障点后继续运行；6.future/callable实现异步任务执行并获取结果；7.线程池（threadpool）复用线程降低创建销毁开销；8.blockingqueue提供线程安全的数据传递机制；9.原子变量（atomic variables）如atomicinteger实现无锁线程安全操作；10.fork/join框架支持任务拆分与合并，适用于并行计算。选择合适的并发模式需结合具体需求，例如简单共享变量可用原子类，读多写少用读写锁。避免死锁应破坏其四个必要条件，如避免嵌套锁、采用锁排序、设置超时机制或使用死锁检测工具。此外，静态分析、测试和性能分析工具也能辅助并发编程优化。掌握这些模式和工具能显著提升并发程序的稳定性和效率。
并发编程旨在提高程序效率，但稍有不慎，就会掉入各种陷阱。掌握一些核心模式，能帮助我们更好地应对并发挑战。
解决方案
并发编程的核心模式涵盖了线程管理、数据同步、任务调度等多个方面。以下列出十大核心模式，并提供代码模板，希望能帮助你快速上手：
1. 互斥锁（Mutex）
互斥锁用于保护临界区，确保同一时间只有一个线程可以访问共享资源。
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class MutexExample {
private final Lock lock = new ReentrantLock();
private int count = 0;
public void increment() {
lock.lock();
try {
count++;
} finally {
lock.unlock();
}
}
public int getCount() {
return count;
}
public static void main(String[] args) throws InterruptedException {
MutexExample example = new MutexExample();
Thread t1 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
example.increment();
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
example.increment();
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println("Count: " + example.getCount()); // Expected: 2000
}
}2. 读写锁（ReadWriteLock）
读写锁允许多个线程同时读取共享资源，但只允许一个线程写入。
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class ReadWriteLockExample {
private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
private int data = 0;
public int readData() {
rwLock.readLock().lock();
try {
return data;
} finally {
rwLock.readLock().unlock();
}
}
public void writeData(int newData) {
rwLock.writeLock().lock();
try {
data = newData;
} finally {
rwLock.writeLock().unlock();
}
}
public static void main(String[] args) {
ReadWriteLockExample example = new ReadWriteLockExample();
// Writer thread
new Thread(() -> {
for (int i = 0; i < 5; i++) {
example.writeData(i);
System.out.println("Writer: Wrote " + i);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
// Reader threads
for (int i = 0; i < 3; i++) {
new Thread(() -> {
for (int j = 0; j < 10; j++) {
int value = example.readData();
System.out.println("Reader " + Thread.currentThread().getName() + ": Read " + value);
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "Reader-" + i).start();
}
}
}3. 信号量（Semaphore）
信号量用于控制同时访问某个资源的线程数量。
import java.util.concurrent.Semaphore;
public class SemaphoreExample {
private static final int MAX_PERMITS = 3;
private final Semaphore semaphore = new Semaphore(MAX_PERMITS);
public void accessResource(int threadId) {
try {
semaphore.acquire();
System.out.println("Thread " + threadId + " acquired permit.");
// Simulate resource access
Thread.sleep((long) (Math.random() * 1000));
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("Thread " + threadId + " releasing permit.");
semaphore.release();
}
}
public static void main(String[] args) {
SemaphoreExample example = new SemaphoreExample();
for (int i = 0; i < 5; i++) {
final int threadId = i;
new Thread(() -> example.accessResource(threadId)).start();
}
}
}4. CountDownLatch
CountDownLatch允许一个或多个线程等待其他线程完成操作。
import java.util.concurrent.CountDownLatch;
public class CountDownLatchExample {
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {
final int taskNumber = i;
new Thread(() -> {
try {
System.out.println("Task " + taskNumber + " is running...");
Thread.sleep((long) (Math.random() * 2000));
System.out.println("Task " + taskNumber + " completed.");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown();
}
}).start();
}
latch.await(); // Wait for all tasks to complete
System.out.println("All tasks completed. Main thread continues.");
}
}5. CyclicBarrier
CyclicBarrier允许一组线程互相等待，直到所有线程都到达某个屏障点。
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
public static void main(String[] args) {
CyclicBarrier barrier = new CyclicBarrier(3, () -> System.out.println("All threads reached barrier!"));
for (int i = 0; i < 3; i++) {
final int threadId = i;
new Thread(() -> {
try {
System.out.println("Thread " + threadId + " is working...");
Thread.sleep((long) (Math.random() * 2000));
System.out.println("Thread " + threadId + " reached the barrier.");
barrier.await(); // Wait for other threads
System.out.println("Thread " + threadId + " continues after barrier.");
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
}6. Future/Callable
Future/Callable 允许异步执行任务并获取结果。
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class FutureCallableExample {
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(2);
Callable<String> task = () -> {
System.out.println("Task is running...");
Thread.sleep(1000);
return "Task completed!";
};
Future<String> future = executor.submit(task);
System.out.println("Waiting for task to complete...");
String result = future.get(); // Blocks until the task is done
System.out.println("Result: " + result);
executor.shutdown();
}
}7. 线程池（ThreadPool）
线程池用于管理和复用线程，减少线程创建和销毁的开销。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExample {
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
final int taskNumber = i;
executor.execute(() -> {
System.out.println("Task " + taskNumber + " is running in thread: " + Thread.currentThread().getName());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Task " + taskNumber + " completed.");
});
}
executor.shutdown();
}
}8. BlockingQueue
BlockingQueue 是一种线程安全的队列，可以在多个线程之间传递数据。
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class BlockingQueueExample {
public static void main(String[] args) {
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);
// Producer thread
new Thread(() -> {
try {
for (int i = 0; i < 15; i++) {
queue.put(i); // Blocks if queue is full
System.out.println("Produced: " + i);
Thread.sleep(100);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// Consumer thread
new Thread(() -> {
try {
while (true) {
Integer value = queue.take(); // Blocks if queue is empty
System.out.println("Consumed: " + value);
Thread.sleep(200);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}9. 原子变量（Atomic Variables）
原子变量提供线程安全的原子操作，避免使用锁。
import java.util.concurrent.atomic.AtomicInteger;
public class AtomicIntegerExample {
private AtomicInteger count = new AtomicInteger(0);
public void increment() {
count.incrementAndGet();
}
public int getCount() {
return count.get();
}
public static void main(String[] args) throws InterruptedException {
AtomicIntegerExample example = new AtomicIntegerExample();
Thread t1 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
example.increment();
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
example.increment();
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println("Count: " + example.getCount()); // Expected: 2000
}
}10. Fork/Join
Fork/Join 框架用于将大任务分解成小任务并行执行，然后合并结果。
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
public class ForkJoinExample {
static class SumTask extends RecursiveTask<Long> {
private static final int THRESHOLD = 100;
private final long[] array;
private final int start;
private final int end;
public SumTask(long[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
if (end - start <= THRESHOLD) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
} else {
int middle = (start + end) / 2;
SumTask leftTask = new SumTask(array, start, middle);
SumTask rightTask = new SumTask(array, middle, end);
leftTask.fork();
rightTask.fork();
return leftTask.join() + rightTask.join();
}
}
}
public static void main(String[] args) {
long[] array = new long[1000];
for (int i = 0; i < array.length; i++) {
array[i] = i + 1;
}
ForkJoinPool pool = new ForkJoinPool();
SumTask task = new SumTask(array, 0, array.length);
long result = pool.invoke(task);
System.out.println("Sum: " + result); // Expected: 500500
}
}如何选择合适的并发模式？
选择并发模式不能一概而论，需要根据具体场景来判断。比如，如果只需要保护一个简单的共享变量，原子变量可能就足够了，没必要动用互斥锁。如果读多写少，读写锁会更高效。
并发编程中常见的死锁问题如何避免？
死锁是并发编程中的一大难题。避免死锁的关键在于破坏死锁产生的四个必要条件：互斥、占有且等待、不可剥夺、环路等待。常用的方法包括：
- 避免嵌套锁： 尽量减少锁的持有时间，避免在一个锁的范围内获取另一个锁。
- 锁排序： 如果需要获取多个锁，按照固定的顺序获取，避免环路等待。
- 使用超时机制： 在获取锁时设置超时时间，如果超时未获取到锁，则释放已持有的锁。
- 死锁检测： 定期检测系统中是否存在死锁，如果发现死锁，则采取措施解除死锁。
除了代码模板，还有哪些工具可以辅助并发编程？
除了代码模板，还有一些工具可以辅助并发编程，例如：
- 静态分析工具： 可以帮助检查代码中的潜在并发问题，例如死锁、竞争条件等。
- 并发测试工具： 可以模拟高并发场景，帮助发现并发代码中的性能瓶颈和错误。
- 性能分析工具： 可以帮助分析并发程序的性能，找出性能瓶颈。
掌握这些核心模式和工具，能让你在并发编程的道路上走得更稳、更远。
