1. 線程操作基礎(chǔ)概念線程是操作系統(tǒng)能夠進行運算調(diào)度的最小單位它被包含在進程之中是進程中的實際運作單位。一個進程可以包含多個線程這些線程共享進程的內(nèi)存空間和系統(tǒng)資源。線程與進程的主要區(qū)別在于進程是資源分配的基本單位線程是CPU調(diào)度的基本單位同一進程內(nèi)的線程共享內(nèi)存空間進程間的通信需要特殊機制而線程間可以直接讀寫數(shù)據(jù)2. 線程的創(chuàng)建與啟動在大多數(shù)現(xiàn)代編程語言中創(chuàng)建線程通常有以下幾種方式2.1 繼承Thread類class MyThread extends Thread { public void run() { // 線程執(zhí)行的代碼 } } // 創(chuàng)建并啟動線程 MyThread t new MyThread(); t.start();2.2 實現(xiàn)Runnable接口class MyRunnable implements Runnable { public void run() { // 線程執(zhí)行的代碼 } } // 創(chuàng)建并啟動線程 Thread t new Thread(new MyRunnable()); t.start();2.3 使用線程池ExecutorService executor Executors.newFixedThreadPool(5); executor.execute(new Runnable() { public void run() { // 線程執(zhí)行的代碼 } });3. 線程的生命周期管理線程在其生命周期中會經(jīng)歷多種狀態(tài)新建(NEW)線程對象被創(chuàng)建但尚未啟動可運行(RUNNABLE)線程正在JVM中執(zhí)行或等待CPU時間片阻塞(BLOCKED)線程等待獲取監(jiān)視器鎖等待(WAITING)線程無限期等待其他線程執(zhí)行特定操作超時等待(TIMED_WAITING)線程在指定時間內(nèi)等待終止(TERMINATED)線程已完成執(zhí)行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 { // 臨界區(qū)代碼 } 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, // 核心線程數(shù) 10, // 最大線程數(shù) 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); // 延遲執(zhí)行 scheduler.schedule(() - { // 任務代碼 }, 10, TimeUnit.SECONDS); // 周期性執(zhí)行 scheduler.scheduleAtFixedRate(() - { // 任務代碼 }, 0, 1, TimeUnit.SECONDS);7. 并發(fā)工具類7.1 CountDownLatchCountDownLatch latch new CountDownLatch(3); // 工作線程 new Thread(() - { // 執(zhí)行任務 latch.countDown(); }).start(); // 主線程等待 latch.await();7.2 CyclicBarrierCyclicBarrier barrier new CyclicBarrier(3, () - { // 所有線程到達屏障后執(zhí)行 }); new Thread(() - { // 執(zhí)行任務 barrier.await(); }).start();7.3 SemaphoreSemaphore semaphore new Semaphore(3); // 允許3個線程同時訪問 public void method() throws InterruptedException { semaphore.acquire(); try { // 臨界區(qū)代碼 } 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. 并發(fā)集合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. 線程性能優(yōu)化9.1 減少鎖競爭縮小同步代碼塊范圍使用讀寫鎖替代獨占鎖使用無鎖數(shù)據(jù)結(jié)構(gòu)9.2 避免死鎖按固定順序獲取多個鎖使用tryLock()設(shè)置超時避免在持有鎖時調(diào)用外部方法9.3 線程池調(diào)優(yōu)根據(jù)任務類型選擇合適的工作隊列合理設(shè)置核心線程數(shù)和最大線程數(shù)使用合適的拒絕策略10. 線程調(diào)試與監(jiān)控10.1 線程轉(zhuǎn)儲分析# 獲取Java進程的線程轉(zhuǎn)儲 jstack pid thread_dump.txt10.2 使用VisualVM監(jiān)控線程查看線程狀態(tài)檢測死鎖分析線程CPU使用情況10.3 使用JConsole監(jiān)控線程數(shù)量查看線程堆棧檢測死鎖情況11. 常見線程問題與解決方案11.1 死鎖// 錯誤的鎖獲取順序可能導致死鎖 public void transfer(Account from, Account to, int amount) { synchronized(from) { synchronized(to) { // 轉(zhuǎn)賬操作 } } }解決方案// 使用固定的鎖獲取順序 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) { // 轉(zhuǎn)賬操作 } } }11.2 活鎖// 兩個線程不斷改變狀態(tài)導致無法繼續(xù)執(zhí)行 while (!tryAcquireLock()) { // 釋放資源并重試 releaseSomeResources(); Thread.sleep(100); // 隨機延遲可以緩解活鎖 }11.3 線程饑餓確保公平的鎖獲取機制避免高優(yōu)先級線程獨占資源使用公平鎖或合理的調(diào)度策略12. 現(xiàn)代并發(fā)模式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 - { // 處理結(jié)果 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. 線程最佳實踐優(yōu)先使用線程池避免頻繁創(chuàng)建和銷毀線程合理設(shè)置線程優(yōu)先級大多數(shù)情況下使用默認優(yōu)先級避免過度同步只在必要時使用同步機制使用并發(fā)工具類優(yōu)先使用Java并發(fā)包中的高級工具注意線程安全確保共享資源的線程安全訪問合理處理異常為線程設(shè)置未捕獲異常處理器避免長時間持有鎖減少鎖的持有時間考慮使用不可變對象簡化線程安全設(shè)計合理使用volatile確保變量的可見性測試并發(fā)代碼使用壓力測試驗證并發(fā)正確性14. 線程調(diào)試技巧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監(jiān)控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. 線程安全設(shè)計模式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 生產(chǎn)者-消費者模式BlockingQueueItem queue new LinkedBlockingQueue(); // 生產(chǎn)者 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() { // 任務分解與執(zhí)行 } } pool.invoke(new MyTask());17. 線程與內(nèi)存模型17.1 happens-before關(guān)系程序順序規(guī)則監(jiān)視器鎖規(guī)則volatile變量規(guī)則線程啟動規(guī)則線程終止規(guī)則線程中斷規(guī)則終結(jié)器規(guī)則傳遞性17.2 內(nèi)存屏障// 使用volatile實現(xiàn)內(nèi)存屏障 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處理并發(fā)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. 線程與數(shù)據(jù)庫交互20.1 連接池配置HikariConfig config new HikariConfig(); config.setJdbcUrl(jdbc:mysql://localhost:3306/db); config.setUsername(user); config.setPassword(password); config.setMaximumPoolSize(20); // 最大連接數(shù) config.setMinimumIdle(5); // 最小空閑連接 HikariDataSource ds new HikariDataSource(config);20.2 事務隔離級別READ_UNCOMMITTEDREAD_COMMITTEDREPEATABLE_READSERIALIZABLE20.3 批量操作優(yōu)化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 服務調(diào)用超時設(shè)置HystrixCommand(fallbackMethod fallbackMethod, commandProperties { HystrixProperty(name execution.isolation.thread.timeoutInMilliseconds, value 1000) }) public String callService() { // 調(diào)用遠程服務 }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() { // 調(diào)用遠程服務 }22.3 異步服務調(diào)用Async public CompletableFutureString asyncCall() { // 異步執(zhí)行的任務 return CompletableFuture.completedFuture(result); }23. 線程與函數(shù)式編程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 并發(fā)測試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進行壓力測試創(chuàng)建線程組設(shè)置并發(fā)用戶數(shù)添加HTTP請求采樣器配置斷言和監(jiān)聽器分析測試結(jié)果24.3 使用TestContainers測試并發(fā)數(shù)據(jù)庫訪問Testcontainers class DatabaseTest { Container static PostgreSQLContainer? postgres new PostgreSQLContainer(postgres:13); Test void testConcurrentTransactions() { // 并發(fā)測試數(shù)據(jù)庫訪問 } }25. 線程與安全25.1 線程安全的密碼哈希String hashed new BCryptPasswordEncoder().encode(password);25.2 安全隨機數(shù)生成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. 線程與性能調(diào)優(yōu)26.1 線程上下文切換分析ThreadMXBean threadMXBean ManagementFactory.getThreadMXBean(); long totalContextSwitches 0; for (long id : threadMXBean.getAllThreadIds()) { totalContextSwitches threadMXBean.getThreadInfo(id).getWaitedCount(); }26.2 CPU緩存優(yōu)化// 偽共享問題解決方案 Contended public class VolatileLong { public volatile long value 0L; }26.3 內(nèi)存屏障使用// 使用Unsafe實現(xiàn)內(nèi)存屏障 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 容器內(nèi)線程數(shù)計算int availableProcessors Runtime.getRuntime().availableProcessors(); int threadPoolSize Math.max(2, availableProcessors * 2);27.3 優(yōu)雅停機處理Runtime.getRuntime().addShutdownHook(new Thread(() - { executor.shutdown(); try { if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); } }));28. 線程與云原生28.1 無服務器函數(shù)并發(fā)控制// AWS Lambda并發(fā)控制 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); // 每個監(jiān)聽器3個線程 return factory; }28.3 分布式鎖實現(xiàn)// 使用Redis實現(xiàn)分布式鎖 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 并行模型訓練// 使用并行流處理數(shù)據(jù) double[] weights dataList.parallelStream() .mapToDouble(this::computeWeight) .toArray();29.2 使用ForkJoinPool處理大數(shù)據(jù)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. 線程與區(qū)塊鏈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 共識算法實現(xiàn)// 簡單的PBFT實現(xiàn) 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); }