處理大規(guī)模數(shù)據(jù)計(jì)算任務(wù)，F(xiàn)ork/Join框架是您的得力助手！

作者：你的老師父 2023-05-31 08:37:06

本文介紹Java并發(fā)編程中常用的幾種并發(fā)工具類和框架，包括線程池、鎖、原子類、同步隊(duì)列、同步工具類、CompletableFuture和Fork/Join框架等，并提供了簡單的示例代碼，希望可以為讀者在實(shí)踐中應(yīng)用并發(fā)編程提供一些參考和啟示。

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í)數(shù)據(jù)的正確性和一致性。
性能：JUC包中的一些并發(fā)工具類采用了高效的算法和數(shù)據(jù)結(jié)構(gòu)，如CAS算法、樂觀鎖等，可以提高程序的并發(fā)性能。
可擴(kuò)展性：JUC包中的一些工具類支持可擴(kuò)展性，如線程池、ForkJoin框架等，可以根據(jù)實(shí)際情況進(jìn)行動(dòng)態(tài)調(diào)整。

2、原子操作類

在多線程環(huán)境下，由于多個(gè)線程同時(shí)訪問同一個(gè)變量可能會(huì)導(dǎo)致數(shù)據(jù)不一致的問題。原子操作類可以保證在多線程環(huán)境下對變量的操作是原子性的，即不會(huì)出現(xiàn)線程安全問題。

JJUC包中提供了以下幾個(gè)原子操作類：

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：用于對某個(gè)對象中的int類型字段進(jìn)行原子操作。
AtomicLongFieldUpdater：用于對某個(gè)對象中的long類型字段進(jìn)行原子操作。
AtomicReferenceFieldUpdater：用于對某個(gè)對象中的引用類型字段進(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包中提供了以下幾個(gè)同步隊(duì)列類：

ArrayBlockingQueue：一個(gè)由數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列。
LinkedBlockingQueue：一個(gè)由鏈表結(jié)構(gòu)組成的有界（但大小默認(rèn)值為Integer.MAX_VALUE）阻塞隊(duì)列。
PriorityBlockingQueue：一個(gè)支持優(yōu)先級排序的無界阻塞隊(duì)列。
SynchronousQueue：一個(gè)不存儲元素的阻塞隊(duì)列，每個(gè)插入操作必須等待另一個(gè)線程的移除操作，否則插入操作會(huì)一直阻塞。

這些同步隊(duì)列類提供了一系列的方法，如put、take、offer、poll等，可以實(shí)現(xiàn)對隊(duì)列的操作。同步隊(duì)列類還提供了一些擴(kuò)展方法，如drainTo、peek等。

同步隊(duì)列類的特點(diǎn)在于它們可以實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模式。多個(gè)線程可以同時(shí)往隊(duì)列中添加元素或者同時(shí)從隊(duì)列中取出元素，當(dāng)隊(duì)列為空或者已滿時(shí)，線程會(huì)被阻塞，直到有其他線程進(jìn)行相應(yīng)的操作。這種機(jī)制可以有效地控制線程間的同步和協(xié)作，避免了線程間的競爭和死鎖問題。

使用同步隊(duì)列類時(shí)需要注意以下幾點(diǎn)：

隊(duì)列大小：由于同步隊(duì)列類是有界的，所以需要根據(jù)實(shí)際情況來設(shè)置隊(duì)列的大小。
隊(duì)列類型：不同的同步隊(duì)列類適用于不同的場景，需要根據(jù)具體情況進(jìn)行選擇。

（1）ArrayBlockingQueue

ArrayBlockingQueue是一個(gè)有界隊(duì)列，它的容量是固定的。當(dāng)隊(duì)列已滿時(shí)，添加元素的線程會(huì)被阻塞，直到有其他線程取出元素后才能繼續(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是一個(gè)無界隊(duì)列，它的容量是不限制的。當(dāng)隊(duì)列為空時(shí)，取出元素的線程會(huì)被阻塞，直到有其他線程添加元素后才能繼續(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是一個(gè)沒有緩沖的隊(duì)列，它的每個(gè)插入操作必須等待另一個(gè)線程執(zhí)行相應(yīng)的刪除操作，反之亦然。當(dāng)隊(duì)列中有一個(gè)元素時(shí)，插入操作會(huì)被阻塞，直到有其他線程取出元素后才能繼續(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是一種同步工具類，它可以使一個(gè)或多個(gè)線程等待另一組線程完成操作后再繼續(xù)執(zhí)行。CountDownLatch的作用類似于“計(jì)數(shù)器”，在初始化時(shí)設(shè)置一個(gè)計(jì)數(shù)值，每當(dāng)一個(gè)線程完成任務(wù)后就將計(jì)數(shù)值減1，當(dāng)計(jì)數(shù)值變?yōu)?時(shí)，等待線程就會(huì)被喚醒。

CountDownLatch類提供了兩個(gè)主要方法：

countDown：將計(jì)數(shù)值減1。
await：等待計(jì)數(shù)值變?yōu)?。

使用CountDownLatch可以很方便地實(shí)現(xiàn)線程間的協(xié)作和同步，尤其適用于某些場景下需要等待多個(gè)線程都完成某項(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á)某個(gè)屏障點(diǎn)之前相互等待，然后同時(shí)執(zhí)行某個(gè)操作。CyclicBarrier的作用類似于“柵欄”，在初始化時(shí)設(shè)置一個(gè)屏障點(diǎn)，每當(dāng)一個(gè)線程到達(dá)屏障點(diǎn)時(shí)就會(huì)被阻塞，直到所有線程都到達(dá)屏障點(diǎn)后才會(huì)繼續(xù)執(zhí)行。

CyclicBarrier類提供了兩個(gè)主要方法：

await：讓當(dāng)前線程到達(dá)屏障點(diǎn)，并等待其他線程到達(dá)。
reset：重置屏障點(diǎn)的計(jì)數(shù)器。

使用CyclicBarrier可以很方便地實(shí)現(xiàn)一組線程的同步和協(xié)作，尤其適用于某些場景下需要多個(gè)線程同時(shí)開始執(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í)訪問某個(gè)資源的線程數(shù)量。JUC包中提供了以下幾個(gè)信號量類：

Semaphore：用于控制訪問某個(gè)共享資源的線程數(shù)量。
CountingSemaphore：是Semaphore的一個(gè)變體，可以限制訪問某個(gè)共享資源的線程數(shù)量，并且支持語義上的“計(jì)數(shù)”。
ReentrantLock：是一個(gè)可重入的互斥鎖，它可以對共享資源進(jìn)行訪問控制，從而保證多線程間對共享資源的安全訪問。

這些信號量類提供了一系列的方法，如acquire、release、tryAcquire等，可以實(shí)現(xiàn)對信號量的操作。使用信號量類可以有效地控制線程的并發(fā)訪問，從而避免競爭和死鎖問題。

Semaphore是一個(gè)同步工具類，用于控制對公共資源的訪問。它通過計(jì)數(shù)器來實(shí)現(xiàn)對資源的訪問控制，可以控制同時(shí)訪問某個(gè)資源的線程數(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是一種同步工具類，它可以使兩個(gè)線程之間交換數(shù)據(jù)。Exchanger的作用類似于“交換機(jī)”，兩個(gè)線程分別調(diào)用Exchanger對象的exchange方法，將各自持有的數(shù)據(jù)傳遞給對方，然后繼續(xù)執(zhí)行。

Exchanger類提供了一個(gè)exchange方法，可以實(shí)現(xiàn)兩個(gè)線程之間的數(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中新增的一個(gè)并發(fā)工具類，它可以以異步的方式執(zhí)行任務(wù)，并支持任務(wù)之間的組合和串聯(lián)操作。CompletableFuture類的主要特點(diǎn)包括：

異步執(zhí)行：可以在新的線程中異步執(zhí)行任務(wù)。
鏈?zhǔn)秸{(diào)用：支持任務(wù)之間的鏈?zhǔn)秸{(diào)用，從而實(shí)現(xiàn)多個(gè)任務(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中引入的一個(gè)并行計(jì)算框架，它可以將一個(gè)大型任務(wù)劃分為若干個(gè)小任務(wù)并行執(zhí)行，然后將各個(gè)小任務(wù)的結(jié)果匯總得到最終結(jié)果。ForkJoin框架的主要特點(diǎn)包括：

任務(wù)分解：可以將一個(gè)大型任務(wù)劃分為若干個(gè)小任務(wù)并行執(zhí)行。
工作竊取：每個(gè)線程都有自己的任務(wù)隊(duì)列，當(dāng)空閑時(shí)會(huì)“竊取”其他線程任務(wù)隊(duì)列中的任務(wù)進(jìn)行執(zhí)行，從而提高計(jì)算效率。
可擴(kuò)展性：可以根據(jù)實(shí)際情況動(dòng)態(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)域時(shí)，它扮演著至關(guān)重要的角色。本文介紹了Java并發(fā)編程中常用的幾種并發(fā)工具類和框架，包括線程池、鎖、原子類、同步隊(duì)列、同步工具類、CompletableFuture和Fork/Join框架等，并提供了簡單的示例代碼，希望可以為讀者在實(shí)踐中應(yīng)用并發(fā)編程提供一些參考和啟示。

責(zé)任編輯：姜華來源：今日頭條