拓冰网站定制 
1. 线程操作基础概念线程是操作系统能够进行运算调度的最小单位它被包含在进程之中是进程中的实际运作单位。一个进程可以包含多个线程这些线程共享进程的内存空间和系统资源。线程与进程的主要区别在于进程是资源分配的基本单位线程是CPU调度的基本单位同一进程内的线程共享内存空间进程间的通信需要特殊机制而线程间可以直接读写数据2. 线程的创建与启动在大多数现代编程语言中创建线程通常有以下几种方式2.1 继承Thread类class MyThread extends Thread { public void run() { // 线程执行的代码 } } // 创建并启动线程 MyThread t new MyThread(); t.start();2.2 实现Runnable接口class MyRunnable implements Runnable { public void run() { // 线程执行的代码 } } // 创建并启动线程 Thread t new Thread(new MyRunnable()); t.start();2.3 使用线程池ExecutorService executor Executors.newFixedThreadPool(5); executor.execute(new Runnable() { public void run() { // 线程执行的代码 } });3. 线程的生命周期管理线程在其生命周期中会经历多种状态新建(NEW)线程对象被创建但尚未启动可运行(RUNNABLE)线程正在JVM中执行或等待CPU时间片阻塞(BLOCKED)线程等待获取监视器锁等待(WAITING)线程无限期等待其他线程执行特定操作超时等待(TIMED_WAITING)线程在指定时间内等待终止(TERMINATED)线程已完成执行4. 线程同步与通信4.1 同步方法public synchronized void method() { // 同步代码块 }4.2 同步代码块public void method() { synchronized(this) { // 同步代码块 } }4.3 使用Lock对象Lock lock new ReentrantLock(); public void method() { lock.lock(); try { // 临界区代码 } finally { lock.unlock(); } }4.4 使用条件变量Lock lock new ReentrantLock(); Condition condition lock.newCondition(); public void await() throws InterruptedException { lock.lock(); try { condition.await(); } finally { lock.unlock(); } } public void signal() { lock.lock(); try { condition.signal(); } finally { lock.unlock(); } }5. 线程安全实践5.1 不可变对象public final class ImmutableValue { private final int value; public ImmutableValue(int value) { this.value value; } public int getValue() { return value; } }5.2 线程局部变量ThreadLocalInteger threadLocal new ThreadLocal(); public void method() { threadLocal.set(1); Integer value threadLocal.get(); }5.3 原子变量AtomicInteger counter new AtomicInteger(0); public void increment() { counter.incrementAndGet(); }6. 线程池的高级使用6.1 自定义线程池ThreadPoolExecutor executor new ThreadPoolExecutor( 5, // 核心线程数 10, // 最大线程数 60, // 空闲线程存活时间 TimeUnit.SECONDS, // 时间单位 new ArrayBlockingQueue(100) // 工作队列 );6.2 线程池拒绝策略ThreadPoolExecutor executor new ThreadPoolExecutor( 5, 10, 60, TimeUnit.SECONDS, new ArrayBlockingQueue(100), new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略 );6.3 定时任务线程池ScheduledExecutorService scheduler Executors.newScheduledThreadPool(3); // 延迟执行 scheduler.schedule(() - { // 任务代码 }, 10, TimeUnit.SECONDS); // 周期性执行 scheduler.scheduleAtFixedRate(() - { // 任务代码 }, 0, 1, TimeUnit.SECONDS);7. 并发工具类7.1 CountDownLatchCountDownLatch latch new CountDownLatch(3); // 工作线程 new Thread(() - { // 执行任务 latch.countDown(); }).start(); // 主线程等待 latch.await();7.2 CyclicBarrierCyclicBarrier barrier new CyclicBarrier(3, () - { // 所有线程到达屏障后执行 }); new Thread(() - { // 执行任务 barrier.await(); }).start();7.3 SemaphoreSemaphore semaphore new Semaphore(3); // 允许3个线程同时访问 public void method() throws InterruptedException { semaphore.acquire(); try { // 临界区代码 } finally { semaphore.release(); } }7.4 ExchangerExchangerString exchanger new Exchanger(); new Thread(() - { String data Thread1 Data; try { data exchanger.exchange(data); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); new Thread(() - { String data Thread2 Data; try { data exchanger.exchange(data); } catch (InterruptedException e) { e.printStackTrace(); } }).start();8. 并发集合8.1 ConcurrentHashMapConcurrentMapString, String map new ConcurrentHashMap(); map.put(key, value); String value map.get(key);8.2 CopyOnWriteArrayListListString list new CopyOnWriteArrayList(); list.add(item); String item list.get(0);8.3 BlockingQueueBlockingQueueString queue new LinkedBlockingQueue(); queue.put(item); // 阻塞直到有空间 String item queue.take(); // 阻塞直到有元素9. 线程性能优化9.1 减少锁竞争缩小同步代码块范围使用读写锁替代独占锁使用无锁数据结构9.2 避免死锁按固定顺序获取多个锁使用tryLock()设置超时避免在持有锁时调用外部方法9.3 线程池调优根据任务类型选择合适的工作队列合理设置核心线程数和最大线程数使用合适的拒绝策略10. 线程调试与监控10.1 线程转储分析# 获取Java进程的线程转储 jstack pid thread_dump.txt10.2 使用VisualVM监控线程查看线程状态检测死锁分析线程CPU使用情况10.3 使用JConsole监控线程数量查看线程堆栈检测死锁情况11. 常见线程问题与解决方案11.1 死锁// 错误的锁获取顺序可能导致死锁 public void transfer(Account from, Account to, int amount) { synchronized(from) { synchronized(to) { // 转账操作 } } }解决方案// 使用固定的锁获取顺序 public void transfer(Account from, Account to, int amount) { Account first from.hashCode() to.hashCode() ? from : to; Account second from.hashCode() to.hashCode() ? to : from; synchronized(first) { synchronized(second) { // 转账操作 } } }11.2 活锁// 两个线程不断改变状态导致无法继续执行 while (!tryAcquireLock()) { // 释放资源并重试 releaseSomeResources(); Thread.sleep(100); // 随机延迟可以缓解活锁 }11.3 线程饥饿确保公平的锁获取机制避免高优先级线程独占资源使用公平锁或合理的调度策略12. 现代并发模式12.1 Fork/Join框架class FibonacciTask extends RecursiveTaskInteger { final int n; FibonacciTask(int n) { this.n n; } protected Integer compute() { if (n 1) return n; FibonacciTask f1 new FibonacciTask(n - 1); f1.fork(); FibonacciTask f2 new FibonacciTask(n - 2); return f2.compute() f1.join(); } } ForkJoinPool pool new ForkJoinPool(); int result pool.invoke(new FibonacciTask(10));12.2 CompletableFutureCompletableFuture.supplyAsync(() - { // 异步任务 return result; }).thenApply(result - { // 处理结果 return result.toUpperCase(); }).thenAccept(System.out::println);12.3 响应式编程FluxString flux Flux.just(Hello, World) .map(String::toUpperCase) .filter(s - s.length() 4); flux.subscribe(System.out::println);13. 线程最佳实践优先使用线程池避免频繁创建和销毁线程合理设置线程优先级大多数情况下使用默认优先级避免过度同步只在必要时使用同步机制使用并发工具类优先使用Java并发包中的高级工具注意线程安全确保共享资源的线程安全访问合理处理异常为线程设置未捕获异常处理器避免长时间持有锁减少锁的持有时间考虑使用不可变对象简化线程安全设计合理使用volatile确保变量的可见性测试并发代码使用压力测试验证并发正确性14. 线程调试技巧14.1 线程命名Thread worker new Thread(() - { // 任务代码 }, Worker-1);14.2 线程局部日志ThreadLocalSimpleDateFormat dateFormat ThreadLocal.withInitial( () - new SimpleDateFormat(yyyy-MM-dd HH:mm:ss) ); public void log(String message) { System.out.println(dateFormat.get().format(new Date()) [ Thread.currentThread().getName() ] message); }14.3 使用ThreadMXBean监控ThreadMXBean threadMXBean ManagementFactory.getThreadMXBean(); long[] threadIds threadMXBean.getAllThreadIds(); for (long id : threadIds) { ThreadInfo info threadMXBean.getThreadInfo(id); System.out.println(info.getThreadName() : info.getThreadState()); }15. 线程性能分析15.1 使用JMH进行基准测试BenchmarkMode(Mode.AverageTime) OutputTimeUnit(TimeUnit.MICROSECONDS) public class MyBenchmark { Benchmark public void testMethod() { // 测试代码 } }15.2 使用Async Profiler./profiler.sh -d 30 -f profile.html pid15.3 分析线程争用ThreadMXBean threadMXBean ManagementFactory.getThreadMXBean(); long[] threadIds threadMXBean.findDeadlockedThreads(); if (threadIds ! null) { ThreadInfo[] infos threadMXBean.getThreadInfo(threadIds); for (ThreadInfo info : infos) { System.out.println(info.getThreadName() is waiting on info.getLockName() held by info.getLockOwnerName()); } }16. 线程安全设计模式16.1 单例模式public class Singleton { private static volatile Singleton instance; private Singleton() {} public static Singleton getInstance() { if (instance null) { synchronized (Singleton.class) { if (instance null) { instance new Singleton(); } } } return instance; } }16.2 生产者-消费者模式BlockingQueueItem queue new LinkedBlockingQueue(); // 生产者 new Thread(() - { while (true) { Item item produceItem(); queue.put(item); } }).start(); // 消费者 new Thread(() - { while (true) { Item item queue.take(); consumeItem(item); } }).start();16.3 工作窃取模式ForkJoinPool pool new ForkJoinPool(); class MyTask extends RecursiveAction { protected void compute() { // 任务分解与执行 } } pool.invoke(new MyTask());17. 线程与内存模型17.1 happens-before关系程序顺序规则监视器锁规则volatile变量规则线程启动规则线程终止规则线程中断规则终结器规则传递性17.2 内存屏障// 使用volatile实现内存屏障 volatile boolean flag false; public void writer() { // 写操作 flag true; // 插入StoreStore屏障 } public void reader() { if (flag) { // 插入LoadLoad屏障 // 读操作 } }17.3 final字段安全性class FinalFieldExample { final int x; public FinalFieldExample() { x 42; // 正确初始化final字段 } }18. 线程与异常处理18.1 未捕获异常处理器Thread.setDefaultUncaughtExceptionHandler((t, e) - { System.err.println(Uncaught exception in thread t.getName()); e.printStackTrace(); });18.2 Future异常处理Future? future executor.submit(() - { // 可能抛出异常的任务 }); try { future.get(); } catch (ExecutionException e) { Throwable cause e.getCause(); // 处理任务抛出的异常 }18.3 CompletableFuture异常处理CompletableFuture.supplyAsync(() - { // 可能抛出异常的任务 return result; }).exceptionally(ex - { // 处理异常 return fallback; });19. 线程与I/O操作19.1 异步I/OAsynchronousFileChannel channel AsynchronousFileChannel.open( Paths.get(file.txt), StandardOpenOption.READ); ByteBuffer buffer ByteBuffer.allocate(1024); channel.read(buffer, 0, buffer, new CompletionHandlerInteger, ByteBuffer() { public void completed(Integer result, ByteBuffer attachment) { // 读取完成处理 } public void failed(Throwable exc, ByteBuffer attachment) { // 读取失败处理 } });19.2 NIO与多路复用Selector selector Selector.open(); ServerSocketChannel serverChannel ServerSocketChannel.open(); serverChannel.configureBlocking(false); serverChannel.register(selector, SelectionKey.OP_ACCEPT); while (true) { selector.select(); SetSelectionKey keys selector.selectedKeys(); for (SelectionKey key : keys) { if (key.isAcceptable()) { // 处理连接请求 } else if (key.isReadable()) { // 处理读事件 } } keys.clear(); }19.3 使用Netty处理并发I/OEventLoopGroup group new NioEventLoopGroup(); try { ServerBootstrap b new ServerBootstrap(); b.group(group) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializerSocketChannel() { Override public void initChannel(SocketChannel ch) { ch.pipeline().addLast(new EchoServerHandler()); } }); ChannelFuture f b.bind(8080).sync(); f.channel().closeFuture().sync(); } finally { group.shutdownGracefully(); }20. 线程与数据库交互20.1 连接池配置HikariConfig config new HikariConfig(); config.setJdbcUrl(jdbc:mysql://localhost:3306/db); config.setUsername(user); config.setPassword(password); config.setMaximumPoolSize(20); // 最大连接数 config.setMinimumIdle(5); // 最小空闲连接 HikariDataSource ds new HikariDataSource(config);20.2 事务隔离级别READ_UNCOMMITTEDREAD_COMMITTEDREPEATABLE_READSERIALIZABLE20.3 批量操作优化try (Connection conn ds.getConnection(); PreparedStatement stmt conn.prepareStatement(INSERT INTO table VALUES (?))) { conn.setAutoCommit(false); for (int i 0; i 1000; i) { stmt.setInt(1, i); stmt.addBatch(); if (i % 100 0) { stmt.executeBatch(); conn.commit(); } } stmt.executeBatch(); conn.commit(); }21. 线程与缓存21.1 使用Caffeine缓存CacheString, Data cache Caffeine.newBuilder() .maximumSize(10_000) .expireAfterWrite(10, TimeUnit.MINUTES) .build(); // 自动加载缓存 LoadingCacheString, Data loadingCache Caffeine.newBuilder() .maximumSize(10_000) .build(key - loadDataFromDatabase(key));21.2 缓存一致性策略写穿透(Write-through)写回(Write-behind)失效(Cache-aside)读穿透(Read-through)21.3 分布式缓存CacheManager cacheManager Caching.getCachingProvider() .getCacheManager(); CacheString, String cache cacheManager.createCache(myCache, new MutableConfigurationString, String() .setTypes(String.class, String.class) .setExpiryPolicyFactory(AccessedExpiryPolicy.factoryOf(Duration.ONE_HOUR)) .setStoreByValue(false));22. 线程与微服务22.1 服务调用超时设置HystrixCommand(fallbackMethod fallbackMethod, commandProperties { HystrixProperty(name execution.isolation.thread.timeoutInMilliseconds, value 1000) }) public String callService() { // 调用远程服务 }22.2 熔断器配置HystrixCommand(fallbackMethod fallbackMethod, commandProperties { HystrixProperty(name circuitBreaker.requestVolumeThreshold, value 20), HystrixProperty(name circuitBreaker.sleepWindowInMilliseconds, value 5000), HystrixProperty(name circuitBreaker.errorThresholdPercentage, value 50) }) public String callService() { // 调用远程服务 }22.3 异步服务调用Async public CompletableFutureString asyncCall() { // 异步执行的任务 return CompletableFuture.completedFuture(result); }23. 线程与函数式编程23.1 并行流ListString results dataList.parallelStream() .filter(item - item.startsWith(A)) .map(String::toUpperCase) .collect(Collectors.toList());23.2 CompletableFuture组合CompletableFutureString future1 CompletableFuture.supplyAsync(() - Hello); CompletableFutureString future2 CompletableFuture.supplyAsync(() - World); CompletableFutureString combined future1.thenCombine(future2, (s1, s2) - s1 s2);23.3 反应式编程FluxString flux Flux.fromIterable(dataList) .parallel() .runOn(Schedulers.parallel()) .map(String::toUpperCase) .sequential();24. 线程与测试24.1 并发测试Test public void testConcurrentAccess() throws InterruptedException { final int THREAD_COUNT 10; ExecutorService executor Executors.newFixedThreadPool(THREAD_COUNT); CountDownLatch latch new CountDownLatch(THREAD_COUNT); for (int i 0; i THREAD_COUNT; i) { executor.execute(() - { try { // 测试代码 } finally { latch.countDown(); } }); } latch.await(); executor.shutdown(); }24.2 使用JMeter进行压力测试创建线程组设置并发用户数添加HTTP请求采样器配置断言和监听器分析测试结果24.3 使用TestContainers测试并发数据库访问Testcontainers class DatabaseTest { Container static PostgreSQLContainer? postgres new PostgreSQLContainer(postgres:13); Test void testConcurrentTransactions() { // 并发测试数据库访问 } }25. 线程与安全25.1 线程安全的密码哈希String hashed new BCryptPasswordEncoder().encode(password);25.2 安全随机数生成SecureRandom random SecureRandom.getInstanceStrong(); byte[] bytes new byte[16]; random.nextBytes(bytes);25.3 线程安全的日志记录private static final Logger logger LoggerFactory.getLogger(MyClass.class); public void method() { logger.info(Thread-safe log message); }26. 线程与性能调优26.1 线程上下文切换分析ThreadMXBean threadMXBean ManagementFactory.getThreadMXBean(); long totalContextSwitches 0; for (long id : threadMXBean.getAllThreadIds()) { totalContextSwitches threadMXBean.getThreadInfo(id).getWaitedCount(); }26.2 CPU缓存优化// 伪共享问题解决方案 Contended public class VolatileLong { public volatile long value 0L; }26.3 内存屏障使用// 使用Unsafe实现内存屏障 Unsafe unsafe Unsafe.getUnsafe(); unsafe.storeFence(); // 插入写屏障 unsafe.loadFence(); // 插入读屏障 unsafe.fullFence(); // 插入全屏障27. 线程与容器化27.1 Kubernetes线程池配置apiVersion: apps/v1 kind: Deployment spec: template: spec: containers: - name: app resources: limits: cpu: 2 requests: cpu: 127.2 容器内线程数计算int availableProcessors Runtime.getRuntime().availableProcessors(); int threadPoolSize Math.max(2, availableProcessors * 2);27.3 优雅停机处理Runtime.getRuntime().addShutdownHook(new Thread(() - { executor.shutdown(); try { if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); } }));28. 线程与云原生28.1 无服务器函数并发控制// AWS Lambda并发控制 public class Handler implements RequestHandlerInput, Output { private static final ExecutorService executor Executors.newFixedThreadPool(10); public Output handleRequest(Input input, Context context) { FutureOutput future executor.submit(() - process(input)); return future.get(); } }28.2 消息队列消费者线程池Bean public ConcurrentKafkaListenerContainerFactoryString, String kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactoryString, String factory new ConcurrentKafkaListenerContainerFactory(); factory.setConcurrency(3); // 每个监听器3个线程 return factory; }28.3 分布式锁实现// 使用Redis实现分布式锁 public boolean tryLock(String lockKey, String requestId, int expireTime) { return redisTemplate.opsForValue().setIfAbsent( lockKey, requestId, expireTime, TimeUnit.SECONDS); } public boolean releaseLock(String lockKey, String requestId) { String script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end; Long result redisTemplate.execute( new DefaultRedisScript(script, Long.class), Collections.singletonList(lockKey), requestId); return result ! null result 1; }29. 线程与机器学习29.1 并行模型训练// 使用并行流处理数据 double[] weights dataList.parallelStream() .mapToDouble(this::computeWeight) .toArray();29.2 使用ForkJoinPool处理大数据class TrainingTask extends RecursiveAction { private final double[] data; private final int start; private final int end; protected void compute() { if (end - start THRESHOLD) { // 直接计算 } else { int mid (start end) / 2; invokeAll(new TrainingTask(data, start, mid), new TrainingTask(data, mid, end)); } } }29.3 异步模型预测CompletableFuturePredictionResult future CompletableFuture.supplyAsync(() - { return model.predict(input); }, executor);30. 线程与区块链30.1 挖矿线程池ExecutorService miningPool Executors.newWorkStealingPool(); public Block mineBlock(Blockchain blockchain, ListTransaction transactions) { ListFutureBlock futures new ArrayList(); for (int i 0; i Runtime.getRuntime().availableProcessors(); i) { futures.add(miningPool.submit(() - { return blockchain.mineBlock(transactions); })); } return futures.stream() .map(f - { try { return f.get(); } catch (Exception e) { return null; } }) .filter(Objects::nonNull) .findFirst() .orElseThrow(() - new RuntimeException(Mining failed)); }30.2 共识算法实现// 简单的PBFT实现 public class PBFTNode { private final ExecutorService executor Executors.newCachedThreadPool(); private final ListNode nodes; public void onReceiveMessage(Message message) { executor.execute(() - { switch (message.getType()) { case PRE_PREPARE: // 处理预准备消息 break; case PREPARE: // 处理准备消息 break; case COMMIT: // 处理提交消息 break; } }); } }30.3 交易并行验证public boolean validateTransactions(ListTransaction transactions) { return transactions.parallelStream() .allMatch(this::validateTransaction); } Java线程编程：从基础到高级并发实践
1. 线程操作基础概念线程是操作系统能够进行运算调度的最小单位它被包含在进程之中是进程中的实际运作单位。一个进程可以包含多个线程这些线程共享进程的内存空间和系统资源。线程与进程的主要区别在于进程是资源分配的基本单位线程是CPU调度的基本单位同一进程内的线程共享内存空间进程间的通信需要特殊机制而线程间可以直接读写数据2. 线程的创建与启动在大多数现代编程语言中创建线程通常有以下几种方式2.1 继承Thread类class MyThread extends Thread { public void run() { // 线程执行的代码 } } // 创建并启动线程 MyThread t new MyThread(); t.start();2.2 实现Runnable接口class MyRunnable implements Runnable { public void run() { // 线程执行的代码 } } // 创建并启动线程 Thread t new Thread(new MyRunnable()); t.start();2.3 使用线程池ExecutorService executor Executors.newFixedThreadPool(5); executor.execute(new Runnable() { public void run() { // 线程执行的代码 } });3. 线程的生命周期管理线程在其生命周期中会经历多种状态新建(NEW)线程对象被创建但尚未启动可运行(RUNNABLE)线程正在JVM中执行或等待CPU时间片阻塞(BLOCKED)线程等待获取监视器锁等待(WAITING)线程无限期等待其他线程执行特定操作超时等待(TIMED_WAITING)线程在指定时间内等待终止(TERMINATED)线程已完成执行4. 线程同步与通信4.1 同步方法public synchronized void method() { // 同步代码块 }4.2 同步代码块public void method() { synchronized(this) { // 同步代码块 } }4.3 使用Lock对象Lock lock new ReentrantLock(); public void method() { lock.lock(); try { // 临界区代码 } finally { lock.unlock(); } }4.4 使用条件变量Lock lock new ReentrantLock(); Condition condition lock.newCondition(); public void await() throws InterruptedException { lock.lock(); try { condition.await(); } finally { lock.unlock(); } } public void signal() { lock.lock(); try { condition.signal(); } finally { lock.unlock(); } }5. 线程安全实践5.1 不可变对象public final class ImmutableValue { private final int value; public ImmutableValue(int value) { this.value value; } public int getValue() { return value; } }5.2 线程局部变量ThreadLocalInteger threadLocal new ThreadLocal(); public void method() { threadLocal.set(1); Integer value threadLocal.get(); }5.3 原子变量AtomicInteger counter new AtomicInteger(0); public void increment() { counter.incrementAndGet(); }6. 线程池的高级使用6.1 自定义线程池ThreadPoolExecutor executor new ThreadPoolExecutor( 5, // 核心线程数 10, // 最大线程数 60, // 空闲线程存活时间 TimeUnit.SECONDS, // 时间单位 new ArrayBlockingQueue(100) // 工作队列 );6.2 线程池拒绝策略ThreadPoolExecutor executor new ThreadPoolExecutor( 5, 10, 60, TimeUnit.SECONDS, new ArrayBlockingQueue(100), new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略 );6.3 定时任务线程池ScheduledExecutorService scheduler Executors.newScheduledThreadPool(3); // 延迟执行 scheduler.schedule(() - { // 任务代码 }, 10, TimeUnit.SECONDS); // 周期性执行 scheduler.scheduleAtFixedRate(() - { // 任务代码 }, 0, 1, TimeUnit.SECONDS);7. 并发工具类7.1 CountDownLatchCountDownLatch latch new CountDownLatch(3); // 工作线程 new Thread(() - { // 执行任务 latch.countDown(); }).start(); // 主线程等待 latch.await();7.2 CyclicBarrierCyclicBarrier barrier new CyclicBarrier(3, () - { // 所有线程到达屏障后执行 }); new Thread(() - { // 执行任务 barrier.await(); }).start();7.3 SemaphoreSemaphore semaphore new Semaphore(3); // 允许3个线程同时访问 public void method() throws InterruptedException { semaphore.acquire(); try { // 临界区代码 } finally { semaphore.release(); } }7.4 ExchangerExchangerString exchanger new Exchanger(); new Thread(() - { String data Thread1 Data; try { data exchanger.exchange(data); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); new Thread(() - { String data Thread2 Data; try { data exchanger.exchange(data); } catch (InterruptedException e) { e.printStackTrace(); } }).start();8. 并发集合8.1 ConcurrentHashMapConcurrentMapString, String map new ConcurrentHashMap(); map.put(key, value); String value map.get(key);8.2 CopyOnWriteArrayListListString list new CopyOnWriteArrayList(); list.add(item); String item list.get(0);8.3 BlockingQueueBlockingQueueString queue new LinkedBlockingQueue(); queue.put(item); // 阻塞直到有空间 String item queue.take(); // 阻塞直到有元素9. 线程性能优化9.1 减少锁竞争缩小同步代码块范围使用读写锁替代独占锁使用无锁数据结构9.2 避免死锁按固定顺序获取多个锁使用tryLock()设置超时避免在持有锁时调用外部方法9.3 线程池调优根据任务类型选择合适的工作队列合理设置核心线程数和最大线程数使用合适的拒绝策略10. 线程调试与监控10.1 线程转储分析# 获取Java进程的线程转储 jstack pid thread_dump.txt10.2 使用VisualVM监控线程查看线程状态检测死锁分析线程CPU使用情况10.3 使用JConsole监控线程数量查看线程堆栈检测死锁情况11. 常见线程问题与解决方案11.1 死锁// 错误的锁获取顺序可能导致死锁 public void transfer(Account from, Account to, int amount) { synchronized(from) { synchronized(to) { // 转账操作 } } }解决方案// 使用固定的锁获取顺序 public void transfer(Account from, Account to, int amount) { Account first from.hashCode() to.hashCode() ? from : to; Account second from.hashCode() to.hashCode() ? to : from; synchronized(first) { synchronized(second) { // 转账操作 } } }11.2 活锁// 两个线程不断改变状态导致无法继续执行 while (!tryAcquireLock()) { // 释放资源并重试 releaseSomeResources(); Thread.sleep(100); // 随机延迟可以缓解活锁 }11.3 线程饥饿确保公平的锁获取机制避免高优先级线程独占资源使用公平锁或合理的调度策略12. 现代并发模式12.1 Fork/Join框架class FibonacciTask extends RecursiveTaskInteger { final int n; FibonacciTask(int n) { this.n n; } protected Integer compute() { if (n 1) return n; FibonacciTask f1 new FibonacciTask(n - 1); f1.fork(); FibonacciTask f2 new FibonacciTask(n - 2); return f2.compute() f1.join(); } } ForkJoinPool pool new ForkJoinPool(); int result pool.invoke(new FibonacciTask(10));12.2 CompletableFutureCompletableFuture.supplyAsync(() - { // 异步任务 return result; }).thenApply(result - { // 处理结果 return result.toUpperCase(); }).thenAccept(System.out::println);12.3 响应式编程FluxString flux Flux.just(Hello, World) .map(String::toUpperCase) .filter(s - s.length() 4); flux.subscribe(System.out::println);13. 线程最佳实践优先使用线程池避免频繁创建和销毁线程合理设置线程优先级大多数情况下使用默认优先级避免过度同步只在必要时使用同步机制使用并发工具类优先使用Java并发包中的高级工具注意线程安全确保共享资源的线程安全访问合理处理异常为线程设置未捕获异常处理器避免长时间持有锁减少锁的持有时间考虑使用不可变对象简化线程安全设计合理使用volatile确保变量的可见性测试并发代码使用压力测试验证并发正确性14. 线程调试技巧14.1 线程命名Thread worker new Thread(() - { // 任务代码 }, Worker-1);14.2 线程局部日志ThreadLocalSimpleDateFormat dateFormat ThreadLocal.withInitial( () - new SimpleDateFormat(yyyy-MM-dd HH:mm:ss) ); public void log(String message) { System.out.println(dateFormat.get().format(new Date()) [ Thread.currentThread().getName() ] message); }14.3 使用ThreadMXBean监控ThreadMXBean threadMXBean ManagementFactory.getThreadMXBean(); long[] threadIds threadMXBean.getAllThreadIds(); for (long id : threadIds) { ThreadInfo info threadMXBean.getThreadInfo(id); System.out.println(info.getThreadName() : info.getThreadState()); }15. 线程性能分析15.1 使用JMH进行基准测试BenchmarkMode(Mode.AverageTime) OutputTimeUnit(TimeUnit.MICROSECONDS) public class MyBenchmark { Benchmark public void testMethod() { // 测试代码 } }15.2 使用Async Profiler./profiler.sh -d 30 -f profile.html pid15.3 分析线程争用ThreadMXBean threadMXBean ManagementFactory.getThreadMXBean(); long[] threadIds threadMXBean.findDeadlockedThreads(); if (threadIds ! null) { ThreadInfo[] infos threadMXBean.getThreadInfo(threadIds); for (ThreadInfo info : infos) { System.out.println(info.getThreadName() is waiting on info.getLockName() held by info.getLockOwnerName()); } }16. 线程安全设计模式16.1 单例模式public class Singleton { private static volatile Singleton instance; private Singleton() {} public static Singleton getInstance() { if (instance null) { synchronized (Singleton.class) { if (instance null) { instance new Singleton(); } } } return instance; } }16.2 生产者-消费者模式BlockingQueueItem queue new LinkedBlockingQueue(); // 生产者 new Thread(() - { while (true) { Item item produceItem(); queue.put(item); } }).start(); // 消费者 new Thread(() - { while (true) { Item item queue.take(); consumeItem(item); } }).start();16.3 工作窃取模式ForkJoinPool pool new ForkJoinPool(); class MyTask extends RecursiveAction { protected void compute() { // 任务分解与执行 } } pool.invoke(new MyTask());17. 线程与内存模型17.1 happens-before关系程序顺序规则监视器锁规则volatile变量规则线程启动规则线程终止规则线程中断规则终结器规则传递性17.2 内存屏障// 使用volatile实现内存屏障 volatile boolean flag false; public void writer() { // 写操作 flag true; // 插入StoreStore屏障 } public void reader() { if (flag) { // 插入LoadLoad屏障 // 读操作 } }17.3 final字段安全性class FinalFieldExample { final int x; public FinalFieldExample() { x 42; // 正确初始化final字段 } }18. 线程与异常处理18.1 未捕获异常处理器Thread.setDefaultUncaughtExceptionHandler((t, e) - { System.err.println(Uncaught exception in thread t.getName()); e.printStackTrace(); });18.2 Future异常处理Future? future executor.submit(() - { // 可能抛出异常的任务 }); try { future.get(); } catch (ExecutionException e) { Throwable cause e.getCause(); // 处理任务抛出的异常 }18.3 CompletableFuture异常处理CompletableFuture.supplyAsync(() - { // 可能抛出异常的任务 return result; }).exceptionally(ex - { // 处理异常 return fallback; });19. 线程与I/O操作19.1 异步I/OAsynchronousFileChannel channel AsynchronousFileChannel.open( Paths.get(file.txt), StandardOpenOption.READ); ByteBuffer buffer ByteBuffer.allocate(1024); channel.read(buffer, 0, buffer, new CompletionHandlerInteger, ByteBuffer() { public void completed(Integer result, ByteBuffer attachment) { // 读取完成处理 } public void failed(Throwable exc, ByteBuffer attachment) { // 读取失败处理 } });19.2 NIO与多路复用Selector selector Selector.open(); ServerSocketChannel serverChannel ServerSocketChannel.open(); serverChannel.configureBlocking(false); serverChannel.register(selector, SelectionKey.OP_ACCEPT); while (true) { selector.select(); SetSelectionKey keys selector.selectedKeys(); for (SelectionKey key : keys) { if (key.isAcceptable()) { // 处理连接请求 } else if (key.isReadable()) { // 处理读事件 } } keys.clear(); }19.3 使用Netty处理并发I/OEventLoopGroup group new NioEventLoopGroup(); try { ServerBootstrap b new ServerBootstrap(); b.group(group) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializerSocketChannel() { Override public void initChannel(SocketChannel ch) { ch.pipeline().addLast(new EchoServerHandler()); } }); ChannelFuture f b.bind(8080).sync(); f.channel().closeFuture().sync(); } finally { group.shutdownGracefully(); }20. 线程与数据库交互20.1 连接池配置HikariConfig config new HikariConfig(); config.setJdbcUrl(jdbc:mysql://localhost:3306/db); config.setUsername(user); config.setPassword(password); config.setMaximumPoolSize(20); // 最大连接数 config.setMinimumIdle(5); // 最小空闲连接 HikariDataSource ds new HikariDataSource(config);20.2 事务隔离级别READ_UNCOMMITTEDREAD_COMMITTEDREPEATABLE_READSERIALIZABLE20.3 批量操作优化try (Connection conn ds.getConnection(); PreparedStatement stmt conn.prepareStatement(INSERT INTO table VALUES (?))) { conn.setAutoCommit(false); for (int i 0; i 1000; i) { stmt.setInt(1, i); stmt.addBatch(); if (i % 100 0) { stmt.executeBatch(); conn.commit(); } } stmt.executeBatch(); conn.commit(); }21. 线程与缓存21.1 使用Caffeine缓存CacheString, Data cache Caffeine.newBuilder() .maximumSize(10_000) .expireAfterWrite(10, TimeUnit.MINUTES) .build(); // 自动加载缓存 LoadingCacheString, Data loadingCache Caffeine.newBuilder() .maximumSize(10_000) .build(key - loadDataFromDatabase(key));21.2 缓存一致性策略写穿透(Write-through)写回(Write-behind)失效(Cache-aside)读穿透(Read-through)21.3 分布式缓存CacheManager cacheManager Caching.getCachingProvider() .getCacheManager(); CacheString, String cache cacheManager.createCache(myCache, new MutableConfigurationString, String() .setTypes(String.class, String.class) .setExpiryPolicyFactory(AccessedExpiryPolicy.factoryOf(Duration.ONE_HOUR)) .setStoreByValue(false));22. 线程与微服务22.1 服务调用超时设置HystrixCommand(fallbackMethod fallbackMethod, commandProperties { HystrixProperty(name execution.isolation.thread.timeoutInMilliseconds, value 1000) }) public String callService() { // 调用远程服务 }22.2 熔断器配置HystrixCommand(fallbackMethod fallbackMethod, commandProperties { HystrixProperty(name circuitBreaker.requestVolumeThreshold, value 20), HystrixProperty(name circuitBreaker.sleepWindowInMilliseconds, value 5000), HystrixProperty(name circuitBreaker.errorThresholdPercentage, value 50) }) public String callService() { // 调用远程服务 }22.3 异步服务调用Async public CompletableFutureString asyncCall() { // 异步执行的任务 return CompletableFuture.completedFuture(result); }23. 线程与函数式编程23.1 并行流ListString results dataList.parallelStream() .filter(item - item.startsWith(A)) .map(String::toUpperCase) .collect(Collectors.toList());23.2 CompletableFuture组合CompletableFutureString future1 CompletableFuture.supplyAsync(() - Hello); CompletableFutureString future2 CompletableFuture.supplyAsync(() - World); CompletableFutureString combined future1.thenCombine(future2, (s1, s2) - s1 s2);23.3 反应式编程FluxString flux Flux.fromIterable(dataList) .parallel() .runOn(Schedulers.parallel()) .map(String::toUpperCase) .sequential();24. 线程与测试24.1 并发测试Test public void testConcurrentAccess() throws InterruptedException { final int THREAD_COUNT 10; ExecutorService executor Executors.newFixedThreadPool(THREAD_COUNT); CountDownLatch latch new CountDownLatch(THREAD_COUNT); for (int i 0; i THREAD_COUNT; i) { executor.execute(() - { try { // 测试代码 } finally { latch.countDown(); } }); } latch.await(); executor.shutdown(); }24.2 使用JMeter进行压力测试创建线程组设置并发用户数添加HTTP请求采样器配置断言和监听器分析测试结果24.3 使用TestContainers测试并发数据库访问Testcontainers class DatabaseTest { Container static PostgreSQLContainer? postgres new PostgreSQLContainer(postgres:13); Test void testConcurrentTransactions() { // 并发测试数据库访问 } }25. 线程与安全25.1 线程安全的密码哈希String hashed new BCryptPasswordEncoder().encode(password);25.2 安全随机数生成SecureRandom random SecureRandom.getInstanceStrong(); byte[] bytes new byte[16]; random.nextBytes(bytes);25.3 线程安全的日志记录private static final Logger logger LoggerFactory.getLogger(MyClass.class); public void method() { logger.info(Thread-safe log message); }26. 线程与性能调优26.1 线程上下文切换分析ThreadMXBean threadMXBean ManagementFactory.getThreadMXBean(); long totalContextSwitches 0; for (long id : threadMXBean.getAllThreadIds()) { totalContextSwitches threadMXBean.getThreadInfo(id).getWaitedCount(); }26.2 CPU缓存优化// 伪共享问题解决方案 Contended public class VolatileLong { public volatile long value 0L; }26.3 内存屏障使用// 使用Unsafe实现内存屏障 Unsafe unsafe Unsafe.getUnsafe(); unsafe.storeFence(); // 插入写屏障 unsafe.loadFence(); // 插入读屏障 unsafe.fullFence(); // 插入全屏障27. 线程与容器化27.1 Kubernetes线程池配置apiVersion: apps/v1 kind: Deployment spec: template: spec: containers: - name: app resources: limits: cpu: 2 requests: cpu: 127.2 容器内线程数计算int availableProcessors Runtime.getRuntime().availableProcessors(); int threadPoolSize Math.max(2, availableProcessors * 2);27.3 优雅停机处理Runtime.getRuntime().addShutdownHook(new Thread(() - { executor.shutdown(); try { if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); } }));28. 线程与云原生28.1 无服务器函数并发控制// AWS Lambda并发控制 public class Handler implements RequestHandlerInput, Output { private static final ExecutorService executor Executors.newFixedThreadPool(10); public Output handleRequest(Input input, Context context) { FutureOutput future executor.submit(() - process(input)); return future.get(); } }28.2 消息队列消费者线程池Bean public ConcurrentKafkaListenerContainerFactoryString, String kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactoryString, String factory new ConcurrentKafkaListenerContainerFactory(); factory.setConcurrency(3); // 每个监听器3个线程 return factory; }28.3 分布式锁实现// 使用Redis实现分布式锁 public boolean tryLock(String lockKey, String requestId, int expireTime) { return redisTemplate.opsForValue().setIfAbsent( lockKey, requestId, expireTime, TimeUnit.SECONDS); } public boolean releaseLock(String lockKey, String requestId) { String script if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end; Long result redisTemplate.execute( new DefaultRedisScript(script, Long.class), Collections.singletonList(lockKey), requestId); return result ! null result 1; }29. 线程与机器学习29.1 并行模型训练// 使用并行流处理数据 double[] weights dataList.parallelStream() .mapToDouble(this::computeWeight) .toArray();29.2 使用ForkJoinPool处理大数据class TrainingTask extends RecursiveAction { private final double[] data; private final int start; private final int end; protected void compute() { if (end - start THRESHOLD) { // 直接计算 } else { int mid (start end) / 2; invokeAll(new TrainingTask(data, start, mid), new TrainingTask(data, mid, end)); } } }29.3 异步模型预测CompletableFuturePredictionResult future CompletableFuture.supplyAsync(() - { return model.predict(input); }, executor);30. 线程与区块链30.1 挖矿线程池ExecutorService miningPool Executors.newWorkStealingPool(); public Block mineBlock(Blockchain blockchain, ListTransaction transactions) { ListFutureBlock futures new ArrayList(); for (int i 0; i Runtime.getRuntime().availableProcessors(); i) { futures.add(miningPool.submit(() - { return blockchain.mineBlock(transactions); })); } return futures.stream() .map(f - { try { return f.get(); } catch (Exception e) { return null; } }) .filter(Objects::nonNull) .findFirst() .orElseThrow(() - new RuntimeException(Mining failed)); }30.2 共识算法实现// 简单的PBFT实现 public class PBFTNode { private final ExecutorService executor Executors.newCachedThreadPool(); private final ListNode nodes; public void onReceiveMessage(Message message) { executor.execute(() - { switch (message.getType()) { case PRE_PREPARE: // 处理预准备消息 break; case PREPARE: // 处理准备消息 break; case COMMIT: // 处理提交消息 break; } }); } }30.3 交易并行验证public boolean validateTransactions(ListTransaction transactions) { return transactions.parallelStream() .allMatch(this::validateTransaction); }