Java 多线程详解

维度	进程	线程
资源分配	独立的内存空间	共享进程内存
通信方式	IPC（管道、消息队列等）	直接读写共享变量
创建开销	大（需要分配资源）	小（共享资源）
切换开销	大（需要切换地址空间）	小（只切换上下文）
影响范围	崩溃不影响其他进程	崩溃导致整个进程终止
数据共享	困难（需要 IPC）	容易（共享内存）
适用场景	独立应用程序	同一应用内的并发任务

并发（单核）： 时间 → [任务 A] [任务 B] [任务 A] [任务 C] [任务 B] ... ↑ 快速切换，看起来像同时执行 
并行（多核）： 核心 1： [任务 A] [任务 A] [任务 A] [任务 A] ...
核心 2： [任务 B] [任务 B] [任务 B] [任务 B] ...
核心 3： [任务 C] [任务 C] [任务 C] [任务 C] ...
↑ 真正同时执行

维度	并发	并行
执行方式	交替执行	同时执行
CPU 要求	单核或多核	必须多核
目的	提高响应速度	提高吞吐量
实现	时间片轮转	多个 CPU 核心

public class MyThread extends Thread {
    @Override
    public void run() {
        for (int i = 0; i < 5; i++) {
            System.out.println(Thread.currentThread().getName() + ": " + i);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class Main {
    public static void main(String[] args) {
        MyThread t1 = new MyThread();
        MyThread t2 = new MyThread();
        t1.setName("线程 1");
        t2.setName("线程 2");
        t1.start(); // 启动线程
        t2.start();
    }
}

public class MyRunnable implements Runnable {
    @Override
    public void run() {
        for (int i = 0; i < 5; i++) {
            System.out.println(Thread.currentThread().getName() + ": " + i);
        }
    }
}

public class Main {
    public static void main(String[] args) {
        MyRunnable runnable = new MyRunnable();
        Thread t1 = new Thread(runnable, "线程 1");
        Thread t2 = new Thread(runnable, "线程 2");
        t1.start();
        t2.start();
    }
}

import java.util.concurrent.*;

public class MyCallable implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {
        int sum = 0;
        for (int i = 1; i <= 100; i++) {
            sum += i;
        }
        return sum;
    }
}

public class Main {
    public static void main(String[] args) throws Exception {
        MyCallable callable = new MyCallable();
        FutureTask<Integer> task = new FutureTask<>(callable);
        Thread thread = new Thread(task);
        thread.start(); // 获取返回值（会阻塞）
        Integer result = task.get();
        System.out.println("结果：" + result);
    }
}

public class Main {
    public static void main(String[] args) {
        // 方式 1：Runnable
        new Thread(() -> {
            System.out.println("Lambda 线程执行");
        }).start();
        // 方式 2：Callable
        FutureTask<String> task = new FutureTask<>(() -> {
            return "Lambda 返回值";
        });
        new Thread(task).start();
    }
}

方式	优点	缺点	使用场景
继承 Thread	简单直接	不能继承其他类	简单任务
实现 Runnable	可继承其他类，资源共享	无返回值	常规任务
实现 Callable	有返回值，可抛异常	代码复杂	需要返回值

import java.util.concurrent.*;
import java.util.*;

public class ThreadCreationPractice {
    public static void main(String[] args) throws Exception {
        // 案例 1：多线程计算数组和
        int[] arr = new int[1000];
        for (int i = 0; i < arr.length; i++) {
            arr[i] = i + 1;
        }
        long sum = parallelSum(arr, 4);
        System.out.println("数组和：" + sum);

        // 案例 2：多线程文件处理
        List<String> files = Arrays.asList("file1.txt", "file2.txt", "file3.txt");
        processFilesInParallel(files);

        // 案例 3：超时控制
        String result = executeWithTimeout(() -> {
            Thread.sleep(2000);
            return "任务完成";
        }, 3, TimeUnit.SECONDS);
        System.out.println(result);
    }

    // 并行计算数组和
    public static long parallelSum(int[] arr, int threadCount) throws Exception {
        int chunkSize = arr.length / threadCount;
        List<Future<Long>> futures = new ArrayList<>();
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        for (int i = 0; i < threadCount; i++) {
            final int start = i * chunkSize;
            final int end = (i == threadCount - 1) ? arr.length : (i + 1) * chunkSize;
            Future<Long> future = executor.submit(() -> {
                long sum = 0;
                for (int j = start; j < end; j++) {
                    sum += arr[j];
                }
                return sum;
            });
            futures.add(future);
        }
        long totalSum = 0;
        for (Future<Long> future : futures) {
            totalSum += future.get();
        }
        executor.shutdown();
        return totalSum;
    }

    // 并行处理文件
    public static void processFilesInParallel(List<String> files) {
        List<Thread> threads = new ArrayList<>();
        for (String file : files) {
            Thread thread = new Thread(() -> {
                System.out.println("处理文件：" + file + ", 线程：" + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000); // 模拟处理
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "FileProcessor-" + file);
            threads.add(thread);
            thread.start();
        }
        // 等待所有线程完成
        for (Thread thread : threads) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("所有文件处理完成");
    }

    // 带超时的任务执行
    public static <T> T executeWithTimeout(Callable<T> task, long timeout, TimeUnit unit) throws Exception {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        Future<T> future = executor.submit(task);
        try {
            return future.get(timeout, unit);
        } catch (TimeoutException e) {
            future.cancel(true);
            throw new RuntimeException("任务超时");
        } finally {
            executor.shutdown();
        }
    }
}

NEW（新建） ↓ start() 
RUNNABLE（就绪/运行） ↓ 获取锁失败 / wait() / sleep() / join() 
BLOCKED（阻塞）/ WAITING（等待）/ TIMED_WAITING（超时等待） ↓ 获取锁 / notify() / 超时 
RUNNABLE ↓ run() 执行完毕 
TERMINATED（终止）

public class ThreadMethodDemo {
    public static void main(String[] args) throws InterruptedException {
        Thread thread = new Thread(() -> {
            System.out.println("线程执行中");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                System.out.println("线程被中断");
            }
        });

        // 启动线程
        thread.start();

        // 线程名称
        System.out.println("线程名：" + thread.getName());

        // 线程状态
        System.out.println("状态：" + thread.getState());

        // 线程优先级（1-10，默认 5）
        thread.setPriority(Thread.MAX_PRIORITY);

        // 是否存活
        System.out.println("存活：" + thread.isAlive());

        // 等待线程结束
        thread.join();
        System.out.println("主线程结束");
    }
}

特性	sleep()	wait()
所属类	Thread	Object
释放锁	不释放	释放
使用位置	任何地方	同步代码块/方法
唤醒方式	自动（超时）	notify()/notifyAll()

public class UnsafeDemo {
    private static int count = 0;

    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                count++; // 非原子操作
            }
        });
        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                count++;
            }
        });
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println("count: " + count); // 结果小于 20000
    }
}

public class SafeCounter {
    private int count = 0;

    // 同步方法
    public synchronized void increment() {
        count++;
    }

    public synchronized int getCount() {
        return count;
    }
}

public class SafeDemo {
    private int count = 0;
    private Object lock = new Object();

    public void increment() {
        synchronized (lock) {
            count++;
        }
    }

    // 同步 this 对象
    public void decrement() {
        synchronized (this) {
            count--;
        }
    }
}

public class StaticSync {
    private static int count = 0;

    // 锁的是类对象
    public static synchronized void increment() {
        count++;
    }
}

import java.util.concurrent.locks.*;

public class LockDemo {
    private int count = 0;
    private Lock lock = new ReentrantLock();

    public void increment() {
        lock.lock();
        try {
            count++;
        } finally {
            lock.unlock(); // 必须在 finally 中释放
        }
    }

    // 尝试获取锁
    public void tryIncrement() {
        if (lock.tryLock()) {
            try {
                count++;
            } finally {
                lock.unlock();
            }
        } else {
            System.out.println("获取锁失败");
        }
    }
}

import java.util.concurrent.locks.*;

public class ReadWriteLockDemo {
    private int value = 0;
    private ReadWriteLock rwLock = new ReentrantReadWriteLock();

    // 读操作
    public int read() {
        rwLock.readLock().lock();
        try {
            return value;
        } finally {
            rwLock.readLock().unlock();
        }
    }

    // 写操作
    public void write(int value) {
        rwLock.writeLock().lock();
        try {
            this.value = value;
        } finally {
            rwLock.writeLock().unlock();
        }
    }
}

public class VolatileDemo {
    private volatile boolean flag = false;

    public void writer() {
        flag = true; // 写操作立即可见
    }

    public void reader() {
        while (!flag) { // 等待 flag 变为 true
        }
        System.out.println("flag 已改变");
    }
}

特性	volatile	synchronized
原子性	❌ 不保证	✅ 保证
可见性	✅ 保证	✅ 保证
有序性	✅ 保证	✅ 保证
性能	高（无锁）	低（加锁）
使用场景	状态标志	复合操作

import java.util.concurrent.atomic.*;

public class AtomicDemo {
    private AtomicInteger count = new AtomicInteger(0);
    private AtomicLong longValue = new AtomicLong(0);
    private AtomicBoolean flag = new AtomicBoolean(false);

    public void increment() {
        count.incrementAndGet(); // 原子操作：count++
    }

    public void decrement() {
        count.decrementAndGet(); // 原子操作：count--
    }

    public void addValue(int delta) {
        count.addAndGet(delta); // 原子操作：count += delta
    }

    public boolean compareAndSet(int expect, int update) {
        return count.compareAndSet(expect, update); // CAS 操作
    }

    public static void main(String[] args) throws InterruptedException {
        AtomicDemo demo = new AtomicDemo();

        // 创建 10 个线程，每个线程增加 1000 次
        Thread[] threads = new Thread[10];
        for (int i = 0; i < 10; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    demo.increment();
                }
            });
            threads[i].start();
        }

        // 等待所有线程完成
        for (Thread thread : threads) {
            thread.join();
        }
        System.out.println("最终结果：" + demo.count.get()); // 10000
    }
}

类型	说明	示例
`AtomicInteger`	原子整数	`count.incrementAndGet()`
`AtomicLong`	原子长整数	`value.addAndGet(10)`
`AtomicBoolean`	原子布尔	`flag.compareAndSet(false, true)`
`AtomicReference<T>`	原子引用	`ref.set(new Object)`
`LongAdder`	高性能计数器	`adder.increment()`

public class DeadlockDemo {
    private static Object lock1 = new Object();
    private static Object lock2 = new Object();

    public static void main(String[] args) {
        // 线程 1：先获取 lock1，再获取 lock2
        Thread t1 = new Thread(() -> {
            synchronized (lock1) {
                System.out.println("线程 1 获取 lock1");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                }
                synchronized (lock2) {
                    System.out.println("线程 1 获取 lock2");
                }
            });
        });

        // 线程 2：先获取 lock2，再获取 lock1
        Thread t2 = new Thread(() -> {
            synchronized (lock2) {
                System.out.println("线程 2 获取 lock2");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                }
                synchronized (lock1) {
                    System.out.println("线程 2 获取 lock1");
                }
            });
        });

        t1.start();
        t2.start();
        // 结果：死锁，两个线程互相等待
    }
}

// ✅ 正确：所有线程按相同顺序获取锁
synchronized (lock1) {
    synchronized (lock2) {
        // ...
    }
}

Lock lock1 = new ReentrantLock();
Lock lock2 = new ReentrantLock();
if (lock1.tryLock(100, TimeUnit.MILLISECONDS)) {
    try {
        if (lock2.tryLock(100, TimeUnit.MILLISECONDS)) {
            try {
                // 业务逻辑
            } finally {
                lock2.unlock();
            }
        }
    } finally {
        lock1.unlock();
    }
}

try {
    lock.lockInterruptibly(); // 可被中断的锁
    // 业务逻辑
} catch (InterruptedException e) {
    // 处理中断
} finally {
    lock.unlock();
}

// 方式 1：饿汉式（线程安全，但可能浪费内存）
public class Singleton1 {
    private static final Singleton1 INSTANCE = new Singleton1();

    private Singleton1() {}

    public static Singleton1 getInstance() {
        return INSTANCE;
    }
}

// 方式 2：双重检查锁（推荐）
public class Singleton2 {
    private static volatile Singleton2 instance;

    private Singleton2() {}

    public static Singleton2 getInstance() {
        if (instance == null) { // 第一次检查
            synchronized (Singleton2.class) {
                if (instance == null) { // 第二次检查
                    instance = new Singleton2();
                }
            }
        }
        return instance;
    }
}

// 方式 3：静态内部类（推荐）
public class Singleton3 {
    private Singleton3() {}

    private static class Holder {
        private static final Singleton3 INSTANCE = new Singleton3();
    }

    public static Singleton3 getInstance() {
        return Holder.INSTANCE;
    }
}

// 方式 4：枚举（最安全）
public enum Singleton4 {
    INSTANCE;

    public void doSomething() {
        System.out.println("单例方法");
    }
}

public class WaitNotifyDemo {
    private static Object lock = new Object();
    private static boolean flag = false;

    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            synchronized (lock) {
                while (!flag) {
                    try {
                        System.out.println("t1 等待");
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println("t1 继续执行");
            }
        });

        Thread t2 = new Thread(() -> {
            synchronized (lock) {
                flag = true;
                System.out.println("t2 通知");
                lock.notify();
            }
        });

        t1.start();
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
        }
        t2.start();
    }
}

import java.util.LinkedList;
import java.util.Queue;

public class ProducerConsumer {
    private Queue<Integer> queue = new LinkedList<>();
    private int maxSize = 10;

    // 生产者
    public synchronized void produce(int value) throws InterruptedException {
        while (queue.size() == maxSize) {
            wait(); // 队列满，等待
        }
        queue.offer(value);
        System.out.println("生产：" + value + ", 队列大小：" + queue.size());
        notifyAll(); // 通知消费者
    }

    // 消费者
    public synchronized int consume() throws InterruptedException {
        while (queue.isEmpty()) {
            wait(); // 队列空，等待
        }
        int value = queue.poll();
        System.out.println("消费：" + value + ", 队列大小：" + queue.size());
        notifyAll(); // 通知生产者
        return value;
    }
}

// 测试
public class Main {
    public static void main(String[] args) {
        ProducerConsumer pc = new ProducerConsumer();

        // 生产者线程
        new Thread(() -> {
            for (int i = 0; i < 20; i++) {
                try {
                    pc.produce(i);
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        // 消费者线程
        new Thread(() -> {
            for (int i = 0; i < 20; i++) {
                try {
                    pc.consume();
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

import java.util.concurrent.*;

public class ThreadPoolDemo {
    public static void main(String[] args) {
        // 1. 固定大小线程池
        ExecutorService fixedPool = Executors.newFixedThreadPool(3);
        // 2. 单线程线程池
        ExecutorService singlePool = Executors.newSingleThreadExecutor();
        // 3. 缓存线程池
        ExecutorService cachedPool = Executors.newCachedThreadPool();
        // 4. 定时任务线程池
        ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(2);

        // 提交任务
        fixedPool.execute(() -> {
            System.out.println("任务执行");
        });

        // 关闭线程池
        fixedPool.shutdown();
    }
}

import java.util.concurrent.*;

public class CustomThreadPool {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2, // 核心线程数
                5, // 最大线程数
                60L, // 空闲线程存活时间
                TimeUnit.SECONDS, // 时间单位
                new LinkedBlockingQueue<>(10), // 任务队列
                Executors.defaultThreadFactory(), // 线程工厂
                new ThreadPoolExecutor.AbortPolicy() // 拒绝策略
        );

        // 提交任务
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.execute(() -> {
                System.out.println("任务" + taskId + "执行，线程：" + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
    }
}

参数	说明
corePoolSize	核心线程数
maximumPoolSize	最大线程数
keepAliveTime	空闲线程存活时间
unit	时间单位
workQueue	任务队列
threadFactory	线程工厂
handler	拒绝策略

// 1. AbortPolicy：抛出异常（默认）
new ThreadPoolExecutor.AbortPolicy()
// 2. CallerRunsPolicy：调用者线程执行
new ThreadPoolExecutor.CallerRunsPolicy()
// 3. DiscardPolicy：丢弃任务
new ThreadPoolExecutor.DiscardPolicy()
// 4. DiscardOldestPolicy：丢弃最老任务
new ThreadPoolExecutor.DiscardOldestPolicy()

import java.util.concurrent.*;
import java.util.ArrayList;
import java.util.List;

public class DownloadManager {
    private ExecutorService executor;

    public DownloadManager(int threadCount) {
        this.executor = Executors.newFixedThreadPool(threadCount);
    }

    public void downloadFiles(List<String> urls) {
        List<Future<Boolean>> futures = new ArrayList<>();
        for (String url : urls) {
            Future<Boolean> future = executor.submit(() -> {
                return downloadFile(url);
            });
            futures.add(future);
        }

        // 等待所有任务完成
        for (int i = 0; i < futures.size(); i++) {
            try {
                boolean success = futures.get(i).get();
                System.out.println("文件" + i + "下载" + (success ? "成功" : "失败"));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        executor.shutdown();
    }

    private boolean downloadFile(String url) {
        System.out.println("下载：" + url + ", 线程：" + Thread.currentThread().getName());
        try {
            Thread.sleep(1000); // 模拟下载
        } catch (InterruptedException e) {
            return false;
        }
        return true;
    }
}

// 测试
public class Main {
    public static void main(String[] args) {
        List<String> urls = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            urls.add("http://example.com/file" + i + ".zip");
        }
        DownloadManager manager = new DownloadManager(3);
        manager.downloadFiles(urls);
    }
}

import java.util.concurrent.*;

public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        int threadCount = 5;
        CountDownLatch latch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++) {
            final int taskId = i;
            new Thread(() -> {
                System.out.println("任务" + taskId + "开始");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务" + taskId + "完成");
                latch.countDown(); // 计数减 1
            }).start();
        }
        System.out.println("等待所有任务完成...");
        latch.await(); // 等待计数归零
        System.out.println("所有任务已完成");
    }
}

import java.util.concurrent.*;

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 taskId = i;
            new Thread(() -> {
                try {
                    System.out.println("线程" + taskId + "准备中...");
                    Thread.sleep((taskId + 1) * 1000);
                    System.out.println("线程" + taskId + "已就绪");
                    barrier.await(); // 等待其他线程
                    System.out.println("线程" + taskId + "继续执行");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

特性	CountDownLatch	CyclicBarrier
可重用	❌ 不可重用	✅ 可重用
等待方式	一个或多个线程等待	所有线程互相等待
计数方式	递减到 0	递增到 N
使用场景	主线程等待子线程	多线程协同工作

import java.util.concurrent.*;

public class SemaphoreDemo {
    public static void main(String[] args) {
        // 只允许 3 个线程同时访问
        Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            new Thread(() -> {
                try {
                    semaphore.acquire(); // 获取许可
                    System.out.println("线程" + taskId + "获得许可，开始执行");
                    Thread.sleep(2000);
                    System.out.println("线程" + taskId + "执行完毕");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release(); // 释放许可
                }
            }).start();
        }
    }
}

import java.util.concurrent.*;
import java.util.*;

public class ConcurrentCollectionsDemo {
    public static void main(String[] args) {
        // 1. ConcurrentHashMap：线程安全的 HashMap
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
        map.put("key1", 1);
        map.putIfAbsent("key2", 2); // 如果不存在则添加

        // 2. CopyOnWriteArrayList：读多写少场景
        CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
        list.add("item1");
        list.add("item2");

        // 3. ConcurrentLinkedQueue：无界非阻塞队列
        ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
        queue.offer("task1");
        String task = queue.poll();

        // 4. BlockingQueue：阻塞队列
        BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(10);
        try {
            blockingQueue.put("item"); // 队列满时阻塞
            String item = blockingQueue.take(); // 队列空时阻塞
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

场景	推荐集合	说明
高并发 Map	`ConcurrentHashMap`	分段锁，性能好
读多写少 List	`CopyOnWriteArrayList`	写时复制
无界队列	`ConcurrentLinkedQueue`	非阻塞
有界队列	`LinkedBlockingQueue`	阻塞
优先级队列	`PriorityBlockingQueue`	按优先级
延迟队列	`DelayQueue`	延迟获取

import java.util.concurrent.*;

public class CompletableFutureDemo {
    public static void main(String[] args) throws Exception {
        // 1. 异步执行任务
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("异步任务执行中...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            return "结果 1";
        });

        // 2. 任务完成后的回调
        future1.thenAccept(result -> {
            System.out.println("任务完成，结果：" + result);
        });

        // 3. 链式调用
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            return 10;
        }).thenApply(num -> {
            return num * 2; // 20
        }).thenApply(num -> {
            return num + 5; // 25
        });
        System.out.println("链式结果：" + future2.get());

        // 4. 组合多个异步任务
        CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
            return "任务 1";
        });
        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            return "任务 2";
        });

        // 等待所有任务完成
        CompletableFuture<Void> allTasks = CompletableFuture.allOf(task1, task2);
        allTasks.get();
        System.out.println("所有任务完成");

        // 等待任意一个任务完成
        CompletableFuture<Object> anyTask = CompletableFuture.anyOf(task1, task2);
        System.out.println("最快完成的任务：" + anyTask.get());

        // 5. 异常处理
        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            if (Math.random() > 0.5) {
                throw new RuntimeException("任务失败");
            }
            return "成功";
        }).exceptionally(ex -> {
            System.out.println("捕获异常：" + ex.getMessage());
            return "默认值";
        });
        System.out.println("结果：" + future3.get());
    }
}

方法	说明	示例
`supplyAsync()`	异步执行有返回值任务	`supplyAsync(() -> "result")`
`runAsync()`	异步执行无返回值任务	`runAsync(() -> {...})`
`thenApply()`	转换结果	`thenApply(x -> x * 2)`
`thenAccept()`	消费结果	`thenAccept(System.out::println)`
`thenCompose()`	组合 Future	`thenCompose(x -> anotherFuture)`
`allOf()`	等待所有完成	`allOf(f1, f2, f3)`
`anyOf()`	等待任意完成	`anyOf(f1, f2, f3)`
`exceptionally()`	异常处理	`exceptionally(ex -> defaultValue)`

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

public class ParallelProcessingDemo {
    public static void main(String[] args) throws Exception {
        // 模拟大量数据
        List<Integer> data = IntStream.range(0, 1000000).boxed().collect(Collectors.toList());

        // 方式 1：使用线程池
        long start1 = System.currentTimeMillis();
        int result1 = processWithThreadPool(data);
        long end1 = System.currentTimeMillis();
        System.out.println("线程池处理耗时：" + (end1 - start1) + "ms, 结果：" + result1);

        // 方式 2：使用并行流
        long start2 = System.currentTimeMillis();
        int result2 = data.parallelStream().filter(n -> n % 2 == 0).mapToInt(n -> n * 2).sum();
        long end2 = System.currentTimeMillis();
        System.out.println("并行流处理耗时：" + (end2 - start2) + "ms, 结果：" + result2);

        // 方式 3：使用 CompletableFuture
        long start3 = System.currentTimeMillis();
        int result3 = processWithCompletableFuture(data);
        long end3 = System.currentTimeMillis();
        System.out.println("CompletableFuture 处理耗时：" + (end3 - start3) + "ms, 结果：" + result3);
    }

    // 使用线程池处理
    private static int processWithThreadPool(List<Integer> data) throws Exception {
        int threadCount = Runtime.getRuntime().availableProcessors();
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        int chunkSize = data.size() / threadCount;
        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 0; i < threadCount; i++) {
            final int start = i * chunkSize;
            final int end = (i == threadCount - 1) ? data.size() : (i + 1) * chunkSize;
            Future<Integer> future = executor.submit(() -> {
                int sum = 0;
                for (int j = start; j < end; j++) {
                    int n = data.get(j);
                    if (n % 2 == 0) {
                        sum += n * 2;
                    }
                }
                return sum;
            });
            futures.add(future);
        }
        int totalSum = 0;
        for (Future<Integer> future : futures) {
            totalSum += future.get();
        }
        executor.shutdown();
        return totalSum;
    }

    // 使用 CompletableFuture 处理
    private static int processWithCompletableFuture(List<Integer> data) throws Exception {
        int threadCount = Runtime.getRuntime().availableProcessors();
        int chunkSize = data.size() / threadCount;
        List<CompletableFuture<Integer>> futures = new ArrayList<>();
        for (int i = 0; i < threadCount; i++) {
            final int start = i * chunkSize;
            final int end = (i == threadCount - 1) ? data.size() : (i + 1) * chunkSize;
            CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
                int sum = 0;
                for (int j = start; j < end; j++) {
                    int n = data.get(j);
                    if (n % 2 == 0) {
                        sum += n * 2;
                    }
                }
                return sum;
            });
            futures.add(future);
        }
        CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        allOf.get();
        return futures.stream().mapToInt(f -> {
            try {
                return f.get();
            } catch (Exception e) {
                return 0;
            }
        }).sum();
    }
}

import java.util.concurrent.*;

public class ThreadPoolBestPractice {
    // ❌ 不推荐：使用 Executors 创建（可能导致 OOM）
    ExecutorService bad1 = Executors.newFixedThreadPool(10); // 无界队列
    ExecutorService bad2 = Executors.newCachedThreadPool(); // 无限线程

    // ✅ 推荐：手动创建，明确参数
    ThreadPoolExecutor good = new ThreadPoolExecutor(
            4, // 核心线程数
            8, // 最大线程数
            60L, TimeUnit.SECONDS, // 空闲线程存活时间
            new LinkedBlockingQueue<>(100), // 有界队列
            new ThreadFactory() { // 自定义线程工厂
                private int count = 0;

                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setName("MyPool-" + count++);
                    thread.setUncaughtExceptionHandler((t, e) -> {
                        System.err.println("线程" + t.getName() + "异常：" + e.getMessage());
                    });
                    return thread;
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
    );
}

任务类型	推荐大小	说明
CPU 密集型	N + 1	N = CPU 核心数
IO 密集型	2N	或 N / (1 - 阻塞系数)
混合型	根据实际测试调整	监控 CPU 和 IO 使用率

// 获取 CPU 核心数
int cpuCount = Runtime.getRuntime().availableProcessors();
// CPU 密集型任务
int cpuIntensivePoolSize = cpuCount + 1;
// IO 密集型任务（假设阻塞系数 0.9）
int ioIntensivePoolSize = cpuCount * 2;
// 或更精确：cpuCount / (1 - 0.9) = cpuCount * 10

import java.util.concurrent.*;

public class ExceptionHandlingDemo {
    public static void main(String[] args) {
        // 方式 1：try-catch 捕获
        Thread t1 = new Thread(() -> {
            try {
                int result = 10 / 0;
            } catch (Exception e) {
                System.err.println("捕获异常：" + e.getMessage());
            }
        });

        // 方式 2：设置未捕获异常处理器
        Thread t2 = new Thread(() -> {
            int result = 10 / 0; // 抛出异常
        });
        t2.setUncaughtExceptionHandler((thread, throwable) -> {
            System.err.println("线程" + thread.getName() + "异常：" + throwable.getMessage());
        });

        // 方式 3：线程池的 afterExecute
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10)) {
            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                super.afterExecute(r, t);
                if (t != null) {
                    System.err.println("任务执行异常：" + t.getMessage());
                }
            }
        };

        // 方式 4：Future.get() 捕获
        ExecutorService pool = Executors.newFixedThreadPool(2);
        Future<Integer> future = pool.submit(() -> {
            return 10 / 0;
        });
        try {
            future.get();
        } catch (ExecutionException e) {
            System.err.println("任务异常：" + e.getCause().getMessage());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        t1.start();
        t2.start();
        executor.shutdown();
        pool.shutdown();
    }
}

public class CommonPitfalls {
    // ❌ 陷阱 1：直接调用 run() 而不是 start()
    public void pitfall1() {
        Thread thread = new Thread(() -> System.out.println("执行"));
        thread.run(); // ❌ 错误：在当前线程执行，不会创建新线程
        thread.start(); // ✅ 正确：创建新线程执行
    }

    // ❌ 陷阱 2：忘记释放锁
    private Object lock = new Object();
    public void pitfall2() {
        synchronized (lock) {
            // 业务逻辑
            if (someCondition()) {
                return; // ❌ 可能导致锁未释放
            }
        }
        // ✅ 使用 try-finally 确保释放
    }

    // ❌ 陷阱 3：在循环中创建线程
    public void pitfall3() {
        for (int i = 0; i < 1000; i++) {
            new Thread(() -> { // 任务
            }).start(); // ❌ 创建大量线程，资源浪费
        }
        // ✅ 使用线程池
        ExecutorService executor = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 1000; i++) {
            executor.execute(() -> { // 任务
            });
        }
        executor.shutdown();
    }

    // ❌ 陷阱 4：共享可变状态
    private int count = 0; // 共享变量
    public void pitfall4() {
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                count++; // ❌ 非线程安全
            }).start();
        }
        // ✅ 使用原子类或同步
        private AtomicInteger safeCount = new AtomicInteger(0);
        safeCount.incrementAndGet();
    }

    // ❌ 陷阱 5：忘记关闭线程池
    public void pitfall5() {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        executor.execute(() -> { // 任务
        });
        // ❌ 忘记关闭，程序不会退出
        // ✅ 正确关闭
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }

    private boolean someCondition() {
        return true;
    }
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class PerformanceOptimization {
    // 1. 使用 ThreadLocal 避免同步
    private static ThreadLocal<SimpleDateFormat> dateFormat = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd"));

    public String formatDate(Date date) {
        return dateFormat.get().format(date); // 每个线程独立实例
    }

    // 2. 使用 LongAdder 替代 AtomicLong（高并发计数）
    private LongAdder counter = new LongAdder(); // 性能更好

    public void increment() {
        counter.increment();
    }

    public long getCount() {
        return counter.sum();
    }

    // 3. 减小锁粒度
    private Object lock1 = new Object();
    private Object lock2 = new Object();

    public void fineLock() {
        synchronized (lock1) {
            // 操作 1
        }
        synchronized (lock2) {
            // 操作 2（独立的锁）
        }
    }

    // 4. 使用读写锁
    private ReadWriteLock rwLock = new ReentrantReadWriteLock();
    private Map<String, String> cache = new HashMap<>();

    public String read(String key) {
        rwLock.readLock().lock();
        try {
            return cache.get(key); // 多个读线程可并发
        } finally {
            rwLock.readLock().unlock();
        }
    }

    public void write(String key, String value) {
        rwLock.writeLock().lock();
        try {
            cache.put(key, value); // 写操作独占
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    // 5. 批量操作减少锁竞争
    private BlockingQueue<String> queue = new LinkedBlockingQueue<>();

    public void batchProcess() {
        List<String> batch = new ArrayList<>();
        queue.drainTo(batch, 100); // 一次取出多个，减少锁竞争
        for (String item : batch) {
            // 处理
        }
    }
}

import java.lang.management.*;
import java.util.concurrent.*;

public class ThreadMonitoring {
    public static void main(String[] args) {
        // 1. 获取线程信息
        ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
        System.out.println("当前线程数：" + threadMXBean.getThreadCount());
        System.out.println("峰值线程数：" + threadMXBean.getPeakThreadCount());
        System.out.println("总启动线程数：" + threadMXBean.getTotalStartedThreadCount());

        // 2. 检测死锁
        long[] deadlockedThreads = threadMXBean.findDeadlockedThreads();
        if (deadlockedThreads != null) {
            System.out.println("检测到死锁，涉及线程数：" + deadlockedThreads.length);
            ThreadInfo[] threadInfos = threadMXBean.getThreadInfo(deadlockedThreads);
            for (ThreadInfo info : threadInfos) {
                System.out.println("死锁线程：" + info.getThreadName());
            }
        }

        // 3. 监控线程池状态
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10));
        System.out.println("活跃线程数：" + executor.getActiveCount());
        System.out.println("核心线程数：" + executor.getCorePoolSize());
        System.out.println("最大线程数：" + executor.getMaximumPoolSize());
        System.out.println("当前线程数：" + executor.getPoolSize());
        System.out.println("队列大小：" + executor.getQueue().size());
        System.out.println("已完成任务数：" + executor.getCompletedTaskCount());
        System.out.println("总任务数：" + executor.getTaskCount());

        // 4. 打印所有线程堆栈
        Map<Thread, StackTraceElement[]> allThreads = Thread.getAllStackTraces();
        for (Map.Entry<Thread, StackTraceElement[]> entry : allThreads.entrySet()) {
            Thread thread = entry.getKey();
            System.out.println("\n线程：" + thread.getName() + ", 状态：" + thread.getState());
            for (StackTraceElement element : entry.getValue()) {
                System.out.println(" " + element);
            }
        }
        executor.shutdown();
    }
}

需要返回值？ ├─ 是 → Callable └─ 否 → 需要继承其他类？ ├─ 是 → Runnable └─ 否 → Thread 或 Runnable

场景	推荐方式	说明
简单同步	`synchronized`	简单易用
需要尝试获取锁	`Lock.tryLock()`	可超时
读多写少	`ReadWriteLock`	读并发
可见性保证	`volatile`	轻量级
原子操作	`AtomicXxx`	无锁
高并发计数	`LongAdder`	性能最好

场景	线程池类型	说明
固定任务数	`FixedThreadPool`	固定线程数
单线程顺序执行	`SingleThreadExecutor`	顺序执行
大量短任务	`CachedThreadPool`	动态扩展
定时任务	`ScheduledThreadPool`	定时调度
自定义需求	`ThreadPoolExecutor`	完全控制

需求	工具类	使用场景
等待多个线程完成	`CountDownLatch`	主线程等待
多线程协同	`CyclicBarrier`	互相等待
限流	`Semaphore`	控制并发数
线程安全 Map	`ConcurrentHashMap`	高并发读写
线程安全 List	`CopyOnWriteArrayList`	读多写少
阻塞队列	`BlockingQueue`	生产消费
异步编程	`CompletableFuture`	链式调用

thread.start() // 启动线程 
thread.join() // 等待线程结束 
thread.interrupt() // 中断线程 
Thread.sleep(millis) // 休眠 
Thread.yield() // 让出 CPU 
thread.setName(name) // 设置名称 
thread.setPriority(priority) // 设置优先级 
thread.isAlive() // 是否存活 
thread.getState() // 获取状态

lock.lock() // 获取锁 
lock.unlock() // 释放锁 
lock.tryLock() // 尝试获取锁 
lock.tryLock(time, unit) // 超时获取锁 
lock.lockInterruptibly() // 可中断获取锁

executor.execute(runnable) // 执行任务（无返回值） 
executor.submit(callable) // 提交任务（有返回值） 
executor.shutdown() // 优雅关闭 
executor.shutdownNow() // 立即关闭 
executor.awaitTermination() // 等待终止

主题	核心内容
线程创建	优先使用 Runnable/Callable，避免继承 Thread
线程安全	使用 synchronized、Lock、原子类保证安全
线程池	手动创建 ThreadPoolExecutor，避免 Executors
并发工具	善用 JUC 工具类，不要重复造轮子
性能优化	ThreadLocal、LongAdder、读写锁、批量操作
异常处理	必须捕获异常，避免线程意外终止
资源管理	及时关闭线程池，使用 try-finally 释放锁

Java 多线程详解

一、线程基础

1.1 进程与线程

1.2 并发与并行

1.3 多线程的挑战

二、创建线程

2.1 继承 Thread 类

2.2 实现 Runnable 接口

2.3 实现 Callable 接口

2.4 Lambda 表达式创建

2.5 三种方式对比

2.6 线程创建实战案例

三、线程生命周期

3.1 线程状态

3.2 线程常用方法

3.3 sleep vs wait

四、线程同步

4.1 线程安全问题

4.2 synchronized 关键字

4.3 Lock 接口

4.4 ReadWriteLock

4.5 volatile 关键字

4.6 原子类

4.7 死锁问题

4.8 线程安全的单例模式

五、线程通信

5.1 wait/notify

5.2 生产者消费者模式

六、线程池

6.1 为什么使用线程池

6.2 创建线程池

6.3 自定义线程池

6.4 线程池参数

6.5 拒绝策略

6.6 实战案例：批量下载

七、并发工具类

7.1 CountDownLatch

7.2 CyclicBarrier

7.3 Semaphore

7.4 并发集合

7.5 CompletableFuture

7.6 实战：并行处理大数据

八、最佳实践

8.1 线程池配置

8.2 异常处理

8.3 避免常见陷阱

8.4 性能优化技巧

8.5 监控和调试

九、快速参考

线程创建选择

同步方式选择

线程池选择

并发工具选择

常用方法速查

最佳实践清单

十、总结

1. 线程基础

2. 创建线程

3. 线程生命周期

4. 线程同步

5. 线程通信

6. 线程池

7. 并发工具类

8. 最佳实践

关键要点

学习路径

推荐资源

下一步

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具