# Java多线程编程：从基础到高级实践

## 线程基础概念

### 什么是线程

线程是操作系统能够进行运算调度的最小单位，它被包含在进程之中，是进程中的实际运作单位。在Java中，每个线程都拥有独立的程序计数器、虚拟机栈和本地方法栈，但共享堆内存和方法区。

### 线程创建方式

1. 继承Thread类

```java

public class MyThread extends Thread {

@Override

public void run() {

System.out.println(线程执行: + Thread.currentThread().getName());

}

public static void main(String[] args) {

MyThread thread = new MyThread();

thread.start();

}

}

```

2. 实现Runnable接口

```java

public class MyRunnable implements Runnable {

@Override

public void run() {

System.out.println(Runnable线程执行: + Thread.currentThread().getName());

}

public static void main(String[] args) {

Thread thread = new Thread(new MyRunnable());

thread.start();

}

}

```

3. 实现Callable接口

```java

public class MyCallable implements Callable {

@Override

public String call() throws Exception {

return Callable执行结果: + Thread.currentThread().getName();

}

public static void main(String[] args) throws Exception {

ExecutorService executor = Executors.newSingleThreadExecutor();

Future future = executor.submit(new MyCallable());

System.out.println(future.get());

executor.shutdown();

}

}

```

## 线程状态管理

### 线程生命周期

Java线程包含六种状态：NEW、RUNNABLE、BLOCKED、WAITING、TIMED_WAITING、TERMINATED。

```java

public class ThreadStateDemo {

public static void main(String[] args) throws InterruptedException {

Thread thread = new Thread(() -> {

try {

Thread.sleep(1000);

synchronized (ThreadStateDemo.class) {

ThreadStateDemo.class.wait();

}

} catch (InterruptedException e) {

e.printStackTrace();

}

});

System.out.println(创建后状态: + thread.getState()); // NEW

thread.start();

System.out.println(启动后状态: + thread.getState()); // RUNNABLE

Thread.sleep(100);

System.out.println(睡眠中状态: + thread.getState()); // TIMED_WAITING

Thread.sleep(2000);

System.out.println(等待中状态: + thread.getState()); // WAITING

synchronized (ThreadStateDemo.class) {

ThreadStateDemo.class.notify();

}

thread.join();

System.out.println(结束状态: + thread.getState()); // TERMINATED

}

}

```

## 线程同步机制

### synchronized关键字

同步方法

```java

public class SynchronizedCounter {

private int count = 0;

public synchronized void increment() {

count++;

}

public synchronized int getCount() {

return count;

}

}

```

同步代码块

```java

public class SynchronizedBlock {

private final Object lock = new Object();

private int value = 0;

public void increment() {

synchronized (lock) {

value++;

}

}

}

```

### Lock接口

```java

public class LockDemo {

private final ReentrantLock lock = new ReentrantLock();

private int counter = 0;

public void increment() {

lock.lock();

try {

counter++;

} finally {

lock.unlock();

}

}

public boolean tryIncrement() {

if (lock.tryLock()) {

try {

counter++;

return true;

} finally {

lock.unlock();

}

}

return false;

}

}

```

## 线程间通信

### wait/notify机制

```java

public class ProducerConsumer {

private final Queue queue = new LinkedList<>();

private final int CAPACITY = 5;

public void produce() throws InterruptedException {

int value = 0;

while (true) {

synchronized (this) {

while (queue.size() == CAPACITY) {

wait();

}

System.out.println(生产: + value);

queue.add(value++);

notify();

Thread.sleep(1000);

}

}

}

public void consume() throws InterruptedException {

while (true) {

synchronized (this) {

while (queue.isEmpty()) {

wait();

}

int value = queue.poll();

System.out.println(消费: + value);

notify();

Thread.sleep(1000);

}

}

}

}

```

### Condition条件变量

```java

public class ConditionDemo {

private final Lock lock = new ReentrantLock();

private final Condition notFull = lock.newCondition();

private final Condition notEmpty = lock.newCondition();

private final Queue queue = new LinkedList<>();

private final int CAPACITY = 5;

public void put(int value) throws InterruptedException {

lock.lock();

try {

while (queue.size() == CAPACITY) {

notFull.await();

}

queue.add(value);

notEmpty.signal();

} finally {

lock.unlock();

}

}

public int take() throws InterruptedException {

lock.lock();

try {

while (queue.isEmpty()) {

notEmpty.await();

}

int value = queue.poll();

notFull.signal();

return value;

} finally {

lock.unlock();

}

}

}

```

## 并发工具类

### CountDownLatch

```java

public class CountDownLatchDemo {

public static void main(String[] args) throws InterruptedException {

int threadCount = 5;

CountDownLatch startSignal = new CountDownLatch(1);

CountDownLatch doneSignal = new CountDownLatch(threadCount);

for (int i = 0; i < threadCount; i++) {

new Thread(() -> {

try {

startSignal.await();

System.out.println(Thread.currentThread().getName() + 开始工作);

Thread.sleep(1000);

doneSignal.countDown();

} catch (InterruptedException e) {

e.printStackTrace();

}

}).start();

}

Thread.sleep(1000);

System.out.println(主线程发出开始信号);

startSignal.countDown();

doneSignal.await();

System.out.println(所有线程工作完成);

}

}

```

### CyclicBarrier

```java

public class CyclicBarrierDemo {

public static void main(String[] args) {

int threadCount = 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(1000);

barrier.await();

System.out.println(线程 + threadId + 开始第二阶段);

Thread.sleep(1000);

barrier.await();

System.out.println(线程 + threadId + 完成工作);

} catch (Exception e) {

e.printStackTrace();

}

}).start();

}

}

}

```

### Semaphore

```java

public class SemaphoreDemo {

public static void main(String[] args) {

int permits = 3;

Semaphore semaphore = new Semaphore(permits);

int threadCount = 10;

for (int i = 0; i < threadCount; i++) {

final int threadId = i;

new Thread(() -> {

try {

semaphore.acquire();

System.out.println(线程 + threadId + 获得许可，开始工作);

Thread.sleep(2000);

System.out.println(线程 + threadId + 释放许可);

semaphore.release();

} catch (InterruptedException e) {

e.printStackTrace();

}

}).start();

}

}

}

```

## 线程池技术

### Executor框架

```java

public class ThreadPoolDemo {

public static void main(String[] args) {

// 固定大小线程池

ExecutorService fixedPool = Executors.newFixedThreadPool(5);

// 缓存线程池

ExecutorService cachedPool = Executors.newCachedThreadPool();

// 单线程池

ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();

// 定时任务线程池

ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(3);

// 自定义线程池

ThreadPoolExecutor customPool = new ThreadPoolExecutor(

2, // 核心线程数

10, // 最大线程数

60L, // 空闲线程存活时间

TimeUnit.SECONDS, // 时间单位

new ArrayBlockingQueue<>(100), // 工作队列

Executors.defaultThreadFactory(), // 线程工厂

new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略

);

// 提交任务

for (int i = 0; i < 10; i++) {

final int taskId = i;

fixedPool.submit(() -> {

System.out.println(执行任务: + taskId + 线程: + Thread.currentThread().getName());

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

});

}

fixedPool.shutdown();

}

}

```

## 并发集合

### ConcurrentHashMap

```java

public class ConcurrentHashMapDemo {

public static void main(String[] args) {

ConcurrentHashMap map = new ConcurrentHashMap<>();

// 并发添加

ExecutorService executor = Executors.newFixedThreadPool(5);

for (int i = 0; i < 10; i++) {

final int threadId = i;

executor.submit(() -> {

for (int j = 0; j < 100; j++) {

String key = key- + threadId + - + j;

map.put(key, j);

}

});

}

executor.shutdown();

try {

executor.awaitTermination(1, TimeUnit.MINUTES);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println(Map大小: + map.size());

// 原子操作

map.computeIfAbsent(new-key, k -> 100);

map.merge(new-key, 50, Integer::sum);

}

}

```

### CopyOnWriteArrayList

```java

public class CopyOnWriteDemo {

public static void main(String[] args) {

CopyOnWriteArrayList list = new CopyOnWriteArrayList<>();

// 写操作较慢，但读操作很快且线程安全

list.add(item1);

list.add(item2);

list.addIfAbsent(item1); // 不存在才添加

// 迭代过程中可以安全修改

for (String item : list) {

System.out.println(item);

if (item2.equals(item)) {

list.add(item3);

}

}

}

}

```

## 原子操作类

```java

public class AtomicDemo {

private final AtomicInteger counter = new AtomicInteger(0);

private final AtomicReference reference = new AtomicReference<>(initial);

private final AtomicLongArray longArray = new AtomicLongArray(10);

public void increment() {

// CAS操作

counter.incrementAndGet();

}

public boolean updateReference(String expect, String update) {

return reference.compareAndSet(expect, update);

}

public void updateArray(int index, long value) {

longArray.set(index, value);

}

}

```

## 高级并发模式

### Future模式

```java

public class FutureDemo {

public static void main(String[] args) throws Exception {

ExecutorService executor = Executors.newFixedThreadPool(3);

// 提交Callable任务

Future future1 = executor.submit(() -> {

Thread.sleep(2000);

return 任务1完成;

});

Future future2 = executor.submit(() -> {

Thread.sleep(1000);

return 42;

});

// 非阻塞获取结果

while (!future1.isDone() || !future2.isDone()) {

if (future2.isDone()) {

System.out.println(任务2结果: + future2.get());

}

Thread.sleep(100);

}

System.out.println(任务1结果: + future1.get());

executor.shutdown();

}

}

```

### CompletableFuture

```java

public class CompletableFutureDemo {

public static void main(String[] args) throws Exception {

// 异步执行

CompletableFuture future = CompletableFuture.supplyAsync(() -> {

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

return Hello;

});

// 链式调用

CompletableFuture result = future

.thenApply(s -> s + World)

.thenApply(String::toUpperCase)

.thenCompose(s -> CompletableFuture.supplyAsync(() -> s + !))

.exceptionally(ex -> 错误: + ex.getMessage());

System.out.println(最终结果: + result.get());

// 组合多个Future

CompletableFuture future1 = CompletableFuture.supplyAsync(() -> Task1);

CompletableFuture future2 = CompletableFuture.supplyAsync(() -> Task2);

CompletableFuture combined = future1.thenCombine(future2, (r1, r2) -> r1 + & + r2);

System.out.println(组合结果: + combined.get());

}

}

```

## 性能优化与最佳实践

### 避免死锁

```java

public class DeadlockPrevention {

private final Object lock1 = new Object();

private final Object lock2 = new Object();

public void method1() {

synchronized (lock1) {

System.out.println(获得lock1);

synchronized (lock2) {

System.out.println(获得lock2);

}

}

}

public void method2() {

// 使用相同的锁顺序避免死锁

synchronized (lock1) {

System.out.println(获得lock1);

synchronized (lock2) {

System.out.println(获得lock2);

}

}

}

}

```

### 线程局部变量

```java

public class ThreadLocalDemo {

private static final ThreadLocal dateFormatThreadLocal =

ThreadLocal.withInitial(() -> new SimpleDateFormat(yyyy-MM-dd));

private static final ThreadLocal userThreadLocal = new ThreadLocal<>();

public void processUser(int userId) {

userThreadLocal.set(userId);

try {

// 使用线程局部变量

System.out.println(处理用户: + userThreadLocal.get());

} finally {

userThreadLocal.remove(); // 防止内存泄漏

}

}

}

```

### 性能考虑

```java

public class PerformanceConsiderations {

// 减少锁粒度

private final Object[] locks;

private final int[] data;

public PerformanceConsiderations(int size) {

data = new int[size];

locks = new Object[size];

for (int i = 0; i < size; i++) {

locks[i] = new Object();

}

}

public void update(int index, int value) {

synchronized (locks[index]) { // 细粒度锁

data[index] = value;

}

}

// 使用读写锁提高读性能

private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

private final List items = new ArrayList<>();

public void addItem(String item) {

readWriteLock.writeLock().lock();

try {

items.add(item);

} finally {

readWriteLock.writeLock().unlock();

}

}

public String getItem(int index) {

readWriteLock.readLock().lock();

try {

return items.get(index);

} finally {

readWriteLock.readLock().unlock();

}

}

}

```

通过系统学习Java多线程编程的基础概念、同步机制、并发工具类和高级模式，开发者可以构建出高效、可靠的并发应用程序。在实际开发中，需要根据具体场景选择合适的并发策略，并注意线程安全、性能优化和资源管理等问题。