ConcurrentHashMap 多线程并发数

在 JDK1.8 中，`ConcurrentHashMap` 的并发度是由其内部的 Segment 数组的长度决定的，而不是由 node 数组的长度决定的。每个 Segment 内部都是一个类似于 HashTable 的结构，可以独立地进行加锁和解锁操作，这样多个线程可以同时访问不同的 Segment，从而提高了并发度。 在 JDK1.8 中，`ConcurrentHashMap` 默认创建 16 个 Segment，因此其默认的并发度为 16。如果需要更高的并发度，可以通过构造方法中的 `concurrencyLevel` 参数指定。请注意，虽然增加 Segment 数量可以提高并发度，但也会带来一定的空间开销和时间开销。因此，在实际使用中需要根据具体情况进行权衡和调整。

<think>在Java中实现多线程并发处理任务有多种方式，包括直接使用Thread类、实现Runnable接口、使用ExecutorService框架、使用Fork/Join框架等。下面将详细介绍几种常用的方法，并给出代码示例。 ### 1. 继承Thread类 通过继承Thread类并重写run()方法来实现多线程。 ```java public class MyThread extends Thread { @Override public void run() { System.out.println("线程运行中: " + Thread.currentThread().getName()); } public static void main(String[] args) { MyThread t1 = new MyThread(); MyThread t2 = new MyThread(); t1.start(); t2.start(); } } ``` ### 2. 实现Runnable接口 实现Runnable接口，然后将Runnable实例作为参数传递给Thread类的构造器。 ```java public class MyRunnable implements Runnable { @Override public void run() { System.out.println("线程运行中: " + Thread.currentThread().getName()); } public static void main(String[] args) { Thread t1 = new Thread(new MyRunnable()); Thread t2 = new Thread(new MyRunnable()); t1.start(); t2.start(); } } ``` ### 3. 使用ExecutorService 使用ExecutorService线程池来管理线程，可以更有效地控制线程的创建和销毁。 ```java import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class ExecutorExample { public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(2); Runnable task1 = () -> System.out.println("任务1运行在: " + Thread.currentThread().getName()); Runnable task2 = () -> System.out.println("任务2运行在: " + Thread.currentThread().getName()); executor.submit(task1); executor.submit(task2); executor.shutdown(); // 关闭线程池 } } ``` ### 4. 使用Callable和Future 如果需要获取线程执行的结果，可以使用Callable和Future。 ```java import java.util.concurrent.*; public class CallableExample { public static void main(String[] args) throws ExecutionException, InterruptedException { ExecutorService executor = Executors.newFixedThreadPool(2); Callable<String> task1 = () -> "任务1完成"; Callable<String> task2 = () -> "任务2完成"; Future<String> future1 = executor.submit(task1); Future<String> future2 = executor.submit(task2); System.out.println(future1.get()); System.out.println(future2.get()); executor.shutdown(); } } ``` ### 5. 使用CompletableFuture（Java 8及以上） CompletableFuture提供了更强大的异步编程能力，支持链式调用和组合多个异步任务。 ```java import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; public class CompletableFutureExample { public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello"); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World"); // 组合两个任务 CompletableFuture<String> combined = future1.thenCombine(future2, (s1, s2) -> s1 + " " + s2); System.out.println(combined.get()); // 输出: Hello World } } ``` ### 6. 使用Fork/Join框架 Fork/Join框架适用于任务可以递归拆分的场景，如大规模数据处理。 ```java import java.util.concurrent.RecursiveTask; import java.util.concurrent.ForkJoinPool; public class FibonacciTask extends RecursiveTask<Integer> { private final int n; public FibonacciTask(int n) { this.n = n; } @Override protected Integer compute() { if (n <= 1) { return n; } FibonacciTask task1 = new FibonacciTask(n - 1); task1.fork(); FibonacciTask task2 = new FibonacciTask(n - 2); return task2.compute() + task1.join(); } public static void main(String[] args) { ForkJoinPool pool = new ForkJoinPool(); FibonacciTask task = new FibonacciTask(10); System.out.println(pool.invoke(task)); // 输出: 55 } } ``` ### 解释： - **继承Thread类**：简单直接，但Java不支持多重继承，因此如果已经继承了其他类，则无法使用此方法。 - **实现Runnable接口**：更灵活，因为一个类可以实现多个接口，还可以继承其他类。 - **ExecutorService**：提供了线程池管理，避免了频繁创建和销毁线程的开销，提高了性能。 - **Callable和Future**：Callable可以返回结果和抛出异常，Future用于获取异步计算的结果。 - **CompletableFuture**：提供了更高级的异步编程方式，支持函数式编程，可以组合多个异步任务。 - **Fork/Join框架**：适用于将大任务拆分成小任务并行执行，然后合并结果，特别适合递归任务。 ### 注意事项： - 多线程编程中需要注意线程安全问题，如共享变量的访问需要同步（synchronized）或使用并发工具类（如AtomicInteger, ConcurrentHashMap等）。 - 避免死锁，确保线程获取锁的顺序一致。 - 合理使用线程池，避免资源耗尽（如使用无界队列可能导致OOM）。 §§相关问题§§: 1. Java中如何保证线程安全？ 2. 什么是线程池？为什么使用线程池？ 3. synchronized关键字和Lock接口有什么区别？ 4. 什么是死锁？如何避免死锁？</think>在Java中实现多线程并发处理任务有多种方式，下面我将详细介绍四种常用方法及其代码实现： ### 1. 使用基础Thread类（不推荐用于复杂任务） ```java public class BasicThreadExample { public static void main(String[] args) { // 创建任务对象 Runnable task = () -> { System.out.println(Thread.currentThread().getName() + " 正在处理任务"); // 模拟任务处理耗时 try { Thread.sleep(1000); } catch (InterruptedException e) {} }; // 创建并启动多个线程 for (int i = 0; i < 5; i++) { new Thread(task, "Worker-" + i).start(); } } } ``` ### 2. 使用ExecutorService线程池（推荐） ```java import java.util.concurrent.*; public class ThreadPoolExample { public static void main(String[] args) { // 创建固定大小的线程池 ExecutorService executor = Executors.newFixedThreadPool(3); // 提交10个任务 for (int i = 1; i <= 10; i++) { final int taskId = i; executor.submit(() -> { System.out.println(Thread.currentThread().getName() + " 处理任务-" + taskId); try { Thread.sleep(500); } catch (InterruptedException e) {} }); } // 关闭线程池 executor.shutdown(); try { // 等待所有任务完成（最多等待1分钟） executor.awaitTermination(1, TimeUnit.MINUTES); } catch (InterruptedException e) { e.printStackTrace(); } } } ``` ### 3. 使用Fork/Join框架（适合可拆分任务） ```java import java.util.concurrent.*; public class ForkJoinExample { static class SumTask extends RecursiveTask<Long> { private final long[] array; private final int start, end; SumTask(long[] array, int start, int end) { this.array = array; this.start = start; this.end = end; } @Override protected Long compute() { // 小任务直接计算 if (end - start <= 1000) { long sum = 0; for (int i = start; i < end; i++) { sum += array[i]; } return sum; } // 大任务拆分成两个子任务 int mid = (start + end) >>> 1; SumTask left = new SumTask(array, start, mid); SumTask right = new SumTask(array, mid, end); // 并行执行子任务 left.fork(); long rightResult = right.compute(); long leftResult = left.join(); return leftResult + rightResult; } } public static void main(String[] args) { // 创建测试数组 long[] array = new long[10_000]; Arrays.fill(array, 1); // 创建ForkJoinPool ForkJoinPool pool = new ForkJoinPool(); long result = pool.invoke(new SumTask(array, 0, array.length)); System.out.println("计算结果: " + result); // 应输出10000 } } ``` ### 4. 使用CompletableFuture（Java8+ 异步编程） ```java import java.util.concurrent.*; public class CompletableFutureExample { public static void main(String[] args) { // 创建线程池 ExecutorService executor = Executors.newFixedThreadPool(4); // 创建异步任务链 CompletableFuture.supplyAsync(() -> { System.out.println("任务1开始"); try { Thread.sleep(1000); } catch (InterruptedException e) {} return 10; }, executor).thenApplyAsync(result -> { System.out.println("任务2处理结果: " + result); return result * 2; }, executor).thenAcceptAsync(finalResult -> { System.out.println("最终结果: " + finalResult); }, executor); // 关闭线程池（实际生产环境应复用线程池） executor.shutdown(); } } ``` ### 关键概念解释： 1. **线程池（ExecutorService）**： - 管理线程生命周期，避免频繁创建/销毁线程的开销 - 控制并发线程数量（通过`newFixedThreadPool()`等方法） - 提供任务队列（默认使用无界队列，可通过`ThreadPoolExecutor`自定义） 2. **任务提交方式**： - `execute(Runnable)`：无返回结果 - `submit(Callable)`：返回Future对象获取结果 - `invokeAll()`：批量提交任务 3. **并发控制工具**： - `CountDownLatch`：等待多个任务完成 - `CyclicBarrier`：多线程到达屏障点后继续执行 - `Semaphore`：控制同时访问的线程数量 4. **线程安全注意事项**： - 使用`ConcurrentHashMap`等并发集合替代同步集合 - 使用`AtomicInteger`等原子类替代基本类型 - 避免共享可变状态，使用局部变量或ThreadLocal ### 最佳实践建议： 1. 优先使用线程池（`ExecutorService`）而非直接创建线程 2. 合理设置线程池大小（CPU密集型任务：CPU核心数+1；IO密集型任务：可更大） 3. 使用`Callable`+`Future`获取任务结果，而非共享变量 4. 需要任务组合时优先选择`CompletableFuture` 5. 大计算量任务考虑使用Fork/Join框架