處理大規(guī)模數(shù)據(jù)計(jì)算任務(wù)，F(xiàn)ork/Join框架是您的得力助手！
1、JUC包概述
JUC是Java Util Concurrency的縮寫，即Java并發(fā)工具包。JUC包提供了一些常用的線程和并發(fā)編程工具類，幫助開發(fā)者更方便地開發(fā)多線程應(yīng)用程序，提高程序的并發(fā)性能。JUC包的主要特點(diǎn)包括：
- 安全性：JUC包提供了一些線程安全的數(shù)據(jù)結(jié)構(gòu)和工具類，如原子類、同步隊(duì)列等，可以保證多線程訪問時數(shù)據(jù)的正確性和一致性。
- 性能：JUC包中的一些并發(fā)工具類采用了高效的算法和數(shù)據(jù)結(jié)構(gòu)，如CAS算法、樂觀鎖等，可以提高程序的并發(fā)性能。
- 可擴(kuò)展性：JUC包中的一些工具類支持可擴(kuò)展性，如線程池、ForkJoin框架等，可以根據(jù)實(shí)際情況進(jìn)行動態(tài)調(diào)整。
2、原子操作類
在多線程環(huán)境下，由于多個線程同時訪問同一個變量可能會導(dǎo)致數(shù)據(jù)不一致的問題。原子操作類可以保證在多線程環(huán)境下對變量的操作是原子性的，即不會出現(xiàn)線程安全問題。
JJUC包中提供了以下幾個原子操作類：
- AtomicInteger：用于對int類型的變量進(jìn)行原子操作。
- AtomicLong：用于對long類型的變量進(jìn)行原子操作。
- AtomicBoolean：用于對boolean類型的變量進(jìn)行原子操作。
- AtomicIntegerArray：用于對int數(shù)組中的元素進(jìn)行原子操作。
- AtomicLongArray：用于對long數(shù)組中的元素進(jìn)行原子操作。
- AtomicReference：用于對引用類型的變量進(jìn)行原子操作。
- AtomicStampedReference：用于對引用類型的變量進(jìn)行原子操作，并能夠檢測變量是否被修改過。
- AtomicIntegerFieldUpdater：用于對某個對象中的int類型字段進(jìn)行原子操作。
- AtomicLongFieldUpdater：用于對某個對象中的long類型字段進(jìn)行原子操作。
- AtomicReferenceFieldUpdater：用于對某個對象中的引用類型字段進(jìn)行原子操作。
這些原子操作類都提供了一系列的方法，如get、set、addAndGet、compareAndSet等，可以實(shí)現(xiàn)對變量的原子操作。值得注意的是，使用原子操作類并不能解決所有的線程安全問題，需要根據(jù)具體情況進(jìn)行判斷和選擇。
（1）AtomicInteger
AtomicInteger用于對int類型的變量進(jìn)行原子操作。
import java.util.concurrent.atomic.AtomicInteger;
public class AtomicIntegerDemo {
private static AtomicInteger count = new AtomicInteger(0);
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(() -> {
for (int j = 0; j < 10000; j++) {
count.getAndIncrement();
}
}).start();
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Count: " + count.get());
}
}（2）AtomicLong
AtomicLong用于對long類型的變量進(jìn)行原子操作。
import java.util.concurrent.atomic.AtomicLong;
public class AtomicLongDemo {
private static AtomicLong count = new AtomicLong(0);
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(() -> {
for (int j = 0; j < 10000; j++) {
count.getAndIncrement();
}
}).start();
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Count: " + count.get());
}
}（3）AtomicBoolean
AtomicBoolean用于對boolean類型的變量進(jìn)行原子操作。
import java.util.concurrent.atomic.AtomicBoolean;
public class AtomicBooleanDemo {
private static AtomicBoolean flag = new AtomicBoolean(true);
public static void main(String[] args) {
new Thread(() -> {
while (flag.get()) {
System.out.println("Running...");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
flag.set(false);
System.out.println("Stopped.");
}
}（4）AtomicIntegerArray
AtomicIntegerArray用于對int數(shù)組中的元素進(jìn)行原子操作。
import java.util.concurrent.atomic.AtomicIntegerArray;
public class AtomicIntegerArrayDemo {
private static AtomicIntegerArray arr = new AtomicIntegerArray(new int[]{0, 0});
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(() -> {
for (int j = 0; j < 10000; j++) {
arr.getAndIncrement(j % 2);
}
}).start();
}
try {
Thread.sleep(1000);
} catch (InterruptedExceptione) {
e.printStackTrace();
}
System.out.println("Array: " + arr);
}
}（5）AtomicReference
AtomicReference用于對引用類型的變量進(jìn)行原子操作。
import java.util.concurrent.atomic.AtomicReference;
public class AtomicReferenceDemo {
static class Person {
String name;
int age;
public Person(String name, int age) {
this.name = name;
this.age = age;
}
@Override
public String toString() {
return "Person{name='" + name + "', age=" + age + "}";
}
}
private static AtomicReference<Person> personRef = new AtomicReference<>(new Person("Tom", 18));
public static void main(String[] args) {
Person oldPerson = personRef.get();
Person newPerson = new Person("Jerry", 20);
if (personRef.compareAndSet(oldPerson, newPerson)) {
System.out.println("Update success, old value: " + oldPerson + ", new value: " + newPerson);
} else {
System.out.println("Update failed.");
}
System.out.println("Person: " + personRef.get());
}
}3、同步隊(duì)列類
同步隊(duì)列類是一種特殊的隊(duì)列，它可以在多線程環(huán)境下實(shí)現(xiàn)數(shù)據(jù)的生產(chǎn)和消費(fèi)過程的同步。JUC包中提供了以下幾個同步隊(duì)列類：
- ArrayBlockingQueue：一個由數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列。
- LinkedBlockingQueue：一個由鏈表結(jié)構(gòu)組成的有界（但大小默認(rèn)值為Integer.MAX_VALUE）阻塞隊(duì)列。
- PriorityBlockingQueue：一個支持優(yōu)先級排序的無界阻塞隊(duì)列。
- SynchronousQueue：一個不存儲元素的阻塞隊(duì)列，每個插入操作必須等待另一個線程的移除操作，否則插入操作會一直阻塞。
這些同步隊(duì)列類提供了一系列的方法，如put、take、offer、poll等，可以實(shí)現(xiàn)對隊(duì)列的操作。同步隊(duì)列類還提供了一些擴(kuò)展方法，如drainTo、peek等。
同步隊(duì)列類的特點(diǎn)在于它們可以實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模式。多個線程可以同時往隊(duì)列中添加元素或者同時從隊(duì)列中取出元素，當(dāng)隊(duì)列為空或者已滿時，線程會被阻塞，直到有其他線程進(jìn)行相應(yīng)的操作。這種機(jī)制可以有效地控制線程間的同步和協(xié)作，避免了線程間的競爭和死鎖問題。
使用同步隊(duì)列類時需要注意以下幾點(diǎn)：
- 隊(duì)列大?。河捎谕疥?duì)列類是有界的，所以需要根據(jù)實(shí)際情況來設(shè)置隊(duì)列的大小。
- 隊(duì)列類型：不同的同步隊(duì)列類適用于不同的場景，需要根據(jù)具體情況進(jìn)行選擇。
（1）ArrayBlockingQueue
ArrayBlockingQueue是一個有界隊(duì)列，它的容量是固定的。當(dāng)隊(duì)列已滿時，添加元素的線程會被阻塞，直到有其他線程取出元素后才能繼續(xù)添加。
import java.util.concurrent.ArrayBlockingQueue;
public class ArrayBlockingQueueDemo {
private static ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
public static void main(String[] args) {
new Thread(() -> {
for (int i = 0; i < 100; i++) {
try {
queue.put(i);
System.out.println("Producer: " + i);
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
new Thread(() -> {
while (true) {
try {
Integer value = queue.take();
System.out.println("Consumer: " + value);
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}（2）LinkedBlockingQueue
LinkedBlockingQueue是一個無界隊(duì)列，它的容量是不限制的。當(dāng)隊(duì)列為空時，取出元素的線程會被阻塞，直到有其他線程添加元素后才能繼續(xù)取出。
import java.util.concurrent.LinkedBlockingQueue;
public class LinkedBlockingQueueDemo {
private static LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
public static void main(String[] args) {
new Thread(() -> {
for (int i = 0; i < 100; i++) {
try {
queue.put(i);
System.out.println("Producer: " + i);
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
new Thread(() -> {
while (true) {
try {
Integer value = queue.take();
System.out.println("Consumer: " + value);
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}（3）SynchronousQueue
SynchronousQueue是一個沒有緩沖的隊(duì)列，它的每個插入操作必須等待另一個線程執(zhí)行相應(yīng)的刪除操作，反之亦然。當(dāng)隊(duì)列中有一個元素時，插入操作會被阻塞，直到有其他線程取出元素后才能繼續(xù)插入。
import java.util.concurrent.SynchronousQueue;
public class SynchronousQueueDemo {
private static SynchronousQueue queue = new SynchronousQueue<>();
public static void main(String[] args) {
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
System.out.println("Producer: " + i);
queue.put(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
new Thread(() -> {
while (true) {
try {
Integer value = queue.take();
System.out.println("Consumer: " + value);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
}4、CountDownLatch類
CountDownLatch是一種同步工具類，它可以使一個或多個線程等待另一組線程完成操作后再繼續(xù)執(zhí)行。CountDownLatch的作用類似于“計(jì)數(shù)器”，在初始化時設(shè)置一個計(jì)數(shù)值，每當(dāng)一個線程完成任務(wù)后就將計(jì)數(shù)值減1，當(dāng)計(jì)數(shù)值變?yōu)?時，等待線程就會被喚醒。
CountDownLatch類提供了兩個主要方法：
- countDown：將計(jì)數(shù)值減1。
- await：等待計(jì)數(shù)值變?yōu)?。
使用CountDownLatch可以很方便地實(shí)現(xiàn)線程間的協(xié)作和同步，尤其適用于某些場景下需要等待多個線程都完成某項(xiàng)任務(wù)后才能進(jìn)行下一步操作的情況。
import java.util.concurrent.CountDownLatch;
public class CountDownLatchDemo {
private static CountDownLatch latch = new CountDownLatch(3);
public static void main(String[] args) {
new Thread(() -> {
try {
Thread.sleep(1000);
System.out.println("Thread A finished.");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(2000);
System.out.println("Thread B finished.");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(3000);
System.out.println("Thread C finished.");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
try {
latch.await();
System.out.println("All threads finished.");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}5、CyclicBarrier類
CyclicBarrier也是一種同步工具類，它可以讓一組線程在到達(dá)某個屏障點(diǎn)之前相互等待，然后同時執(zhí)行某個操作。CyclicBarrier的作用類似于“柵欄”，在初始化時設(shè)置一個屏障點(diǎn)，每當(dāng)一個線程到達(dá)屏障點(diǎn)時就會被阻塞，直到所有線程都到達(dá)屏障點(diǎn)后才會繼續(xù)執(zhí)行。
CyclicBarrier類提供了兩個主要方法：
- await：讓當(dāng)前線程到達(dá)屏障點(diǎn)，并等待其他線程到達(dá)。
- reset：重置屏障點(diǎn)的計(jì)數(shù)器。
使用CyclicBarrier可以很方便地實(shí)現(xiàn)一組線程的同步和協(xié)作，尤其適用于某些場景下需要多個線程同時開始執(zhí)行某項(xiàng)任務(wù)的情況。
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierDemo {
private static CyclicBarrier barrier = new CyclicBarrier(3, () -> {
System.out.println("All threads arrived at the barrier.");
});
public static void main(String[] args) {
new Thread(() -> {
try {
Thread.sleep(1000);
System.out.println("Thread A arrived at the barrier.");
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(2000);
System.out.println("Thread B arrived at the barrier.");
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(3000);
System.out.println("Thread C arrived at the barrier.");
barrier.await();
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}6、Semaphore類
信號量是一種經(jīng)典的并發(fā)編程工具，它可以用來限制同時訪問某個資源的線程數(shù)量。JUC包中提供了以下幾個信號量類：
- Semaphore：用于控制訪問某個共享資源的線程數(shù)量。
- CountingSemaphore：是Semaphore的一個變體，可以限制訪問某個共享資源的線程數(shù)量，并且支持語義上的“計(jì)數(shù)”。
- ReentrantLock：是一個可重入的互斥鎖，它可以對共享資源進(jìn)行訪問控制，從而保證多線程間對共享資源的安全訪問。
這些信號量類提供了一系列的方法，如acquire、release、tryAcquire等，可以實(shí)現(xiàn)對信號量的操作。使用信號量類可以有效地控制線程的并發(fā)訪問，從而避免競爭和死鎖問題。
Semaphore是一個同步工具類，用于控制對公共資源的訪問。它通過計(jì)數(shù)器來實(shí)現(xiàn)對資源的訪問控制，可以控制同時訪問某個資源的線程數(shù)量。
import java.util.concurrent.Semaphore;
public class SemaphoreDemo {
private static Semaphore semaphore = new Semaphore(2);
public static void main(String[] args) {
for (int i = 0; i < 5; i++) {
new Thread(() -> {
try {
semaphore.acquire();
System.out.println(Thread.currentThread().getName() + " acquired the semaphore.");
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName() + " released the semaphore.");
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
}7、Exchanger類
Exchanger是一種同步工具類，它可以使兩個線程之間交換數(shù)據(jù)。Exchanger的作用類似于“交換機(jī)”，兩個線程分別調(diào)用Exchanger對象的exchange方法，將各自持有的數(shù)據(jù)傳遞給對方，然后繼續(xù)執(zhí)行。
Exchanger類提供了一個exchange方法，可以實(shí)現(xiàn)兩個線程之間的數(shù)據(jù)交換。使用Exchanger可以很方便地實(shí)現(xiàn)數(shù)據(jù)在不同線程之間的傳遞和同步，尤其適用于某些場景下需要進(jìn)行線程間數(shù)據(jù)交互的情況。
import java.util.concurrent.Exchanger;
public class ExchangerDemo {
private static Exchanger<String> exchanger = new Exchanger<>();
public static void main(String[] args) {
new Thread(() -> {
try {
String data = "Hello World";
System.out.println("Thread A: before exchange, data = " + data);
data = exchanger.exchange(data);
System.out.println("Thread A: after exchange, data = " + data);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
String data = "123456789";
System.out.println("Thread B: before exchange, data = " + data);
data = exchanger.exchange(data);
System.out.println("Thread B: after exchange, data = " + data);
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}8、CompletableFuture類
CompletableFuture是Java8中新增的一個并發(fā)工具類，它可以以異步的方式執(zhí)行任務(wù)，并支持任務(wù)之間的組合和串聯(lián)操作。CompletableFuture類的主要特點(diǎn)包括：
- 異步執(zhí)行：可以在新的線程中異步執(zhí)行任務(wù)。
- 鏈?zhǔn)秸{(diào)用：支持任務(wù)之間的鏈?zhǔn)秸{(diào)用，從而實(shí)現(xiàn)多個任務(wù)的組合和串聯(lián)操作。
- 回調(diào)機(jī)制：可以通過回調(diào)機(jī)制來處理任務(wù)執(zhí)行的結(jié)果。
CompletableFuture類提供了一系列的方法，如supplyAsync、thenApply、thenAccept、thenCompose等，可以實(shí)現(xiàn)對任務(wù)的異步執(zhí)行、組合和串聯(lián)操作。使用CompletableFuture可以很方便地實(shí)現(xiàn)高效、簡潔的異步編程方式。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompletableFutureDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("Task 1 is running.");
return "Result 1";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
System.out.println("Task 2 is running.");
return "Result 2";
});
CompletableFuture<Integer> combinedFuture = future1.thenCombine(future2, (result1, result2) -> {
System.out.println("Task 3 is running.");
System.out.println("result1: " + result1);
System.out.println("result2: " + result2);
return result1.length() + result2.length();
});
System.out.println("Combined result: " + combinedFuture.get());
}
}9、Fork/Join框架
ForkJoin框架是JDK7中引入的一個并行計(jì)算框架，它可以將一個大型任務(wù)劃分為若干個小任務(wù)并行執(zhí)行，然后將各個小任務(wù)的結(jié)果匯總得到最終結(jié)果。ForkJoin框架的主要特點(diǎn)包括：
- 任務(wù)分解：可以將一個大型任務(wù)劃分為若干個小任務(wù)并行執(zhí)行。
- 工作竊取：每個線程都有自己的任務(wù)隊(duì)列，當(dāng)空閑時會“竊取”其他線程任務(wù)隊(duì)列中的任務(wù)進(jìn)行執(zhí)行，從而提高計(jì)算效率。
- 可擴(kuò)展性：可以根據(jù)實(shí)際情況動態(tài)增加或減少線程數(shù)。
ForkJoin框架通過ForkJoinPool類來管理線程池和任務(wù)調(diào)度。使用ForkJoin框架可以很方便地實(shí)現(xiàn)高效、簡潔的并行計(jì)算代碼。
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
public class ForkJoinDemo {
static class Fibonacci extends RecursiveTask<Integer> {
final int n;
Fibonacci(int n) {
this.n = n;
}
protected Integer compute() {
if (n <= 1)
return n;
Fibonacci f1 = new Fibonacci(n - 1);
f1.fork();
Fibonacci f2 = new Fibonacci(n - 2);
return f2.compute() + f1.join();
}
}
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool();
Fibonacci task = new Fibonacci(10);
int result = pool.invoke(task);
System.out.println(result);
}
}10、總結(jié)
Java并發(fā)編程是一門非常重要的技術(shù)，在面對大規(guī)模并發(fā)處理、高性能計(jì)算、分布式系統(tǒng)和云計(jì)算等領(lǐng)域時，它扮演著至關(guān)重要的角色。本文介紹了Java并發(fā)編程中常用的幾種并發(fā)工具類和框架，包括線程池、鎖、原子類、同步隊(duì)列、同步工具類、CompletableFuture和Fork/Join框架等，并提供了簡單的示例代碼，希望可以為讀者在實(shí)踐中應(yīng)用并發(fā)編程提供一些參考和啟示。
