Java多线程下的其他组件之CyclicBarrier、Callable、Future和FutureTask详解

Java多线程下的其他组件之CyclicBarrier

CyclicBarrier概述

CyclicBarrier是Java中一个同步工具类，用于在多线程中等待所有线程到达某个同步点，然后再一起执行后续操作，这个同步点就是所谓的屏障（barrier），它可重用，即当到达屏障的线程数量达到指定值时，所有线程都可以通过屏障，继续执行下一个操作。

CyclicBarrier同样可以通过计数器（CountDownLatch）实现线程同步的作用，其主要区别在于，CyclicBarrier能够重复使用，而CountDownLatch则无法重复使用。

CyclicBarrier原理

CyclicBarrier的主要原理是利用一个计数器维护需要等待的线程数量，当线程到达屏障时，都会进行计数-1的操作，当计数器为0时，表示所有参与线程都已到达屏障，屏障就会打开，线程继续执行下一个操作。

CyclicBarrier具备以下特点：

• 可以重复使用
• 可以在构造时指定一个barrierAction，在屏障打开前最后一个到达的线程执行barrierAction操作
• 可以在到达屏障时等待一定时间，如果时间到了还没有达到指定线程数，就直接继续执行下一步操作

CyclicBarrier使用示例

下面的例子展示了如何使用CyclicBarrier实现多个线程同时对数组进行求和操作。

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

public class CyclicBarrierDemo {
    private static final int THREAD_COUNT = 5;
    private static final int ARRAY_SIZE = 100;
    private static final int SUM_PER_THREAD = ARRAY_SIZE / THREAD_COUNT;
    private static final int[] ARRAY = new int[ARRAY_SIZE];
    private static int result = 0;

    public static void main(String[] args) {
        //初始化数组
        for (int i = 0; i < ARRAY_SIZE; i++) {
            ARRAY[i] = i + 1;
        }
        //创建CyclicBarrier实例
        CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT, () -> {
            //所有线程到达屏障后，执行求和操作
            System.out.println("结果：" + result);
        });

        //创建多个线程
        for (int i = 0; i < THREAD_COUNT; i++) {
            new Thread(new SumTask(i * SUM_PER_THREAD, (i + 1) * SUM_PER_THREAD, barrier)).start();
        }
    }

    static class SumTask implements Runnable {
        private final int startIndex;
        private final int endIndex;
        private final CyclicBarrier barrier;

        public SumTask(int startIndex, int endIndex, CyclicBarrier barrier) {
            this.startIndex = startIndex;
            this.endIndex = endIndex;
            this.barrier = barrier;
        }

        @Override
        public void run() {
            int sum = 0;
            //对相应的数组元素进行求和
            for (int i = startIndex; i < endIndex; i++) {
                sum += ARRAY[i];
            }
            System.out.println("线程" + Thread.currentThread().getId() + "求和结果：" + sum);
            result += sum;
            //等待其他线程
            try {
                barrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
}

在这个例子中，我们将数组分成了5个部分，创建了5个线程，每个线程对应一个区间，每个线程对自己区间内的数组元素进行求和，最后由主线程输出所有线程求和的结果。

Java多线程下的其他组件之Callable和Future

Callable和Future概述

Callable和Future是Java中用于实现带返回值的多线程程序的两个接口，其中Callable代表一个有返回值的任务，而Future则表示异步计算的结果。

Callable和Future的主要特点如下：

• Callable接口实现了Runnable接口，其call()方法具备返回值
• Future接口主要提供了以下三个方法：
• boolean isDone()：判断异步计算是否完成
• V get()：获取异步计算的结果，如果计算未完成，该方法会阻塞当前线程，直到计算完成
• V get(long timeout, TimeUnit unit)：在指定的时间内获取异步计算的结果，如果计算未完成，超时就会抛出TimeoutException异常

Callable和Future使用示例

下面的例子展示了如何使用Callable和Future实现多个线程同时求和操作，并输出结果。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class CallableDemo {
    private static final int THREAD_COUNT = 5;
    private static final int ARRAY_SIZE = 100;
    private static final int SUM_PER_THREAD = ARRAY_SIZE / THREAD_COUNT;
    private static final int[] ARRAY = new int[ARRAY_SIZE];

    public static void main(String[] args) {
        // 初始化数组
        for (int i = 0; i < ARRAY_SIZE; i++) {
            ARRAY[i] = i + 1;
        }
        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
        List<Future<Integer>> resultList = new ArrayList<>(THREAD_COUNT);
        // 分配任务，调用submit方法提交任务
        for (int i = 0; i < THREAD_COUNT; i++) {
            int startIndex = i * SUM_PER_THREAD;
            int endIndex = (i + 1) * SUM_PER_THREAD;
            SumTask task = new SumTask(startIndex, endIndex);
            Future<Integer> future = executor.submit(task);
            resultList.add(future);
        }
        // 输出结果
        int result = 0;
        for (Future<Integer> future : resultList) {
            try {
                result += future.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
        System.out.println("结果：" + result);
        // 关闭线程池
        executor.shutdown();
    }

    static class SumTask implements Callable<Integer> {
        private final int startIndex;
        private final int endIndex;

        public SumTask(int startIndex, int endIndex) {
            this.startIndex = startIndex;
            this.endIndex = endIndex;
        }

        @Override
        public Integer call() {
            int sum = 0;
            // 对相应的数组元素进行求和
            for (int i = startIndex; i < endIndex; i++) {
                sum += ARRAY[i];
            }
            System.out.println("线程" + Thread.currentThread().getId() + "求和结果：" + sum);
            return sum;
        }
    }
}

在这个例子中，我们将数组分成了5个部分，创建了5个Callable任务，每个任务对自己区间内的数组元素进行求和，将结果返回，主线程对所有计算结果求和，得到最终结果输出。