处理大规模数据计算任务，Fork/Join框架是您的得力助手！

2023-05-06 今日头条你的老师父

加入收藏

1. JUC包概述

JUC是JAVA Util Concurrency的缩写，即Java并发工具包。JUC包提供了一些常用的线程和并发编程工具类，帮助开发者更方便地开发多线程应用程序，提高程序的并发性能。JUC包的主要特点包括：

安全性：JUC包提供了一些线程安全的数据结构和工具类，如原子类、同步队列等，可以保证多线程访问时数据的正确性和一致性。
性能：JUC包中的一些并发工具类采用了高效的算法和数据结构，如CAS算法、乐观锁等，可以提高程序的并发性能。
可扩展性：JUC包中的一些工具类支持可扩展性，如线程池、ForkJoin框架等，可以根据实际情况进行动态调整。

2. 原子操作类

在多线程环境下，由于多个线程同时访问同一个变量可能会导致数据不一致的问题。原子操作类可以保证在多线程环境下对变量的操作是原子性的，即不会出现线程安全问题。

JJUC包中提供了以下几个原子操作类：

AtomicInteger：用于对int类型的变量进行原子操作。
AtomicLong：用于对long类型的变量进行原子操作。
AtomicBoolean：用于对boolean类型的变量进行原子操作。
AtomicIntegerArray：用于对int数组中的元素进行原子操作。
AtomicLongArray：用于对long数组中的元素进行原子操作。
AtomicReference：用于对引用类型的变量进行原子操作。
AtomicStampedReference：用于对引用类型的变量进行原子操作，并能够检测变量是否被修改过。
AtomicIntegerFieldUpdater：用于对某个对象中的int类型字段进行原子操作。
AtomicLongFieldUpdater：用于对某个对象中的long类型字段进行原子操作。
AtomicReferenceFieldUpdater：用于对某个对象中的引用类型字段进行原子操作。

这些原子操作类都提供了一系列的方法，如get、set、addAndGet、compareAndSet等，可以实现对变量的原子操作。值得注意的是，使用原子操作类并不能解决所有的线程安全问题，需要根据具体情况进行判断和选择。

2.1 AtomicInteger

AtomicInteger用于对int类型的变量进行原子操作。

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.2 AtomicLong

AtomicLong用于对long类型的变量进行原子操作。

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());
    }
}

2.3 AtomicBoolean

AtomicBoolean用于对boolean类型的变量进行原子操作。

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.");
    }
}

2.4 AtomicIntegerArray

AtomicIntegerArray用于对int数组中的元素进行原子操作。

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);
    }
}

2.5 AtomicReference

AtomicReference用于对引用类型的变量进行原子操作。

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. 同步队列类

同步队列类是一种特殊的队列，它可以在多线程环境下实现数据的生产和消费过程的同步。JUC包中提供了以下几个同步队列类：

ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列。
LinkedBlockingQueue：一个由链表结构组成的有界（但大小默认值为Integer.MAX_VALUE）阻塞队列。
PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。
SynchronousQueue：一个不存储元素的阻塞队列，每个插入操作必须等待另一个线程的移除操作，否则插入操作会一直阻塞。

这些同步队列类提供了一系列的方法，如put、take、offer、poll等，可以实现对队列的操作。同步队列类还提供了一些扩展方法，如drainTo、peek等。

同步队列类的特点在于它们可以实现生产者-消费者模式。多个线程可以同时往队列中添加元素或者同时从队列中取出元素，当队列为空或者已满时，线程会被阻塞，直到有其他线程进行相应的操作。这种机制可以有效地控制线程间的同步和协作，避免了线程间的竞争和死锁问题。

使用同步队列类时需要注意以下几点：

队列大小：由于同步队列类是有界的，所以需要根据实际情况来设置队列的大小。
队列类型：不同的同步队列类适用于不同的场景，需要根据具体情况进行选择。

3.1 ArrayBlockingQueue

ArrayBlockingQueue是一个有界队列，它的容量是固定的。当队列已满时，添加元素的线程会被阻塞，直到有其他线程取出元素后才能继续添加。

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();
    }
}

3.2 LinkedBlockingQueue

LinkedBlockingQueue是一个无界队列，它的容量是不限制的。当队列为空时，取出元素的线程会被阻塞，直到有其他线程添加元素后才能继续取出。

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.3 SynchronousQueue

SynchronousQueue是一个没有缓冲的队列，它的每个插入操作必须等待另一个线程执行相应的删除操作，反之亦然。当队列中有一个元素时，插入操作会被阻塞，直到有其他线程取出元素后才能继续插入。

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是一种同步工具类，它可以使一个或多个线程等待另一组线程完成操作后再继续执行。CountDownLatch的作用类似于“计数器”，在初始化时设置一个计数值，每当一个线程完成任务后就将计数值减1，当计数值变为0时，等待线程就会被唤醒。

CountDownLatch类提供了两个主要方法：

countDown：将计数值减1。
await：等待计数值变为0。

使用CountDownLatch可以很方便地实现线程间的协作和同步，尤其适用于某些场景下需要等待多个线程都完成某项任务后才能进行下一步操作的情况。

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也是一种同步工具类，它可以让一组线程在到达某个屏障点之前相互等待，然后同时执行某个操作。CyclicBarrier的作用类似于“栅栏”，在初始化时设置一个屏障点，每当一个线程到达屏障点时就会被阻塞，直到所有线程都到达屏障点后才会继续执行。

CyclicBarrier类提供了两个主要方法：

await：让当前线程到达屏障点，并等待其他线程到达。
reset：重置屏障点的计数器。

使用CyclicBarrier可以很方便地实现一组线程的同步和协作，尤其适用于某些场景下需要多个线程同时开始执行某项任务的情况。

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类

信号量是一种经典的并发编程工具，它可以用来限制同时访问某个资源的线程数量。JUC包中提供了以下几个信号量类：

Semaphore：用于控制访问某个共享资源的线程数量。
CountingSemaphore：是Semaphore的一个变体，可以限制访问某个共享资源的线程数量，并且支持语义上的“计数”。
ReentrantLock：是一个可重入的互斥锁，它可以对共享资源进行访问控制，从而保证多线程间对共享资源的安全访问。

这些信号量类提供了一系列的方法，如acquire、release、tryAcquire等，可以实现对信号量的操作。使用信号量类可以有效地控制线程的并发访问，从而避免竞争和死锁问题。

Semaphore是一个同步工具类，用于控制对公共资源的访问。它通过计数器来实现对资源的访问控制，可以控制同时访问某个资源的线程数量。

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是一种同步工具类，它可以使两个线程之间交换数据。Exchanger的作用类似于“交换机”，两个线程分别调用Exchanger对象的exchange方法，将各自持有的数据传递给对方，然后继续执行。

Exchanger类提供了一个exchange方法，可以实现两个线程之间的数据交换。使用Exchanger可以很方便地实现数据在不同线程之间的传递和同步，尤其适用于某些场景下需要进行线程间数据交互的情况。

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中新增的一个并发工具类，它可以以异步的方式执行任务，并支持任务之间的组合和串联操作。CompletableFuture类的主要特点包括：

异步执行：可以在新的线程中异步执行任务。
链式调用：支持任务之间的链式调用，从而实现多个任务的组合和串联操作。
回调机制：可以通过回调机制来处理任务执行的结果。

CompletableFuture类提供了一系列的方法，如supplyAsync、thenApply、thenAccept、thenCompose等，可以实现对任务的异步执行、组合和串联操作。使用CompletableFuture可以很方便地实现高效、简洁的异步编程方式。

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中引入的一个并行计算框架，它可以将一个大型任务划分为若干个小任务并行执行，然后将各个小任务的结果汇总得到最终结果。ForkJoin框架的主要特点包括：

任务分解：可以将一个大型任务划分为若干个小任务并行执行。
工作窃取：每个线程都有自己的任务队列，当空闲时会“窃取”其他线程任务队列中的任务进行执行，从而提高计算效率。
可扩展性：可以根据实际情况动态增加或减少线程数。

ForkJoin框架通过ForkJoinPool类来管理线程池和任务调度。使用ForkJoin框架可以很方便地实现高效、简洁的并行计算代码。

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. 总结

Java并发编程是一门非常重要的技术，在面对大规模并发处理、高性能计算、分布式系统和云计算等领域时，它扮演着至关重要的角色。本文介绍了Java并发编程中常用的几种并发工具类和框架，包括线程池、锁、原子类、同步队列、同步工具类、CompletableFuture和Fork/Join框架等，并提供了简单的示例代码，希望可以为读者在实践中应用并发编程提供一些参考和启示。