【多线程】Java 多线程与并发编程全解析
作者：不在线第一只蜗牛
- 2025-05-23 福建
本文字数：10430 字
阅读完需：约 34 分钟
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[threadCount]; // 测试非线程安全的方法 counter = 0; for (int i = 0; i < threadCount; i++) { threads[i] = new Thread(ThreadSafetyExample::incrementUnsafe); threads[i].start(); } for (Thread t : threads) t.join(); System.out.println("非线程安全计数器结果: " + counter); // 可能不等于1000 // 测试原子类 for (int i = 0; i < threadCount; i++) { threads[i] = new Thread(ThreadSafetyExample::incrementAtomic); threads[i].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[10]; for (int i = 0; i < 10; i++) { threads[i] = new Thread(() -> { for (int j = 0; j < 1000; j++) { atomicInteger.incrementAndGet(); // 原子递增 } }); threads[i].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[20]; for (int i = 0; i < 20; i++) { adderThreads[i] = new Thread(() -> { for (int j = 0; j < 10000; j++) { longAdder.increment(); } }); adderThreads[i].start(); } // 等待所有线程完成 for (Thread t : adderThreads) { t.join(); } System.out.println("LongAdder最终值: " + longAdder.sum()); }}复制代码
总结
Java 多线程与并发编程是一个复杂但强大的领域，掌握这些核心概念和工具能够帮助你编写高效、安全且易于维护的多线程应用程序。
关键要点回顾：
线程的生命周期和基本操作
线程安全与同步机制（synchronized、ReentrantLock、原子类）
JUC 包中的并发工具类（Executor 框架、CountDownLatch、CyclicBarrier 等）
线程池的原理和最佳实践
并发集合类（ConcurrentHashMap、CopyOnWriteArrayList 等）
原子操作类（AtomicInteger、LongAdder 等）
通过合理使用这些工具和技术，可以有效解决多线程编程中的各种挑战，如竞态条件、内存可见性和线程管理等问题。
文章转载自：佛祖让我来巡山
划线
评论
复制
发布于: 刚刚阅读数: 3
不在线第一只蜗牛
关注
还未添加个人签名 2023-06-19 加入
还未添加个人简介
评论