Java并发编程：深入理解 `java.util.concurrent`

简介

在Java编程中，多线程和并发处理是提升程序性能和响应性的关键技术。java.util.concurrent 包提供了丰富的工具和类，帮助开发者更轻松、高效地处理并发任务。本文将深入探讨 java.util.concurrent 的基础概念、使用方法、常见实践以及最佳实践，帮助读者全面掌握这一强大的并发编程工具集。

基础概念

线程与并发

在Java中，线程是程序中执行的最小单元。并发是指在同一时间段内，多个任务似乎在同时执行。实际上，在单核CPU系统中，这些任务是通过快速切换轮流执行的；而在多核CPU系统中，多个任务可以真正并行执行。

并发工具类概述

java.util.concurrent 包包含了一系列用于处理并发编程的类和接口，主要包括： - 线程池：ExecutorService 及其实现类，用于管理和复用线程，提高线程的创建和销毁效率。 - 并发集合：如 ConcurrentHashMap、CopyOnWriteArrayList 等，这些集合类在多线程环境下能安全高效地进行读写操作。 - 同步工具类：例如 CountDownLatch、CyclicBarrier 和 Semaphore，用于协调多个线程之间的执行顺序和资源访问。

使用方法

线程池的使用

线程池通过 ExecutorService 接口来管理线程。以下是一个简单的线程池使用示例：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolExample {
    public static void main(String[] args) {
        // 创建一个固定大小为 3 的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        for (int i = 0; i < 5; i++) {
            final int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

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

并发集合的应用

以 ConcurrentHashMap 为例，它是一个线程安全的哈希表，适合在多线程环境下进行读写操作：

import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentHashMapExample {
    public static void main(String[] args) {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

        // 多个线程可以同时安全地进行读写操作
        map.put("one", 1);
        map.put("two", 2);

        System.out.println(map.get("one"));
    }
}

同步工具类介绍

CountDownLatch

CountDownLatch 允许一个或多个线程等待，直到其他线程完成一组操作。例如：

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {
    public static void main(String[] args) {
        int numThreads = 3;
        CountDownLatch latch = new CountDownLatch(numThreads);

        for (int i = 0; i < numThreads; i++) {
            final int threadNumber = i;
            new Thread(() -> {
                System.out.println("Thread " + threadNumber + " started");
                try {
                    Thread.sleep((long) (Math.random() * 2000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Thread " + threadNumber + " finished");
                latch.countDown();
            }).start();
        }

        try {
            latch.await();
            System.out.println("All threads have finished");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

CyclicBarrier

CyclicBarrier 用于让一组线程在某个点上相互等待，然后再一起继续执行。例如：

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {
    public static void main(String[] args) {
        int numThreads = 3;
        CyclicBarrier barrier = new CyclicBarrier(numThreads, () -> {
            System.out.println("All threads have reached the barrier");
        });

        for (int i = 0; i < numThreads; i++) {
            final int threadNumber = i;
            new Thread(() -> {
                System.out.println("Thread " + threadNumber + " started");
                try {
                    Thread.sleep((long) (Math.random() * 2000));
                    barrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                System.out.println("Thread " + threadNumber + " continued after barrier");
            }).start();
        }
    }
}

Semaphore

Semaphore 用于控制对共享资源的访问数量。例如：

import java.util.concurrent.Semaphore;

public class SemaphoreExample {
    public static void main(String[] args) {
        int availablePermits = 2;
        Semaphore semaphore = new Semaphore(availablePermits);

        for (int i = 0; i < 5; i++) {
            final int threadNumber = i;
            new Thread(() -> {
                try {
                    semaphore.acquire();
                    System.out.println("Thread " + threadNumber + " acquired a permit");
                    Thread.sleep((long) (Math.random() * 2000));
                    System.out.println("Thread " + threadNumber + " released a permit");
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

常见实践

多线程数据处理

在处理大量数据时，可以使用线程池将数据分成多个部分，由不同线程并行处理，提高处理效率。例如：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ParallelDataProcessing {
    public static void main(String[] args) {
        int[] data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
        int numThreads = 4;
        ExecutorService executorService = Executors.newFixedThreadPool(numThreads);

        int chunkSize = data.length / numThreads;
        for (int i = 0; i < numThreads; i++) {
            final int start = i * chunkSize;
            final int end = (i == numThreads - 1)? data.length : (i + 1) * chunkSize;
            executorService.submit(() -> {
                for (int j = start; j < end; j++) {
                    // 处理数据
                    System.out.println("Processing data " + data[j] + " on thread " + Thread.currentThread().getName());
                }
            });
        }

        executorService.shutdown();
    }
}

并发任务调度

使用 ScheduledExecutorService 可以实现定时任务的并发调度。例如：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledTaskExample {
    public static void main(String[] args) {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

        scheduler.scheduleAtFixedRate(() -> {
            System.out.println("Task 1 executed at " + System.currentTimeMillis());
        }, 0, 2, TimeUnit.SECONDS);

        scheduler.scheduleAtFixedRate(() -> {
            System.out.println("Task 2 executed at " + System.currentTimeMillis());
        }, 1, 3, TimeUnit.SECONDS);
    }
}

最佳实践

避免死锁

死锁是并发编程中常见的问题，当两个或多个线程相互等待对方释放资源时就会发生死锁。为了避免死锁，可以采取以下措施： - 按顺序获取锁：确保所有线程按照相同的顺序获取锁。 - 设置锁超时：使用 tryLock 方法并设置超时时间，避免无限期等待锁。

性能优化技巧

减少锁的粒度：尽量只在必要的代码块上使用锁，避免对整个方法加锁。
使用无锁数据结构：如 ConcurrentHashMap 和 CopyOnWriteArrayList，在某些场景下比传统的同步数据结构性能更好。

小结

java.util.concurrent 包为Java开发者提供了强大的并发编程工具，通过合理使用线程池、并发集合和同步工具类，可以实现高效、安全的多线程应用程序。在实际开发中，遵循最佳实践，如避免死锁和优化性能，能进一步提升程序的质量和稳定性。

参考资料

Java官方文档 - java.util.concurrent
《Effective Java》 - Joshua Bloch
《Java并发编程实战》 - Brian Goetz 等

Java并发编程：深入理解 java.util.concurrent

简介

目录