Java?JUC并發(fā)之.util.concurrent并發(fā)工具包使用指南

更新時(shí)間：2025年08月23日 09:43:51 作者：阿賈克斯的黎明

JUC就是java.util .concurrent工具包的簡稱,這是一個(gè)處理線程的工具包,JDK 1.5開始出現(xiàn)的,這篇文章主要介紹了Java?JUC并發(fā)之.util.concurrent并發(fā)工具包使用的相關(guān)資料,需要的朋友可以參考下

前言

JUC（Java Util Concurrent）即 Java 并發(fā)工具包，是java.util.concurrent包及其子包的簡稱，自 Java 5 引入，為并發(fā)編程提供了高效、安全、可靠的工具類，極大簡化了多線程編程的復(fù)雜度。

JUC 主要包含以下幾類組件：

線程池框架（Executor Framework）
并發(fā)集合（Concurrent Collections）
同步工具（Synchronizers）
原子操作類（Atomic Classes）
鎖機(jī)制（Locks）
并發(fā)工具類（如 CountDownLatch、CyclicBarrier 等）

線程池框架

線程池通過重用線程來減少線程創(chuàng)建和銷毀的開銷，提高系統(tǒng)性能。

核心接口與類

Executor：最基本的線程池接口，定義了執(zhí)行任務(wù)的方法
ExecutorService：擴(kuò)展了 Executor，提供了更豐富的線程池操作
ThreadPoolExecutor：線程池的核心實(shí)現(xiàn)類
Executors：線程池的工具類，提供了常用線程池的創(chuàng)建方法

線程池示例

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

public class ThreadPoolExample {
    public static void main(String[] args) {
        // 創(chuàng)建固定大小的線程池
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        // 提交任務(wù)
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                try {
                    System.out.println("任務(wù) " + taskId + " 由線程 " + 
                                       Thread.currentThread().getName() + " 執(zhí)行");
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        // 關(guān)閉線程池
        executor.shutdown();
        try {
            // 等待所有任務(wù)完成
            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                // 超時(shí)后強(qiáng)制關(guān)閉
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }
}

ThreadPoolExecutor 核心參數(shù)

手動(dòng)創(chuàng)建線程池時(shí)，ThreadPoolExecutor的構(gòu)造函數(shù)提供了最靈活的配置：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

corePoolSize：核心線程數(shù)
maximumPoolSize：最大線程數(shù)
keepAliveTime：非核心線程的空閑超時(shí)時(shí)間
workQueue：任務(wù)等待隊(duì)列
threadFactory：線程工廠
handler：拒絕策略

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class CustomThreadPool {
    public static void main(String[] args) {
        // 自定義線程池配置
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, // 核心線程數(shù)
            5, // 最大線程數(shù)
            30, // 空閑時(shí)間
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(10), // 有界隊(duì)列
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒絕策略
        );
        
        // 提交任務(wù)
        for (int i = 0; i < 20; i++) {
            final int taskId = i;
            executor.execute(() -> {
                try {
                    System.out.println("任務(wù) " + taskId + " 由線程 " +
                                       Thread.currentThread().getName() + " 執(zhí)行");
                    TimeUnit.MILLISECONDS.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        executor.shutdown();
    }
}

并發(fā)集合

JUC 提供了一系列線程安全的集合類，相比傳統(tǒng)的同步集合，通常具有更好的性能。

常用并發(fā)集合

ConcurrentHashMap：線程安全的 HashMap 替代者
CopyOnWriteArrayList：讀多寫少場景下的線程安全 List
CopyOnWriteArraySet：基于 CopyOnWriteArrayList 實(shí)現(xiàn)的 Set
ConcurrentLinkedQueue：高效的并發(fā)隊(duì)列
LinkedBlockingQueue：可阻塞的鏈表隊(duì)列
ArrayBlockingQueue：有界的數(shù)組隊(duì)列
PriorityBlockingQueue：支持優(yōu)先級的阻塞隊(duì)列

ConcurrentHashMap 示例

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ConcurrentHashMapExample {
    public static void main(String[] args) throws InterruptedException {
        Map<String, Integer> concurrentMap = new ConcurrentHashMap<>();
        ExecutorService executor = Executors.newFixedThreadPool(4);
        
        // 并發(fā)寫入
        for (int i = 0; i < 1000; i++) {
            final int num = i;
            executor.submit(() -> {
                String key = "key" + (num % 10);
                // 原子操作：計(jì)算并替換
                concurrentMap.compute(key, (k, v) -> v == null ? 1 : v + 1);
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
        
        // 輸出結(jié)果
        concurrentMap.forEach((k, v) -> System.out.println(k + ": " + v));
    }
}

同步工具類

JUC 提供了多種同步工具，用于協(xié)調(diào)多個(gè)線程之間的協(xié)作。

CountDownLatch

允許一個(gè)或多個(gè)線程等待其他線程完成操作。

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

public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        // 計(jì)數(shù)器為3
        CountDownLatch latch = new CountDownLatch(3);
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        for (int i = 0; i < 3; i++) {
            final int taskId = i;
            executor.submit(() -> {
                try {
                    System.out.println("任務(wù) " + taskId + " 開始執(zhí)行");
                    Thread.sleep(1000 + taskId * 500);
                    System.out.println("任務(wù) " + taskId + " 執(zhí)行完成");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    // 計(jì)數(shù)器減1
                    latch.countDown();
                }
            });
        }
        
        System.out.println("等待所有任務(wù)完成...");
        // 等待計(jì)數(shù)器變?yōu)?
        latch.await();
        System.out.println("所有任務(wù)已完成，繼續(xù)執(zhí)行主線程");
        
        executor.shutdown();
    }
}

CyclicBarrier

讓一組線程到達(dá)一個(gè)屏障時(shí)被阻塞，直到最后一個(gè)線程到達(dá)屏障，所有被阻塞的線程才會(huì)繼續(xù)執(zhí)行。

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

public class CyclicBarrierExample {
    public static void main(String[] args) {
        // 3個(gè)線程到達(dá)屏障后，執(zhí)行Runnable任務(wù)
        CyclicBarrier barrier = new CyclicBarrier(3, () -> 
            System.out.println("所有線程已到達(dá)屏障，開始下一步操作"));
        
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        for (int i = 0; i < 3; i++) {
            final int threadId = i;
            executor.submit(() -> {
                try {
                    System.out.println("線程 " + threadId + " 正在執(zhí)行任務(wù)");
                    Thread.sleep(1000 + threadId * 500);
                    System.out.println("線程 " + threadId + " 到達(dá)屏障");
                    // 等待其他線程到達(dá)
                    barrier.await();
                    System.out.println("線程 " + threadId + " 繼續(xù)執(zhí)行");
                } catch (InterruptedException | BrokenBarrierException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        executor.shutdown();
    }
}

Semaphore

信號量，用于控制同時(shí)訪問特定資源的線程數(shù)量。

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

public class SemaphoreExample {
    public static void main(String[] args) {
        // 允許3個(gè)線程同時(shí)訪問
        Semaphore semaphore = new Semaphore(3);
        ExecutorService executor = Executors.newFixedThreadPool(5);
        
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                try {
                    // 獲取許可
                    semaphore.acquire();
                    System.out.println("任務(wù) " + taskId + " 獲得許可，開始執(zhí)行");
                    TimeUnit.SECONDS.sleep(2);
                    System.out.println("任務(wù) " + taskId + " 執(zhí)行完成，釋放許可");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    // 釋放許可
                    semaphore.release();
                }
            });
        }
        
        executor.shutdown();
    }
}

原子操作類

JUC 提供了一系列原子操作類，用于在不使用鎖的情況下實(shí)現(xiàn)線程安全的原子操作。

主要原子類包括：

基本類型：AtomicInteger、AtomicLong、AtomicBoolean
數(shù)組類型：AtomicIntegerArray、AtomicLongArray等
引用類型：AtomicReference、AtomicStampedReference等
字段更新器：AtomicIntegerFieldUpdater等

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class AtomicExample {
    private static AtomicInteger counter = new AtomicInteger(0);
    
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        // 10個(gè)線程，每個(gè)線程自增1000次
        for (int i = 0; i < 10; i++) {
            executor.submit(() -> {
                for (int j = 0; j < 1000; j++) {
                    // 原子自增操作
                    counter.incrementAndGet();
                }
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
        
        // 結(jié)果應(yīng)該是10000
        System.out.println("最終計(jì)數(shù): " + counter.get());
    }
}

鎖機(jī)制

JUC 的java.util.concurrent.locks包提供了比synchronized更靈活的鎖機(jī)制。

Lock 接口

Lock接口是所有鎖的父接口，主要實(shí)現(xiàn)類有：

ReentrantLock：可重入鎖
ReentrantReadWriteLock：可重入讀寫鎖

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockExample {
    private static int count = 0;
    // 創(chuàng)建可重入鎖
    private static Lock lock = new ReentrantLock();
    
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        
        for (int i = 0; i < 1000; i++) {
            executor.submit(() -> {
                // 獲取鎖
                lock.lock();
                try {
                    count++;
                } finally {
                    // 確保鎖被釋放
                    lock.unlock();
                }
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
        
        System.out.println("最終計(jì)數(shù): " + count);
    }
}

讀寫鎖

ReentrantReadWriteLock提供了讀鎖和寫鎖分離，適合讀多寫少的場景：

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteLockExample {
    private Map<String, String> data = new HashMap<>();
    private ReadWriteLock lock = new ReentrantReadWriteLock();
    
    // 讀操作使用讀鎖
    public String get(String key) {
        lock.readLock().lock();
        try {
            System.out.println("讀取 key: " + key + "，線程: " + Thread.currentThread().getName());
            return data.get(key);
        } finally {
            lock.readLock().unlock();
        }
    }
    
    // 寫操作使用寫鎖
    public void put(String key, String value) {
        lock.writeLock().lock();
        try {
            System.out.println("寫入 key: " + key + "，線程: " + Thread.currentThread().getName());
            data.put(key, value);
        } finally {
            lock.writeLock().unlock();
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        ReadWriteLockExample example = new ReadWriteLockExample();
        ExecutorService executor = Executors.newFixedThreadPool(5);
        
        // 添加寫操作
        executor.submit(() -> example.put("name", "Java"));
        
        // 添加多個(gè)讀操作
        for (int i = 0; i < 4; i++) {
            executor.submit(() -> {
                for (int j = 0; j < 3; j++) {
                    example.get("name");
                    try {
                        TimeUnit.MILLISECONDS.sleep(100);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            });
        }
        
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
    }
}

實(shí)戰(zhàn)案例：生產(chǎn)者消費(fèi)者模型

使用 JUC 的阻塞隊(duì)列實(shí)現(xiàn)經(jīng)典的生產(chǎn)者消費(fèi)者模型：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ProducerConsumerExample {
    // 容量為10的阻塞隊(duì)列
    private static BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
    private static final int MAX_ITEMS = 20;
    
    // 生產(chǎn)者
    static class Producer implements Runnable {
        private int id;
        
        public Producer(int id) {
            this.id = id;
        }
        
        @Override
        public void run() {
            try {
                for (int i = 0; i < MAX_ITEMS; i++) {
                    int item = id * 100 + i;
                    queue.put(item); // 放入隊(duì)列，如果滿了會(huì)阻塞
                    System.out.println("生產(chǎn)者 " + id + " 生產(chǎn)了: " + item + 
                                       "，隊(duì)列大小: " + queue.size());
                    TimeUnit.MILLISECONDS.sleep(100); // 模擬生產(chǎn)耗時(shí)
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
    // 消費(fèi)者
    static class Consumer implements Runnable {
        private int id;
        
        public Consumer(int id) {
            this.id = id;
        }
        
        @Override
        public void run() {
            try {
                for (int i = 0; i < MAX_ITEMS; i++) {
                    int item = queue.take(); // 從隊(duì)列取，如果空了會(huì)阻塞
                    System.out.println("消費(fèi)者 " + id + " 消費(fèi)了: " + item + 
                                       "，隊(duì)列大小: " + queue.size());
                    TimeUnit.MILLISECONDS.sleep(150); // 模擬消費(fèi)耗時(shí)
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        
        // 創(chuàng)建2個(gè)生產(chǎn)者
        executor.submit(new Producer(1));
        executor.submit(new Producer(2));
        
        // 創(chuàng)建2個(gè)消費(fèi)者
        executor.submit(new Consumer(1));
        executor.submit(new Consumer(2));
        
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
    }
}

總結(jié)

JUC 為 Java 并發(fā)編程提供了強(qiáng)大的工具支持，大大簡化了多線程程序的開發(fā)難度。掌握 JUC 的使用，能夠幫助開發(fā)者編寫高效、安全的并發(fā)程序，應(yīng)對多線程環(huán)境下的各種挑戰(zhàn)。在實(shí)際開發(fā)中，應(yīng)根據(jù)具體場景選擇合適的并發(fā)工具，同時(shí)注意線程安全和性能之間的平衡。通過不斷實(shí)踐和深入理解這些工具的原理，您將能夠構(gòu)建出更健壯、更高效的并發(fā)應(yīng)用程序。

到此這篇關(guān)于Java JUC并發(fā)之.util.concurrent并發(fā)工具包使用指南的文章就介紹到這了,更多相關(guān)Java .util.concurrent并發(fā)工具內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: