Java并发编程实战指南

在现代软件开发中，并发编程已成为提升应用性能和响应能力的关键技术。Java作为企业级开发的主流语言，提供了丰富的并发编程工具和机制。本文将深入探讨Java并发编程的核心概念、实现技术和最佳实践。

并发编程基础概念

进程与线程

/**
 * 线程基础示例
 */
public class ThreadBasics {
    
    // 继承Thread类
    static class MyThread extends Thread {
        private String threadName;
        
        public MyThread(String name) {
            this.threadName = name;
        }
        
        @Override
        public void run() {
            for (int i = 0; i < 5; i++) {
                System.out.println(threadName + " - 计数: " + i);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    System.out.println(threadName + " 被中断");
                    return;
                }
            }
            System.out.println(threadName + " 执行完成");
        }
    }
    
    // 实现Runnable接口
    static class MyRunnable implements Runnable {
        private String taskName;
        
        public MyRunnable(String name) {
            this.taskName = name;
        }
        
        @Override
        public void run() {
            System.out.println(taskName + " 开始执行，线程: " + 
                Thread.currentThread().getName());
            
            // 模拟耗时操作
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
            
            System.out.println(taskName + " 执行完成");
        }
    }
    
    public static void main(String[] args) {
        // 创建并启动线程
        MyThread thread1 = new MyThread("线程1");
        thread1.start();
        
        // 使用Runnable创建线程
        Thread thread2 = new Thread(new MyRunnable("任务1"), "工作线程");
        thread2.start();
        
        // 使用Lambda表达式
        Thread thread3 = new Thread(() -> {
            System.out.println("Lambda线程执行: " + 
                Thread.currentThread().getName());
        }, "Lambda线程");
        thread3.start();
        
        // 等待所有线程完成
        try {
            thread1.join();
            thread2.join();
            thread3.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        System.out.println("所有线程执行完成");
    }
}

线程状态和生命周期

/**
 * 线程状态监控示例
 */
public class ThreadStateMonitor {
    
    public static void main(String[] args) throws InterruptedException {
        Thread worker = new Thread(() -> {
            try {
                System.out.println("工作线程开始执行");
                
                // RUNNABLE状态
                for (int i = 0; i < 3; i++) {
                    System.out.println("工作中... " + i);
                    Thread.sleep(1000);
                }
                
                // WAITING状态
                synchronized (ThreadStateMonitor.class) {
                    System.out.println("进入等待状态");
                    ThreadStateMonitor.class.wait(2000);
                }
                
                System.out.println("工作线程完成");
                
            } catch (InterruptedException e) {
                System.out.println("工作线程被中断");
                Thread.currentThread().interrupt();
            }
        }, "WorkerThread");
        
        // 监控线程状态
        Thread monitor = new Thread(() -> {
            while (worker.isAlive()) {
                System.out.println("线程状态: " + worker.getState());
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
            }
            System.out.println("最终状态: " + worker.getState());
        }, "MonitorThread");
        
        System.out.println("启动前状态: " + worker.getState());
        
        monitor.start();
        worker.start();
        
        // 等待2秒后唤醒等待的线程
        Thread.sleep(4000);
        synchronized (ThreadStateMonitor.class) {
            ThreadStateMonitor.class.notifyAll();
        }
        
        worker.join();
        monitor.join();
    }
}

同步机制与锁

synchronized关键字

/**
 * synchronized同步机制示例
 */
public class SynchronizedExample {
    
    private int counter = 0;
    private final Object lock = new Object();
    
    // 同步方法
    public synchronized void incrementMethod() {
        counter++;
        System.out.println("方法同步 - 计数器: " + counter + 
            ", 线程: " + Thread.currentThread().getName());
    }
    
    // 同步代码块
    public void incrementBlock() {
        synchronized (this) {
            counter++;
            System.out.println("代码块同步 - 计数器: " + counter + 
                ", 线程: " + Thread.currentThread().getName());
        }
    }
    
    // 使用自定义锁对象
    public void incrementWithCustomLock() {
        synchronized (lock) {
            counter++;
            System.out.println("自定义锁 - 计数器: " + counter + 
                ", 线程: " + Thread.currentThread().getName());
        }
    }
    
    // 静态同步方法
    public static synchronized void staticSyncMethod() {
        System.out.println("静态同步方法 - 线程: " + 
            Thread.currentThread().getName());
    }
    
    public static void main(String[] args) throws InterruptedException {
        SynchronizedExample example = new SynchronizedExample();
        
        // 创建多个线程测试同步
        Thread[] threads = new Thread[5];
        
        for (int i = 0; i < threads.length; i++) {
            final int threadNum = i;
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 3; j++) {
                    switch (j % 3) {
                        case 0:
                            example.incrementMethod();
                            break;
                        case 1:
                            example.incrementBlock();
                            break;
                        case 2:
                            example.incrementWithCustomLock();
                            break;
                    }
                    
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return;
                    }
                }
            }, "Thread-" + threadNum);
        }
        
        // 启动所有线程
        for (Thread thread : threads) {
            thread.start();
        }
        
        // 等待所有线程完成
        for (Thread thread : threads) {
            thread.join();
        }
        
        System.out.println("最终计数器值: " + example.counter);
    }
}

ReentrantLock显式锁

import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.TimeUnit;

/**
 * ReentrantLock显式锁示例
 */
public class ReentrantLockExample {
    
    private final ReentrantLock lock = new ReentrantLock(true); // 公平锁
    private final Condition condition = lock.newCondition();
    private int sharedResource = 0;
    private boolean ready = false;
    
    // 基本锁使用
    public void basicLockUsage() {
        lock.lock();
        try {
            sharedResource++;
            System.out.println("基本锁操作 - 资源值: " + sharedResource + 
                ", 线程: " + Thread.currentThread().getName());
        } finally {
            lock.unlock();
        }
    }
    
    // 尝试获取锁
    public void tryLockUsage() {
        if (lock.tryLock()) {
            try {
                sharedResource += 2;
                System.out.println("尝试锁成功 - 资源值: " + sharedResource + 
                    ", 线程: " + Thread.currentThread().getName());
                Thread.sleep(1000); // 模拟耗时操作
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                lock.unlock();
            }
        } else {
            System.out.println("尝试锁失败 - 线程: " + 
                Thread.currentThread().getName());
        }
    }
    
    // 带超时的锁获取
    public void tryLockWithTimeout() {
        try {
            if (lock.tryLock(2, TimeUnit.SECONDS)) {
                try {
                    sharedResource += 3;
                    System.out.println("超时锁成功 - 资源值: " + sharedResource + 
                        ", 线程: " + Thread.currentThread().getName());
                } finally {
                    lock.unlock();
                }
            } else {
                System.out.println("超时锁失败 - 线程: " + 
                    Thread.currentThread().getName());
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    // 条件变量使用
    public void producer() {
        lock.lock();
        try {
            while (ready) {
                condition.await(); // 等待消费者消费
            }
            
            sharedResource = (int) (Math.random() * 100);
            ready = true;
            System.out.println("生产者生产: " + sharedResource + 
                ", 线程: " + Thread.currentThread().getName());
            
            condition.signalAll(); // 通知消费者
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }
    }
    
    public void consumer() {
        lock.lock();
        try {
            while (!ready) {
                condition.await(); // 等待生产者生产
            }
            
            System.out.println("消费者消费: " + sharedResource + 
                ", 线程: " + Thread.currentThread().getName());
            ready = false;
            
            condition.signalAll(); // 通知生产者
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }
    }
    
    // 锁信息查询
    public void printLockInfo() {
        System.out.println("锁信息:");
        System.out.println("  是否被锁定: " + lock.isLocked());
        System.out.println("  是否公平锁: " + lock.isFair());
        System.out.println("  等待队列长度: " + lock.getQueueLength());
        System.out.println("  持有锁次数: " + lock.getHoldCount());
    }
    
    public static void main(String[] args) throws InterruptedException {
        ReentrantLockExample example = new ReentrantLockExample();
        
        // 测试基本锁操作
        System.out.println("=== 基本锁操作测试 ===");
        Thread[] basicThreads = new Thread[3];
        for (int i = 0; i < basicThreads.length; i++) {
            basicThreads[i] = new Thread(example::basicLockUsage, "Basic-" + i);
            basicThreads[i].start();
        }
        
        for (Thread thread : basicThreads) {
            thread.join();
        }
        
        // 测试尝试锁
        System.out.println("\n=== 尝试锁测试 ===");
        Thread[] tryThreads = new Thread[3];
        for (int i = 0; i < tryThreads.length; i++) {
            tryThreads[i] = new Thread(example::tryLockUsage, "Try-" + i);
            tryThreads[i].start();
        }
        
        for (Thread thread : tryThreads) {
            thread.join();
        }
        
        // 测试生产者消费者
        System.out.println("\n=== 生产者消费者测试 ===");
        Thread producer = new Thread(example::producer, "Producer");
        Thread consumer = new Thread(example::consumer, "Consumer");
        
        producer.start();
        consumer.start();
        
        producer.join();
        consumer.join();
        
        example.printLockInfo();
    }
}

线程池与执行器框架

ThreadPoolExecutor详解

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

/**
 * 线程池详细示例
 */
public class ThreadPoolExample {
    
    // 自定义线程工厂
    static class CustomThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;
        
        public CustomThreadFactory(String namePrefix) {
            this.namePrefix = namePrefix;
        }
        
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, namePrefix + "-" + 
                threadNumber.getAndIncrement());
            thread.setDaemon(false);
            thread.setPriority(Thread.NORM_PRIORITY);
            return thread;
        }
    }
    
    // 自定义拒绝策略
    static class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            System.out.println("任务被拒绝: " + r.toString() + 
                ", 线程池状态: " + executor.toString());
            
            // 可以选择记录日志、放入队列等处理方式
            try {
                // 尝试重新提交到队列
                if (!executor.getQueue().offer(r, 1, TimeUnit.SECONDS)) {
                    System.out.println("重新提交失败，任务丢弃");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
    // 模拟任务
    static class WorkTask implements Runnable {
        private final int taskId;
        private final int duration;
        
        public WorkTask(int taskId, int duration) {
            this.taskId = taskId;
            this.duration = duration;
        }
        
        @Override
        public void run() {
            System.out.println("任务 " + taskId + " 开始执行，线程: " + 
                Thread.currentThread().getName());
            
            try {
                Thread.sleep(duration);
            } catch (InterruptedException e) {
                System.out.println("任务 " + taskId + " 被中断");
                Thread.currentThread().interrupt();
                return;
            }
            
            System.out.println("任务 " + taskId + " 执行完成，耗时: " + duration + "ms");
        }
        
        @Override
        public String toString() {
            return "WorkTask{taskId=" + taskId + ", duration=" + duration + "}";
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        // 创建自定义线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2,                              // 核心线程数
            4,                              // 最大线程数
            60L,                            // 空闲线程存活时间
            TimeUnit.SECONDS,               // 时间单位
            new ArrayBlockingQueue<>(3),    // 工作队列
            new CustomThreadFactory("Worker"), // 线程工厂
            new CustomRejectedExecutionHandler() // 拒绝策略
        );
        
        // 监控线程池状态
        ScheduledExecutorService monitor = Executors.newScheduledThreadPool(1);
        monitor.scheduleAtFixedRate(() -> {
            System.out.println("线程池状态 - " +
                "活跃线程: " + executor.getActiveCount() + 
                ", 池大小: " + executor.getPoolSize() + 
                ", 队列大小: " + executor.getQueue().size() + 
                ", 完成任务: " + executor.getCompletedTaskCount());
        }, 0, 1, TimeUnit.SECONDS);
        
        // 提交任务
        System.out.println("=== 提交任务 ===");
        for (int i = 1; i <= 10; i++) {
            WorkTask task = new WorkTask(i, 2000 + (int)(Math.random() * 1000));
            try {
                executor.submit(task);
                System.out.println("提交任务 " + i);
            } catch (RejectedExecutionException e) {
                System.out.println("任务 " + i + " 提交失败: " + e.getMessage());
            }
            
            Thread.sleep(200); // 间隔提交
        }
        
        // 等待一段时间观察执行情况
        Thread.sleep(8000);
        
        // 关闭线程池
        System.out.println("\n=== 关闭线程池 ===");
        executor.shutdown();
        
        try {
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                System.out.println("强制关闭线程池");
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        monitor.shutdown();
        System.out.println("程序结束");
    }
}

常用线程池类型

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

/**
 * 常用线程池类型示例
 */
public class ExecutorServiceTypes {
    
    // 测试任务
    static class TestTask implements Callable<String> {
        private final String taskName;
        private final int sleepTime;
        
        public TestTask(String taskName, int sleepTime) {
            this.taskName = taskName;
            this.sleepTime = sleepTime;
        }
        
        @Override
        public String call() throws Exception {
            System.out.println(taskName + " 开始执行，线程: " + 
                Thread.currentThread().getName());
            
            Thread.sleep(sleepTime);
            
            String result = taskName + " 完成，耗时: " + sleepTime + "ms";
            System.out.println(result);
            return result;
        }
    }
    
    // 测试FixedThreadPool
    public static void testFixedThreadPool() throws InterruptedException {
        System.out.println("=== FixedThreadPool 测试 ===");
        
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        List<Future<String>> futures = new ArrayList<>();
        
        // 提交任务
        for (int i = 1; i <= 6; i++) {
            Future<String> future = executor.submit(
                new TestTask("Fixed-Task-" + i, 1000));
            futures.add(future);
        }
        
        // 获取结果
        for (Future<String> future : futures) {
            try {
                String result = future.get();
                System.out.println("结果: " + result);
            } catch (ExecutionException e) {
                System.out.println("任务执行异常: " + e.getCause());
            }
        }
        
        executor.shutdown();
        executor.awaitTermination(5, TimeUnit.SECONDS);
    }
    
    // 测试CachedThreadPool
    public static void testCachedThreadPool() throws InterruptedException {
        System.out.println("\n=== CachedThreadPool 测试 ===");
        
        ExecutorService executor = Executors.newCachedThreadPool();
        
        // 快速提交多个任务
        for (int i = 1; i <= 5; i++) {
            executor.submit(new TestTask("Cached-Task-" + i, 500));
        }
        
        Thread.sleep(2000);
        
        // 再次提交任务，观察线程复用
        for (int i = 6; i <= 8; i++) {
            executor.submit(new TestTask("Cached-Task-" + i, 500));
        }
        
        executor.shutdown();
        executor.awaitTermination(5, TimeUnit.SECONDS);
    }
    
    // 测试SingleThreadExecutor
    public static void testSingleThreadExecutor() throws InterruptedException {
        System.out.println("\n=== SingleThreadExecutor 测试 ===");
        
        ExecutorService executor = Executors.newSingleThreadExecutor();
        
        // 提交多个任务，观察顺序执行
        for (int i = 1; i <= 4; i++) {
            executor.submit(new TestTask("Single-Task-" + i, 800));
        }
        
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);
    }
    
    // 测试ScheduledThreadPool
    public static void testScheduledThreadPool() throws InterruptedException {
        System.out.println("\n=== ScheduledThreadPool 测试 ===");
        
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
        
        // 延迟执行
        System.out.println("提交延迟任务，当前时间: " + System.currentTimeMillis());
        executor.schedule(() -> {
            System.out.println("延迟任务执行，时间: " + System.currentTimeMillis());
        }, 2, TimeUnit.SECONDS);
        
        // 周期性执行
        ScheduledFuture<?> periodicTask = executor.scheduleAtFixedRate(() -> {
            System.out.println("周期任务执行，时间: " + System.currentTimeMillis() + 
                ", 线程: " + Thread.currentThread().getName());
        }, 1, 1, TimeUnit.SECONDS);
        
        // 运行5秒后取消周期任务
        Thread.sleep(5000);
        periodicTask.cancel(false);
        
        executor.shutdown();
        executor.awaitTermination(2, TimeUnit.SECONDS);
    }
    
    // 测试WorkStealingPool (Java 8+)
    public static void testWorkStealingPool() throws InterruptedException {
        System.out.println("\n=== WorkStealingPool 测试 ===");
        
        ExecutorService executor = Executors.newWorkStealingPool();
        
        List<Future<String>> futures = new ArrayList<>();
        
        // 提交不同耗时的任务
        int[] durations = {3000, 1000, 2000, 500, 1500, 800, 2500, 600};
        for (int i = 0; i < durations.length; i++) {
            Future<String> future = executor.submit(
                new TestTask("Stealing-Task-" + (i + 1), durations[i]));
            futures.add(future);
        }
        
        // 获取结果
        for (Future<String> future : futures) {
            try {
                String result = future.get();
                System.out.println("结果: " + result);
            } catch (ExecutionException e) {
                System.out.println("任务执行异常: " + e.getCause());
            }
        }
        
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);
    }
    
    public static void main(String[] args) throws InterruptedException {
        testFixedThreadPool();
        testCachedThreadPool();
        testSingleThreadExecutor();
        testScheduledThreadPool();
        testWorkStealingPool();
        
        System.out.println("\n所有测试完成");
    }
}

并发工具类

CountDownLatch和CyclicBarrier

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

/**
 * CountDownLatch和CyclicBarrier示例
 */
public class ConcurrencyUtilities {
    
    // CountDownLatch示例：等待所有工作线程完成
    public static void countDownLatchExample() throws InterruptedException {
        System.out.println("=== CountDownLatch 示例 ===");
        
        int workerCount = 5;
        CountDownLatch startSignal = new CountDownLatch(1);  // 开始信号
        CountDownLatch doneSignal = new CountDownLatch(workerCount);  // 完成信号
        
        // 创建工作线程
        for (int i = 1; i <= workerCount; i++) {
            new Thread(new Worker(i, startSignal, doneSignal)).start();
        }
        
        System.out.println("主线程准备工作...");
        Thread.sleep(1000);
        
        System.out.println("发出开始信号");
        startSignal.countDown();  // 释放所有等待的工作线程
        
        System.out.println("等待所有工作线程完成...");
        doneSignal.await();  // 等待所有工作线程完成
        
        System.out.println("所有工作完成，主线程继续执行");
    }
    
    static class Worker implements Runnable {
        private final int workerId;
        private final CountDownLatch startSignal;
        private final CountDownLatch doneSignal;
        
        public Worker(int workerId, CountDownLatch startSignal, CountDownLatch doneSignal) {
            this.workerId = workerId;
            this.startSignal = startSignal;
            this.doneSignal = doneSignal;
        }
        
        @Override
        public void run() {
            try {
                System.out.println("工作者 " + workerId + " 等待开始信号");
                startSignal.await();  // 等待开始信号
                
                // 执行工作
                int workTime = 1000 + new Random().nextInt(2000);
                System.out.println("工作者 " + workerId + " 开始工作，预计耗时: " + workTime + "ms");
                Thread.sleep(workTime);
                
                System.out.println("工作者 " + workerId + " 完成工作");
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                doneSignal.countDown();  // 通知完成
            }
        }
    }
    
    // CyclicBarrier示例：多阶段任务同步
    public static void cyclicBarrierExample() throws InterruptedException {
        System.out.println("\n=== CyclicBarrier 示例 ===");
        
        int participantCount = 4;
        
        // 创建屏障，当所有参与者到达时执行汇总任务
        CyclicBarrier barrier = new CyclicBarrier(participantCount, () -> {
            System.out.println("*** 所有参与者到达屏障，执行汇总任务 ***");
        });
        
        // 创建参与者线程
        for (int i = 1; i <= participantCount; i++) {
            new Thread(new Participant(i, barrier)).start();
        }
        
        // 等待所有参与者完成
        Thread.sleep(8000);
    }
    
    static class Participant implements Runnable {
        private final int participantId;
        private final CyclicBarrier barrier;
        
        public Participant(int participantId, CyclicBarrier barrier) {
            this.participantId = participantId;
            this.barrier = barrier;
        }
        
        @Override
        public void run() {
            try {
                // 第一阶段工作
                System.out.println("参与者 " + participantId + " 开始第一阶段工作");
                Thread.sleep(1000 + new Random().nextInt(1000));
                System.out.println("参与者 " + participantId + " 完成第一阶段，等待其他参与者");
                
                barrier.await();  // 等待所有参与者完成第一阶段
                
                // 第二阶段工作
                System.out.println("参与者 " + participantId + " 开始第二阶段工作");
                Thread.sleep(800 + new Random().nextInt(800));
                System.out.println("参与者 " + participantId + " 完成第二阶段，等待其他参与者");
                
                barrier.await();  // 等待所有参与者完成第二阶段
                
                System.out.println("参与者 " + participantId + " 所有工作完成");
                
            } catch (InterruptedException | BrokenBarrierException e) {
                System.out.println("参与者 " + participantId + " 被中断: " + e.getMessage());
                Thread.currentThread().interrupt();
            }
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        countDownLatchExample();
        cyclicBarrierExample();
    }
}

Semaphore信号量

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.Random;

/**
 * Semaphore信号量示例
 */
public class SemaphoreExample {
    
    // 资源池示例
    static class ResourcePool {
        private final Semaphore semaphore;
        private final String[] resources;
        private final boolean[] used;
        
        public ResourcePool(int size) {
            this.semaphore = new Semaphore(size, true); // 公平信号量
            this.resources = new String[size];
            this.used = new boolean[size];
            
            // 初始化资源
            for (int i = 0; i < size; i++) {
                resources[i] = "Resource-" + (i + 1);
            }
        }
        
        public String acquireResource() throws InterruptedException {
            semaphore.acquire(); // 获取许可
            return getResource();
        }
        
        public String tryAcquireResource(long timeout, TimeUnit unit) 
                throws InterruptedException {
            if (semaphore.tryAcquire(timeout, unit)) {
                return getResource();
            }
            return null;
        }
        
        private synchronized String getResource() {
            for (int i = 0; i < resources.length; i++) {
                if (!used[i]) {
                    used[i] = true;
                    System.out.println("获取资源: " + resources[i] + 
                        ", 剩余许可: " + semaphore.availablePermits());
                    return resources[i];
                }
            }
            return null; // 不应该到达这里
        }
        
        public void releaseResource(String resource) {
            if (returnResource(resource)) {
                semaphore.release(); // 释放许可
            }
        }
        
        private synchronized boolean returnResource(String resource) {
            for (int i = 0; i < resources.length; i++) {
                if (resources[i].equals(resource) && used[i]) {
                    used[i] = false;
                    System.out.println("释放资源: " + resource + 
                        ", 可用许可: " + semaphore.availablePermits());
                    return true;
                }
            }
            return false;
        }
        
        public void printStatus() {
            System.out.println("资源池状态 - 可用许可: " + semaphore.availablePermits() + 
                ", 等待队列长度: " + semaphore.getQueueLength());
        }
    }
    
    // 资源使用者
    static class ResourceUser implements Runnable {
        private final int userId;
        private final ResourcePool pool;
        
        public ResourceUser(int userId, ResourcePool pool) {
            this.userId = userId;
            this.pool = pool;
        }
        
        @Override
        public void run() {
            try {
                System.out.println("用户 " + userId + " 尝试获取资源");
                
                // 尝试获取资源
                String resource = pool.tryAcquireResource(2, TimeUnit.SECONDS);
                
                if (resource != null) {
                    System.out.println("用户 " + userId + " 获取到资源: " + resource);
                    
                    // 使用资源
                    int useTime = 1000 + new Random().nextInt(2000);
                    System.out.println("用户 " + userId + " 使用资源 " + resource + 
                        ", 预计使用时间: " + useTime + "ms");
                    Thread.sleep(useTime);
                    
                    // 释放资源
                    pool.releaseResource(resource);
                    System.out.println("用户 " + userId + " 释放资源: " + resource);
                    
                } else {
                    System.out.println("用户 " + userId + " 获取资源超时");
                }
                
            } catch (InterruptedException e) {
                System.out.println("用户 " + userId + " 被中断");
                Thread.currentThread().interrupt();
            }
        }
    }
    
    // 限流示例
    static class RateLimiter {
        private final Semaphore semaphore;
        
        public RateLimiter(int permitsPerSecond) {
            this.semaphore = new Semaphore(permitsPerSecond);
            
            // 启动许可补充线程
            Thread replenisher = new Thread(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        Thread.sleep(1000); // 每秒补充
                        int currentPermits = semaphore.availablePermits();
                        int toAdd = permitsPerSecond - currentPermits;
                        if (toAdd > 0) {
                            semaphore.release(toAdd);
                            System.out.println("补充许可: " + toAdd + 
                                ", 当前可用: " + semaphore.availablePermits());
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            });
            replenisher.setDaemon(true);
            replenisher.start();
        }
        
        public boolean tryAcquire() {
            return semaphore.tryAcquire();
        }
        
        public void acquire() throws InterruptedException {
            semaphore.acquire();
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        // 测试资源池
        System.out.println("=== 资源池测试 ===");
        ResourcePool pool = new ResourcePool(3);
        
        // 创建多个用户
        Thread[] users = new Thread[6];
        for (int i = 1; i <= users.length; i++) {
            users[i - 1] = new Thread(new ResourceUser(i, pool), "User-" + i);
        }
        
        // 启动所有用户
        for (Thread user : users) {
            user.start();
            Thread.sleep(200); // 间隔启动
        }
        
        // 定期打印资源池状态
        for (int i = 0; i < 10; i++) {
            Thread.sleep(1000);
            pool.printStatus();
        }
        
        // 等待所有用户完成
        for (Thread user : users) {
            user.join();
        }
        
        // 测试限流器
        System.out.println("\n=== 限流器测试 ===");
        RateLimiter limiter = new RateLimiter(3); // 每秒3个请求
        
        // 模拟请求
        for (int i = 1; i <= 10; i++) {
            final int requestId = i;
            new Thread(() -> {
                try {
                    if (limiter.tryAcquire()) {
                        System.out.println("请求 " + requestId + " 通过限流，时间: " + 
                            System.currentTimeMillis());
                    } else {
                        System.out.println("请求 " + requestId + " 被限流");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
            
            Thread.sleep(300); // 每300ms发送一个请求
        }
        
        Thread.sleep(5000);
        System.out.println("测试完成");
    }
}

原子类与CAS操作

原子变量使用

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

/**
 * 原子类使用示例
 */
public class AtomicExample {
    
    // 基本原子类型
    private static final AtomicInteger counter = new AtomicInteger(0);
    private static final AtomicLong longCounter = new AtomicLong(0);
    private static final AtomicBoolean flag = new AtomicBoolean(false);
    
    // 原子引用
    private static final AtomicReference<String> atomicString = 
        new AtomicReference<>("initial");
    
    // 原子数组
    private static final AtomicIntegerArray atomicArray = 
        new AtomicIntegerArray(10);
    
    // 原子字段更新器
    private static final AtomicIntegerFieldUpdater<AtomicExample> scoreUpdater =
        AtomicIntegerFieldUpdater.newUpdater(AtomicExample.class, "score");
    
    private volatile int score = 0;
    
    // 基本原子操作示例
    public static void basicAtomicOperations() throws InterruptedException {
        System.out.println("=== 基本原子操作 ===");
        
        int threadCount = 5;
        int incrementsPerThread = 1000;
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        // 创建多个线程进行并发操作
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                for (int j = 0; j < incrementsPerThread; j++) {
                    // 原子递增
                    counter.incrementAndGet();
                    longCounter.addAndGet(2);
                    
                    // CAS操作
                    String current = atomicString.get();
                    String newValue = current + ".";
                    while (!atomicString.compareAndSet(current, newValue)) {
                        current = atomicString.get();
                        newValue = current + ".";
                    }
                }
                latch.countDown();
            }, "Worker-" + i).start();
        }
        
        latch.await();
        
        System.out.println("Counter 最终值: " + counter.get());
        System.out.println("Long Counter 最终值: " + longCounter.get());
        System.out.println("String 长度: " + atomicString.get().length());
        System.out.println("期望 Counter 值: " + (threadCount * incrementsPerThread));
    }
    
    // 原子数组操作
    public static void atomicArrayOperations() throws InterruptedException {
        System.out.println("\n=== 原子数组操作 ===");
        
        int threadCount = 3;
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        // 初始化数组
        for (int i = 0; i < atomicArray.length(); i++) {
            atomicArray.set(i, i * 10);
        }
        
        System.out.println("初始数组: " + java.util.Arrays.toString(
            java.util.stream.IntStream.range(0, atomicArray.length())
                .map(atomicArray::get).toArray()));
        
        // 并发修改数组
        for (int i = 0; i < threadCount; i++) {
            final int threadId = i;
            new Thread(() -> {
                for (int j = 0; j < atomicArray.length(); j++) {
                    // 原子递增
                    atomicArray.incrementAndGet(j);
                    
                    // CAS操作
                    int expected = atomicArray.get(j);
                    while (!atomicArray.compareAndSet(j, expected, expected + threadId)) {
                        expected = atomicArray.get(j);
                    }
                }
                latch.countDown();
            }, "ArrayWorker-" + i).start();
        }
        
        latch.await();
        
        System.out.println("最终数组: " + java.util.Arrays.toString(
            java.util.stream.IntStream.range(0, atomicArray.length())
                .map(atomicArray::get).toArray()));
    }
    
    // 字段更新器示例
    public void fieldUpdaterExample() throws InterruptedException {
        System.out.println("\n=== 字段更新器示例 ===");
        
        int threadCount = 3;
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                for (int j = 0; j < 100; j++) {
                    scoreUpdater.incrementAndGet(this);
                }
                latch.countDown();
            }, "ScoreWorker-" + i).start();
        }
        
        latch.await();
        
        System.out.println("最终分数: " + score);
        System.out.println("期望分数: " + (threadCount * 100));
    }
    
    // ABA问题演示
    public static void abaProblemo() throws InterruptedException {
        System.out.println("\n=== ABA问题演示 ===");
        
        AtomicReference<String> atomicRef = new AtomicReference<>("A");
        AtomicStampedReference<String> stampedRef = 
            new AtomicStampedReference<>("A", 0);
        
        // 线程1：A -> B -> A
        Thread thread1 = new Thread(() -> {
            try {
                Thread.sleep(100);
                
                // A -> B
                boolean success1 = atomicRef.compareAndSet("A", "B");
                System.out.println("Thread1 A->B: " + success1);
                
                Thread.sleep(100);
                
                // B -> A
                boolean success2 = atomicRef.compareAndSet("B", "A");
                System.out.println("Thread1 B->A: " + success2);
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "ABA-Thread1");
        
        // 线程2：尝试 A -> C
        Thread thread2 = new Thread(() -> {
            try {
                String initial = atomicRef.get();
                System.out.println("Thread2 读取初始值: " + initial);
                
                Thread.sleep(300); // 等待线程1完成ABA操作
                
                boolean success = atomicRef.compareAndSet("A", "C");
                System.out.println("Thread2 A->C: " + success + 
                    " (发生了ABA问题，但CAS成功了)");
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "ABA-Thread2");
        
        thread1.start();
        thread2.start();
        
        thread1.join();
        thread2.join();
        
        System.out.println("最终值: " + atomicRef.get());
        
        // 使用AtomicStampedReference解决ABA问题
        System.out.println("\n使用AtomicStampedReference解决ABA问题:");
        
        Thread thread3 = new Thread(() -> {
            try {
                int[] stampHolder = new int[1];
                String ref = stampedRef.get(stampHolder);
                int stamp = stampHolder[0];
                
                System.out.println("Thread3 读取: " + ref + ", 版本: " + stamp);
                
                Thread.sleep(300);
                
                boolean success = stampedRef.compareAndSet("A", "C", stamp, stamp + 1);
                System.out.println("Thread3 A->C: " + success + 
                    " (版本号不匹配，CAS失败)");
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "Stamped-Thread");
        
        // 模拟版本号变化
        stampedRef.compareAndSet("A", "B", 0, 1);
        stampedRef.compareAndSet("B", "A", 1, 2);
        
        thread3.start();
        thread3.join();
    }
    
    public static void main(String[] args) throws InterruptedException {
        basicAtomicOperations();
        atomicArrayOperations();
        
        AtomicExample example = new AtomicExample();
        example.fieldUpdaterExample();
        
        abaProblemo();
    }
}

并发集合

ConcurrentHashMap详解

import java.util.concurrent.*;
import java.util.Map;
import java.util.Random;

/**
 * ConcurrentHashMap使用示例
 */
public class ConcurrentCollectionExample {
    
    // 并发Map操作示例
    public static void concurrentHashMapExample() throws InterruptedException {
        System.out.println("=== ConcurrentHashMap 示例 ===");
        
        ConcurrentHashMap<String, Integer> concurrentMap = new ConcurrentHashMap<>();
        int threadCount = 5;
        int operationsPerThread = 1000;
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        // 并发写入操作
        for (int i = 0; i < threadCount; i++) {
            final int threadId = i;
            new Thread(() -> {
                Random random = new Random();
                
                for (int j = 0; j < operationsPerThread; j++) {
                    String key = "key-" + (random.nextInt(100) + 1);
                    
                    // 原子性增加操作
                    concurrentMap.compute(key, (k, v) -> (v == null) ? 1 : v + 1);
                    
                    // 条件性放入
                    if (j % 10 == 0) {
                        concurrentMap.putIfAbsent("thread-" + threadId + "-" + j, j);
                    }
                    
                    // 原子性替换
                    if (j % 20 == 0) {
                        concurrentMap.replace(key, concurrentMap.get(key), 
                            concurrentMap.getOrDefault(key, 0) * 2);
                    }
                }
                
                latch.countDown();
            }, "MapWorker-" + i).start();
        }
        
        latch.await();
        
        System.out.println("Map 大小: " + concurrentMap.size());
        System.out.println("前10个键值对:");
        concurrentMap.entrySet().stream()
            .limit(10)
            .forEach(entry -> System.out.println("  " + entry.getKey() + 
                " -> " + entry.getValue()));
        
        // 并行流操作
        int sum = concurrentMap.values().parallelStream()
            .mapToInt(Integer::intValue)
            .sum();
        System.out.println("所有值的总和: " + sum);
        
        // 批量操作
        concurrentMap.forEach(10, (key, value) -> {
            if (value > 50) {
                System.out.println("高频键: " + key + " = " + value);
            }
        });
    }
    
    // 并发队列示例
    public static void concurrentQueueExample() throws InterruptedException {
        System.out.println("\n=== 并发队列示例 ===");
        
        // ConcurrentLinkedQueue - 无界非阻塞队列
        ConcurrentLinkedQueue<String> concurrentQueue = new ConcurrentLinkedQueue<>();
        
        // BlockingQueue - 阻塞队列
        BlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<>(10);
        
        // 生产者-消费者模式
        CountDownLatch producerLatch = new CountDownLatch(3);
        CountDownLatch consumerLatch = new CountDownLatch(2);
        
        // 生产者
        for (int i = 1; i <= 3; i++) {
            final int producerId = i;
            new Thread(() -> {
                try {
                    for (int j = 1; j <= 5; j++) {
                        String item = "Producer-" + producerId + "-Item-" + j;
                        concurrentQueue.offer(item);
                        
                        Integer number = producerId * 10 + j;
                        blockingQueue.put(number); // 阻塞式放入
                        
                        System.out.println("生产: " + item + ", 数字: " + number);
                        Thread.sleep(100);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    producerLatch.countDown();
                }
            }, "Producer-" + i).start();
        }
        
        // 消费者
        for (int i = 1; i <= 2; i++) {
            final int consumerId = i;
            new Thread(() -> {
                try {
                    while (!Thread.currentThread().isInterrupted()) {
                        // 非阻塞队列消费
                        String item = concurrentQueue.poll();
                        if (item != null) {
                            System.out.println("消费者 " + consumerId + " 消费: " + item);
                        }
                        
                        // 阻塞队列消费
                        Integer number = blockingQueue.poll(1, TimeUnit.SECONDS);
                        if (number != null) {
                            System.out.println("消费者 " + consumerId + " 消费数字: " + number);
                        }
                        
                        // 检查是否应该停止
                        if (producerLatch.getCount() == 0 && 
                            concurrentQueue.isEmpty() && 
                            blockingQueue.isEmpty()) {
                            break;
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    consumerLatch.countDown();
                }
            }, "Consumer-" + i).start();
        }
        
        producerLatch.await();
        consumerLatch.await();
        
        System.out.println("队列操作完成");
        System.out.println("剩余非阻塞队列元素: " + concurrentQueue.size());
        System.out.println("剩余阻塞队列元素: " + blockingQueue.size());
    }
    
    // CopyOnWriteArrayList示例
    public static void copyOnWriteExample() throws InterruptedException {
        System.out.println("\n=== CopyOnWriteArrayList 示例 ===");
        
        CopyOnWriteArrayList<String> cowList = new CopyOnWriteArrayList<>();
        
        // 初始化数据
        for (int i = 1; i <= 5; i++) {
            cowList.add("Initial-" + i);
        }
        
        CountDownLatch latch = new CountDownLatch(3);
        
        // 写线程
        new Thread(() -> {
            try {
                for (int i = 1; i <= 3; i++) {
                    cowList.add("Writer-" + i);
                    System.out.println("写入: Writer-" + i + ", 当前大小: " + cowList.size());
                    Thread.sleep(500);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                latch.countDown();
            }
        }, "Writer").start();
        
        // 读线程
        for (int i = 1; i <= 2; i++) {
            final int readerId = i;
            new Thread(() -> {
                try {
                    for (int j = 0; j < 5; j++) {
                        System.out.println("读者 " + readerId + " 读取列表:");
                        for (String item : cowList) {
                            System.out.println("  " + item);
                        }
                        System.out.println("读者 " + readerId + " 读取完成，大小: " + 
                            cowList.size());
                        Thread.sleep(300);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    latch.countDown();
                }
            }, "Reader-" + i).start();
        }
        
        latch.await();
        
        System.out.println("最终列表内容:");
        cowList.forEach(item -> System.out.println("  " + item));
    }
    
    public static void main(String[] args) throws InterruptedException {
        concurrentHashMapExample();
        concurrentQueueExample();
        copyOnWriteExample();
    }
}

性能优化与最佳实践

并发性能测试

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 并发性能测试和优化示例
 */
public class ConcurrencyPerformance {
    
    // 不同同步机制的性能对比
    public static void synchronizationPerformanceTest() throws InterruptedException {
        System.out.println("=== 同步机制性能对比 ===");
        
        int threadCount = 10;
        int operationsPerThread = 100000;
        
        // 测试synchronized
        testSynchronized(threadCount, operationsPerThread);
        
        // 测试ReentrantLock
        testReentrantLock(threadCount, operationsPerThread);
        
        // 测试AtomicLong
        testAtomicLong(threadCount, operationsPerThread);
        
        // 测试无同步（不安全）
        testNoSync(threadCount, operationsPerThread);
    }
    
    private static void testSynchronized(int threadCount, int operations) 
            throws InterruptedException {
        
        class SyncCounter {
            private long count = 0;
            
            public synchronized void increment() {
                count++;
            }
            
            public synchronized long getCount() {
                return count;
            }
        }
        
        SyncCounter counter = new SyncCounter();
        long startTime = System.nanoTime();
        
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                for (int j = 0; j < operations; j++) {
                    counter.increment();
                }
                latch.countDown();
            }).start();
        }
        
        latch.await();
        long endTime = System.nanoTime();
        
        System.out.println("Synchronized - 耗时: " + 
            (endTime - startTime) / 1_000_000 + "ms, 结果: " + counter.getCount());
    }
    
    private static void testReentrantLock(int threadCount, int operations) 
            throws InterruptedException {
        
        class LockCounter {
            private final ReentrantLock lock = new ReentrantLock();
            private long count = 0;
            
            public void increment() {
                lock.lock();
                try {
                    count++;
                } finally {
                    lock.unlock();
                }
            }
            
            public long getCount() {
                lock.lock();
                try {
                    return count;
                } finally {
                    lock.unlock();
                }
            }
        }
        
        LockCounter counter = new LockCounter();
        long startTime = System.nanoTime();
        
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                for (int j = 0; j < operations; j++) {
                    counter.increment();
                }
                latch.countDown();
            }).start();
        }
        
        latch.await();
        long endTime = System.nanoTime();
        
        System.out.println("ReentrantLock - 耗时: " + 
            (endTime - startTime) / 1_000_000 + "ms, 结果: " + counter.getCount());
    }
    
    private static void testAtomicLong(int threadCount, int operations) 
            throws InterruptedException {
        
        AtomicLong counter = new AtomicLong(0);
        long startTime = System.nanoTime();
        
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                for (int j = 0; j < operations; j++) {
                    counter.incrementAndGet();
                }
                latch.countDown();
            }).start();
        }
        
        latch.await();
        long endTime = System.nanoTime();
        
        System.out.println("AtomicLong - 耗时: " + 
            (endTime - startTime) / 1_000_000 + "ms, 结果: " + counter.get());
    }
    
    private static void testNoSync(int threadCount, int operations) 
            throws InterruptedException {
        
        class UnsafeCounter {
            private long count = 0;
            
            public void increment() {
                count++; // 非线程安全
            }
            
            public long getCount() {
                return count;
            }
        }
        
        UnsafeCounter counter = new UnsafeCounter();
        long startTime = System.nanoTime();
        
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                for (int j = 0; j < operations; j++) {
                    counter.increment();
                }
                latch.countDown();
            }).start();
        }
        
        latch.await();
        long endTime = System.nanoTime();
        
        System.out.println("无同步(不安全) - 耗时: " + 
            (endTime - startTime) / 1_000_000 + "ms, 结果: " + counter.getCount() + 
            " (期望: " + (threadCount * operations) + ")");
    }
    
    // 线程池大小优化
    public static void threadPoolSizeOptimization() throws InterruptedException {
        System.out.println("\n=== 线程池大小优化 ===");
        
        int[] poolSizes = {1, 2, 4, 8, 16, 32};
        int taskCount = 100;
        
        for (int poolSize : poolSizes) {
            testThreadPoolPerformance(poolSize, taskCount);
        }
    }
    
    private static void testThreadPoolPerformance(int poolSize, int taskCount) 
            throws InterruptedException {
        
        ExecutorService executor = Executors.newFixedThreadPool(poolSize);
        long startTime = System.nanoTime();
        
        CountDownLatch latch = new CountDownLatch(taskCount);
        
        for (int i = 0; i < taskCount; i++) {
            executor.submit(() -> {
                try {
                    // 模拟CPU密集型任务
                    long sum = 0;
                    for (int j = 0; j < 1000000; j++) {
                        sum += j;
                    }
                    
                    // 模拟I/O操作
                    Thread.sleep(10);
                    
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    latch.countDown();
                }
            });
        }
        
        latch.await();
        long endTime = System.nanoTime();
        
        executor.shutdown();
        executor.awaitTermination(5, TimeUnit.SECONDS);
        
        System.out.println("线程池大小: " + poolSize + 
            ", 耗时: " + (endTime - startTime) / 1_000_000 + "ms");
    }
    
    public static void main(String[] args) throws InterruptedException {
        synchronizationPerformanceTest();
        threadPoolSizeOptimization();
    }
}

并发编程最佳实践

1. 线程安全设计原则

• 不可变性优先：优先使用不可变对象，避免共享可变状态
• 最小化共享：减少线程间共享的数据，使用线程本地存储
• 正确使用同步：选择合适的同步机制，避免过度同步
• 避免死锁：统一锁顺序，使用超时机制，避免嵌套锁

2. 性能优化策略

// 使用ThreadLocal避免同步
public class ThreadLocalExample {
    private static final ThreadLocal<SimpleDateFormat> dateFormat = 
        ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
    
    public static String formatDate(Date date) {
        return dateFormat.get().format(date);
    }
}

// 使用分段锁减少竞争
public class SegmentedCounter {
    private final AtomicLong[] segments;
    private final int segmentCount;
    
    public SegmentedCounter(int segmentCount) {
        this.segmentCount = segmentCount;
        this.segments = new AtomicLong[segmentCount];
        for (int i = 0; i < segmentCount; i++) {
            segments[i] = new AtomicLong(0);
        }
    }
    
    public void increment() {
        int segment = Thread.currentThread().hashCode() % segmentCount;
        segments[Math.abs(segment)].incrementAndGet();
    }
    
    public long sum() {
        long total = 0;
        for (AtomicLong segment : segments) {
            total += segment.get();
        }
        return total;
    }
}

3. 异常处理和监控

// 线程异常处理
public class ThreadExceptionHandler implements Thread.UncaughtExceptionHandler {
    @Override
    public void uncaughtException(Thread t, Throwable e) {
        System.err.println("线程 " + t.getName() + " 发生未捕获异常: " + e.getMessage());
        e.printStackTrace();
        
        // 记录日志、发送告警等
        logException(t, e);
        
        // 根据异常类型决定是否重启线程
        if (shouldRestartThread(e)) {
            restartThread(t);
        }
    }
    
    private void logException(Thread t, Throwable e) {
        // 实现日志记录逻辑
    }
    
    private boolean shouldRestartThread(Throwable e) {
        // 判断是否需要重启线程
        return !(e instanceof InterruptedException);
    }
    
    private void restartThread(Thread t) {
        // 重启线程逻辑
    }
}

常见并发问题及解决方案

1. 死锁问题

// 死锁检测和预防
public class DeadlockPrevention {
    private static final Object lock1 = new Object();
    private static final Object lock2 = new Object();
    
    // 统一锁顺序避免死锁
    public static void transferMoney(Account from, Account to, double amount) {
        Account firstLock = from.getId() < to.getId() ? from : to;
        Account secondLock = from.getId() < to.getId() ? to : from;
        
        synchronized (firstLock) {
            synchronized (secondLock) {
                from.withdraw(amount);
                to.deposit(amount);
            }
        }
    }
    
    // 使用超时避免死锁
    public static boolean tryTransferWithTimeout(Account from, Account to, 
            double amount, long timeout, TimeUnit unit) throws InterruptedException {
        
        long deadline = System.nanoTime() + unit.toNanos(timeout);
        
        if (from.getLock().tryLock(timeout, unit)) {
            try {
                long remaining = deadline - System.nanoTime();
                if (to.getLock().tryLock(remaining, TimeUnit.NANOSECONDS)) {
                    try {
                        from.withdraw(amount);
                        to.deposit(amount);
                        return true;
                    } finally {
                        to.getLock().unlock();
                    }
                }
            } finally {
                from.getLock().unlock();
            }
        }
        return false;
    }
}

2. 内存可见性问题

// 正确使用volatile关键字
public class VisibilityExample {
    private volatile boolean running = true;
    private volatile int counter = 0;
    
    public void stop() {
        running = false;  // volatile保证可见性
    }
    
    public void run() {
        while (running) {
            counter++;  // volatile保证可见性
            // 执行业务逻辑
        }
    }
    
    // 使用synchronized保证可见性
    private final Object lock = new Object();
    private boolean flag = false;
    
    public void setFlag(boolean value) {
        synchronized (lock) {
            flag = value;
        }
    }
    
    public boolean getFlag() {
        synchronized (lock) {
            return flag;
        }
    }
}

总结

Java并发编程是一个复杂而重要的主题，需要深入理解其底层原理和实践技巧：

1. 基础概念掌握：理解线程、进程、同步、异步等基本概念
2. 同步机制选择：根据场景选择合适的同步工具（synchronized、Lock、原子类等）
3. 线程池使用：合理配置线程池参数，避免资源浪费和性能问题
4. 并发工具应用：熟练使用CountDownLatch、CyclicBarrier、Semaphore等工具
5. 性能优化：通过性能测试找出瓶颈，采用合适的优化策略
6. 问题排查：掌握常见并发问题的识别和解决方法

在实际开发中，应该遵循”先保证正确性，再优化性能”的原则，编写安全、高效的并发代码。同时要注意监控和日志记录，及时发现和解决并发问题。

通过本文的学习和实践，相信你已经对Java并发编程有了更深入的理解。继续实践和探索，你将能够在复杂的并发场景中游刃有余。
```