Java?JUC并發(fā)之.util.concurrent并發(fā)工具包使用指南
前言
JUC(Java Util Concurrent)即 Java 并發(fā)工具包,是java.util.concurrent包及其子包的簡(jiǎn)稱,自 Java 5 引入,為并發(fā)編程提供了高效、安全、可靠的工具類,極大簡(jiǎn)化了多線程編程的復(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:讀多寫少場(chǎng)景下的線程安全 ListCopyOnWriteArraySet:基于 CopyOnWriteArrayList 實(shí)現(xiàn)的 SetConcurrentLinkedQueue:高效的并發(fā)隊(duì)列LinkedBlockingQueue:可阻塞的鏈表隊(duì)列ArrayBlockingQueue:有界的數(shù)組隊(duì)列PriorityBlockingQueue:支持優(yōu)先級(jí)的阻塞隊(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
信號(hào)量,用于控制同時(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提供了讀鎖和寫鎖分離,適合讀多寫少的場(chǎng)景:
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)大的工具支持,大大簡(jiǎn)化了多線程程序的開發(fā)難度。掌握 JUC 的使用,能夠幫助開發(fā)者編寫高效、安全的并發(fā)程序,應(yīng)對(duì)多線程環(huán)境下的各種挑戰(zhàn)。在實(shí)際開發(fā)中,應(yīng)根據(jù)具體場(chǎng)景選擇合適的并發(fā)工具,同時(shí)注意線程安全和性能之間的平衡。通過不斷實(shí)踐和深入理解這些工具的原理,您將能夠構(gòu)建出更健壯、更高效的并發(fā)應(yīng)用程序。
到此這篇關(guān)于Java JUC并發(fā)之.util.concurrent并發(fā)工具包使用指南的文章就介紹到這了,更多相關(guān)Java .util.concurrent并發(fā)工具內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- java.util.concurrent.ExecutionException 問題解決方法
- 深入Synchronized和java.util.concurrent.locks.Lock的區(qū)別詳解
- 出現(xiàn)java.util.ConcurrentModificationException 問題及解決辦法
- Java?報(bào)錯(cuò)?java.util.ConcurrentModificationException:?null?的原因及解決方案
- Java報(bào)錯(cuò):java.util.concurrent.ExecutionException的解決辦法
- java.util.ConcurrentModificationException 解決方法
- 淺談java.util.concurrent包中的線程池和消息隊(duì)列
- Java中JUC包(java.util.concurrent)下的常用子類
相關(guān)文章
springboot實(shí)現(xiàn)上傳并解析Excel過程解析
這篇文章主要介紹了springboot實(shí)現(xiàn)上傳并解析Excel過程解析,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-09-09
java數(shù)據(jù)隨機(jī)分頁實(shí)現(xiàn)方案
本文主要介紹了java數(shù)據(jù)隨機(jī)分頁實(shí)現(xiàn)方案,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-06-06
Failed to execute goal org...的解決辦法
這篇文章主要介紹了Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.1的解決辦法的相關(guān)資料,需要的朋友可以參考下2017-06-06
SpringBoot實(shí)現(xiàn)單點(diǎn)登錄的實(shí)現(xiàn)詳解
在現(xiàn)代的Web應(yīng)用程序中,單點(diǎn)登錄(Single?Sign-On)已經(jīng)變得越來越流行,在本文中,我們將使用Spring?Boot構(gòu)建一個(gè)基本的單點(diǎn)登錄系統(tǒng),需要的可以參考一下2023-05-05
mybatis中string和date的轉(zhuǎn)換方式
這篇文章主要介紹了mybatis中string和date的轉(zhuǎn)換方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-08-08
java實(shí)現(xiàn)清理DNS Cache的方法
這篇文章主要介紹了java實(shí)現(xiàn)清理DNS Cache的方法,分析了幾種常用的清理方法,并給出了反射清理的完整實(shí)例,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-01-01

