【多线程】Java多线程与并发编程全解析
Java多线程与并发编程全解析多线程编程是Java中最具挑战性的部分之一，它能够显著提升应用程序的性能和响应能力。本文将全面解析Java多线程与并发编程的核心概念、线程安全机制以及JUC工具类的使用，并提供完整的代码示例。
1. 线程的基本操作与生命周期
Java线程的生命周期包括新建(New)、就绪(Runnable)、运行(Running)、阻塞(Blocked)、等待(Waiting)、超时等待(Timed Waiting)和终止(Terminated)七个状态。
public class ThreadLifecycleExample {
public static void main(String[] args) throws InterruptedException {
// 创建线程
Thread t = new Thread(() -> {
System.out.println("线程状态1: " + Thread.currentThread().getState()); // RUNNABLE
try {
// 线程休眠，进入TIMED_WAITING状态
Thread.sleep(1000);
System.out.println("线程状态2: " + Thread.currentThread().getState());
// 同步块，可能进入BLOCKED状态
synchronized (ThreadLifecycleExample.class) {
System.out.println("线程获得锁");
}
// 线程等待，进入WAITING状态
synchronized (ThreadLifecycleExample.class) {
ThreadLifecycleExample.class.wait();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("线程状态3: " + Thread.currentThread().getState()); // RUNNABLE
});
System.out.println("线程状态0: " + t.getState()); // NEW
// 启动线程
t.start();
System.out.println("线程状态4: " + t.getState()); // RUNNABLE或TIMED_WAITING
// 主线程休眠
Thread.sleep(2000);
System.out.println("线程状态5: " + t.getState()); // 可能是WAITING或TERMINATED
// 唤醒等待的线程
synchronized (ThreadLifecycleExample.class) {
ThreadLifecycleExample.class.notify();
}
// 等待线程执行完毕
t.join();
System.out.println("线程状态6: " + t.getState()); // TERMINATED
}
}2. 线程安全与同步机制
线程安全问题主要由竞态条件(Race Condition)和内存可见性问题引起。Java提供了多种同步机制来解决这些问题。
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ThreadSafetyExample {
private static int counter = 0; // 共享资源
private static final Object lock = new Object(); // 锁对象
private static final Lock reentrantLock = new ReentrantLock(); // 可重入锁
// 方式1: synchronized方法
public static synchronized void incrementSynchronized() {
counter++;
}
// 方式2: synchronized块
public static void incrementBlock() {
synchronized (lock) {
counter++;
}
}
// 方式3: ReentrantLock
public static void incrementReentrantLock() {
reentrantLock.lock();
try {
counter++;
} finally {
reentrantLock.unlock();
}
}
// 方式4: 使用原子类
private static java.util.concurrent.atomic.AtomicInteger atomicCounter = new java.util.concurrent.atomic.AtomicInteger(0);
public static void incrementAtomic() {
atomicCounter.incrementAndGet();
}
// 演示线程不安全的情况
public static void incrementUnsafe() {
counter++; // 非线程安全操作
}
public static void main(String[] args) throws InterruptedException {
int threadCount = 1000;
Thread[] threads = new Thread;
// 测试非线程安全的方法
counter = 0;
for (int i = 0; i < threadCount; i++) {
threads = new Thread(ThreadSafetyExample::incrementUnsafe);
threads.start();
}
for (Thread t : threads) t.join();
System.out.println("非线程安全计数器结果: " + counter); // 可能不等于1000
// 测试原子类
for (int i = 0; i < threadCount; i++) {
threads = new Thread(ThreadSafetyExample::incrementAtomic);
threads.start();
}
for (Thread t : threads) t.join();
System.out.println("原子类计数器结果: " + atomicCounter.get()); // 一定等于1000
}
}3. JUC包中的并发工具类
JUC(java.util.concurrent)包提供了丰富的并发工具类，极大简化了多线程编程。
3.1 Executor框架与线程池
Executor框架是管理线程的核心组件，线程池是其主要实现。
import java.util.concurrent.*;
public class ExecutorFrameworkExample {
public static void main(String[] args) throws InterruptedException {
// 创建固定大小的线程池
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
// 创建缓存线程池
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
// 创建单线程执行器
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
// 创建定时任务线程池
ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(2);
// 提交任务到固定大小线程池
for (int i = 0; i < 5; i++) {
final int taskId = i;
fixedThreadPool.submit(() -> {
System.out.println("任务" + taskId + "在固定大小线程池执行");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
// 提交定时任务
scheduledExecutor.schedule(() -> {
System.out.println("延迟3秒执行的定时任务");
}, 3, TimeUnit.SECONDS);
// 提交周期性任务
scheduledExecutor.scheduleAtFixedRate(() -> {
System.out.println("每2秒执行一次的周期性任务");
}, 1, 2, TimeUnit.SECONDS);
// 关闭线程池
fixedThreadPool.shutdown();
cachedThreadPool.shutdown();
singleThreadExecutor.shutdown();
// 等待定时任务执行一段时间后关闭
Thread.sleep(10000);
scheduledExecutor.shutdown();
}
}3.2 CountDownLatch
CountDownLatch用于让一个或多个线程等待其他线程完成操作。
import java.util.concurrent.CountDownLatch;
public class CountDownLatchExample {
public static void main(String[] args) throws InterruptedException {
int workerCount = 5;
CountDownLatch latch = new CountDownLatch(workerCount);
// 创建并启动工作线程
for (int i = 0; i < workerCount; i++) {
final int workerId = i;
new Thread(() -> {
System.out.println("工作线程" + workerId + "开始执行");
try {
// 模拟工作耗时
Thread.sleep((long) (Math.random() * 5000));
System.out.println("工作线程" + workerId + "完成任务");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 计数减1
latch.countDown();
}
}).start();
}
// 主线程等待所有工作线程完成
System.out.println("主线程等待所有工作线程完成...");
latch.await();
System.out.println("所有工作线程已完成，主线程继续执行");
}
}3.3 CyclicBarrier
CyclicBarrier用于多个线程互相等待，直到所有线程都到达某个屏障点。
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
public static void main(String[] args) {
int threadCount = 3;
// 创建CyclicBarrier，当3个线程都到达屏障时执行回调
CyclicBarrier barrier = new CyclicBarrier(threadCount, () -> {
System.out.println("所有线程都已到达屏障，继续执行");
});
// 创建并启动线程
for (int i = 0; i < threadCount; i++) {
final int threadId = i;
new Thread(() -> {
try {
System.out.println("线程" + threadId + "正在执行前置任务");
Thread.sleep((long) (Math.random() * 3000));
System.out.println("线程" + threadId + "已到达屏障");
// 等待其他线程到达屏障
barrier.await();
System.out.println("线程" + threadId + "继续执行后续任务");
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
}3.4 Semaphore
Semaphore用于控制同时访问某个资源的线程数量。
import java.util.concurrent.Semaphore;
public class SemaphoreExample {
private static final int MAX_PERMITS = 3; // 最多允许3个线程同时访问
private static final Semaphore semaphore = new Semaphore(MAX_PERMITS);
public static void main(String[] args) {
// 创建10个线程，但最多只允许3个同时执行
for (int i = 0; i < 10; i++) {
final int threadId = i;
new Thread(() -> {
try {
// 获取许可
semaphore.acquire();
System.out.println("线程" + threadId + "获取到许可，开始执行");
// 模拟执行任务
Thread.sleep((long) (Math.random() * 5000));
System.out.println("线程" + threadId + "执行完毕，释放许可");
// 释放许可
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
}3.5 Exchanger
Exchanger用于两个线程之间交换数据。
import java.util.concurrent.Exchanger;
public class ExchangerExample {
public static void main(String[] args) {
Exchanger<String> exchanger = new Exchanger<>();
// 生产者线程
new Thread(() -> {
try {
String dataToSend = "来自生产者的数据";
System.out.println("生产者发送: " + dataToSend);
// 交换数据
String receivedData = exchanger.exchange(dataToSend);
System.out.println("生产者收到: " + receivedData);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// 消费者线程
new Thread(() -> {
try {
String dataToSend = "来自消费者的数据";
System.out.println("消费者发送: " + dataToSend);
// 交换数据
String receivedData = exchanger.exchange(dataToSend);
System.out.println("消费者收到: " + receivedData);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}3.6 Future和CompletableFuture
Future用于异步获取计算结果，CompletableFuture是Future的增强版，提供了更丰富的异步编程功能。
import java.util.concurrent.*;
public class FutureExample {
public static void main(String[] args) throws InterruptedException, ExecutionException {
ExecutorService executor = Executors.newSingleThreadExecutor();
// 使用Future
Future<Integer> future = executor.submit(() -> {
// 模拟耗时计算
Thread.sleep(2000);
return 1 + 2;
});
// 主线程可以做其他事情
System.out.println("主线程继续执行");
// 获取异步计算结果
if (future.isDone()) {
System.out.println("计算已完成，结果: " + future.get());
} else {
System.out.println("计算未完成，等待...");
System.out.println("计算结果: " + future.get()); // 阻塞直到计算完成
}
// 使用CompletableFuture
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 3 + 4;
});
// 链式调用，处理计算结果
completableFuture
.thenApply(result -> result * 2)
.thenAccept(finalResult -> System.out.println("CompletableFuture最终结果: " + finalResult));
// 多任务组合
CompletableFuture<Integer> task1 = CompletableFuture.supplyAsync(() -> 10);
CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> 20);
CompletableFuture<Void> allTasks = CompletableFuture.allOf(task1, task2);
CompletableFuture<Integer> combinedResult = allTasks.thenApply(v -> {
try {
return task1.get() + task2.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
return 0;
}
});
System.out.println("组合任务结果: " + combinedResult.get());
executor.shutdown();
}
}4. 线程池的原理与最佳实践
线程池通过复用线程减少线程创建和销毁的开销，提高性能。
import java.util.concurrent.*;
public class ThreadPoolBestPractice {
public static void main(String[] args) {
// 自定义线程池配置
ThreadPoolExecutor executor = new ThreadPoolExecutor(
5, // 核心线程数
10, // 最大线程数
60, // 线程空闲时间
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(100), // 任务队列
Executors.defaultThreadFactory(), // 线程工厂
new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
);
// 提交任务
for (int i = 0; i < 20; i++) {
final int taskId = i;
executor.submit(() -> {
System.out.println("任务" + taskId + "由线程" + Thread.currentThread().getName() + "执行");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
// 监控线程池状态
System.out.println("线程池状态: 核心线程数=" + executor.getCorePoolSize() +
", 最大线程数=" + executor.getMaximumPoolSize() +
", 当前线程数=" + executor.getPoolSize() +
", 活跃线程数=" + executor.getActiveCount() +
", 队列任务数=" + executor.getQueue().size());
// 关闭线程池
executor.shutdown(); // 不再接受新任务，但会执行完已提交的任务
try {
// 等待所有任务完成
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
executor.shutdownNow(); // 强制关闭
}
} catch (InterruptedException e) {
executor.shutdownNow();
}
System.out.println("线程池已关闭");
}
}5. 并发集合类
JUC包提供了多种线程安全的集合类，替代了传统的同步集合。
import java.util.*;
import java.util.concurrent.*;
public class ConcurrentCollectionExample {
public static void main(String[] args) throws InterruptedException {
// ConcurrentHashMap示例
ConcurrentHashMap<String, Integer> concurrentMap = new ConcurrentHashMap<>();
// 多个线程同时操作map
Thread t1 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
concurrentMap.put("key" + i, i);
}
});
Thread t2 = new Thread(() -> {
for (int i = 0; i < 1000; i++) {
concurrentMap.get("key" + i);
}
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println("ConcurrentHashMap大小: " + concurrentMap.size());
// CopyOnWriteArrayList示例
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
Thread writer = new Thread(() -> {
for (int i = 0; i < 100; i++) {
list.add("element" + i);
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread reader = new Thread(() -> {
for (int i = 0; i < 20; i++) {
System.out.println("List内容: " + list);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
writer.start();
reader.start();
writer.join();
reader.join();
// ConcurrentLinkedQueue示例
ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
// 生产者线程
Thread producer = new Thread(() -> {
for (int i = 0; i < 10; i++) {
queue.offer("item" + i);
System.out.println("生产: " + "item" + i);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
// 消费者线程
Thread consumer = new Thread(() -> {
while (true) {
String item = queue.poll();
if (item != null) {
System.out.println("消费: " + item);
} else if (producer.getState() == Thread.State.TERMINATED) {
break; // 生产者已结束且队列为空
}
try {
Thread.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
}
}6. 原子操作类
原子操作类基于CAS(Compare-And-Swap)实现，提供了高效的线程安全操作。
import java.util.concurrent.atomic.*;
public class AtomicExample {
public static void main(String[] args) throws InterruptedException {
// AtomicInteger示例
AtomicInteger atomicInteger = new AtomicInteger(0);
// 多个线程同时递增
Thread[] threads = new Thread;
for (int i = 0; i < 10; i++) {
threads = new Thread(() -> {
for (int j = 0; j < 1000; j++) {
atomicInteger.incrementAndGet(); // 原子递增
}
});
threads.start();
}
// 等待所有线程完成
for (Thread t : threads) {
t.join();
}
System.out.println("AtomicInteger最终值: " + atomicInteger.get()); // 应输出10000
// AtomicReference示例
AtomicReference<String> atomicReference = new AtomicReference<>("初始值");
Thread t1 = new Thread(() -> {
boolean updated = atomicReference.compareAndSet("初始值", "新值1");
System.out.println("线程1更新结果: " + updated);
});
Thread t2 = new Thread(() -> {
boolean updated = atomicReference.compareAndSet("初始值", "新值2");
System.out.println("线程2更新结果: " + updated);
});
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println("AtomicReference最终值: " + atomicReference.get());
// LongAdder示例 - 高并发场景下比AtomicLong更高效
LongAdder longAdder = new LongAdder();
Thread[] adderThreads = new Thread;
for (int i = 0; i < 20; i++) {
adderThreads = new Thread(() -> {
for (int j = 0; j < 10000; j++) {
longAdder.increment();
}
});
adderThreads.start();
}
// 等待所有线程完成
for (Thread t : adderThreads) {
t.join();
}
System.out.println("LongAdder最终值: " + longAdder.sum());
}
}总结
Java多线程与并发编程是一个复杂但强大的领域，掌握这些核心概念和工具能够帮助你编写高效、安全且易于维护的多线程应用程序。
关键要点回顾：
[*]线程的生命周期和基本操作
[*]线程安全与同步机制（synchronized、ReentrantLock、原子类）
[*]JUC包中的并发工具类（Executor框架、CountDownLatch、CyclicBarrier等）
[*]线程池的原理和最佳实践
[*]并发集合类（ConcurrentHashMap、CopyOnWriteArrayList等）
[*]原子操作类（AtomicInteger、LongAdder等）
通过合理使用这些工具和技术，可以有效解决多线程编程中的各种挑战，如竞态条件、内存可见性和线程管理等问题。
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]