AtomicLong 的 incrementAndGet 方法阻塞调用？

Question

我正在研究Multithreaded code，我正在尝试测量一个特定方法所花费的时间，因为我正在尝试对我们的大多数队友代码进行基准测试，就像我正在做Load and Performance 测试我们的Client code 和然后是我们的Service code。

所以对于这个性能测量，我正在使用-

System.nanoTime();

我有多线程代码，我从中生成多个线程并尝试测量该代码花费了多少时间。

下面是我试图测量任何代码性能的示例示例-在下面的代码中我试图测量-

beClient.getAttributes method

下面是代码-

public class BenchMarkTest {

    public static void main(String[] args) {

        ExecutorService executor = Executors.newFixedThreadPool(5);

        try {

            for (int i = 0; i < 3 * 5; i++) {
                executor.submit(new ThreadTask(i));
            }

            executor.shutdown();
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        } catch (InterruptedException e) {

        }
    }

}

下面是实现Runnable接口的类

class ThreadTask implements Runnable {
    private int id;
    public static ConcurrentHashMap<Long, AtomicLong> selectHistogram = new ConcurrentHashMap<Long, AtomicLong>();


    public ThreadTask(int id) {
        this.id = id;
    }

    @Override
    public void run() {


        long start = System.nanoTime();

        attributes = beClient.getAttributes(columnsList);

        long end = System.nanoTime() - start;

        final AtomicLong before = selectHistogram.putIfAbsent(end / 1000000L, new AtomicLong(1L));
        if (before != null) {
            before.incrementAndGet();
        }
    }
}

无论我想测量什么代码，我通常将下面的行放在该方法的上方

long start = System.nanoTime();

这两行在相同的方法之后，但ConcurrentHashMap不同

long end = System.nanoTime() - start;

final AtomicLong before = selectHistogram.putIfAbsent(end / 1000000L, new AtomicLong(1L));
        if (before != null) {
            before.incrementAndGet();
        }

今天我和我的一位资深人士开会，他说ConcurrentHashMap 的incrementAndGet 方法是一个阻塞调用。所以你的线程会在那里等待一段时间。

他让我做那个Asynchronous call。

是否有可能进行异步调用？

因为在我们所有的客户端代码和服务代码中来衡量每个方法的性能，我使用上面相同的三行，我通常在每个方法之前和之后放置来衡量这些方法的性能。程序完成后，我将这些地图的结果打印出来。

所以现在我正在考虑制作Asynchronous call？谁能帮我做这件事？

基本上，我正在尝试以异步方式测量特定方法的性能，以便每个线程都不会等待并被阻塞。

我想，我可以使用Futures 来做到这一点。谁能提供一个相关的例子？

感谢您的帮助。

Answer 1

行：

if (before != null) {
    before.incrementAndGet();
}

会锁定当前线程直到before.incrementAndGet()获得锁（如果一定要知道，其实没有锁，有while(true)和compare-and-swap方法）并返回long值（即你是不使用）。

您可以通过在它自己的线程中调用该特定方法来使其异步，从而不会阻塞当前线程。

要做到这一点，我相信你已经知道如何做到这一点：使用 Thread.start()、ExecutorService 或 FutureTask（查看“How to asynchronously call a method in Java”，了解如何以优雅的方式进行操作）。

如果我不清楚，这是使用FutureTask的解决方案：

public class BenchMarkTest {

    public static void main(String[] args) {

        ExecutorService executor = Executors.newFixedThreadPool(5);

        int threadNum = 2;
        ExecutorService taskExecutor = Executors.newFixedThreadPool(threadNum);
        List<FutureTask<Long>> taskList = new ArrayList<FutureTask<Long>>();

        try {

            for (int i = 0; i < 3 * 5; i++) {
                executor.submit(new ThreadTask(i, taskExecutor, taskList));
            }

            executor.shutdown();
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        } catch (InterruptedException e) {

        }

        for (FutureTask<Long> futureTask : taskList) {
            futureTask.get(); // doing a job similar to joining threads
        }
        taskExecutor.shutdown();
    }

}

ThreadTask类：

class ThreadTask implements Runnable {
    private int id;
    public static ConcurrentHashMap<Long, AtomicLong> selectHistogram = new ConcurrentHashMap<Long, AtomicLong>();

    private ExecutorService taskExecutor;
    private List<FutureTask<Long>> taskList;    

    public ThreadTask(int id, ExecutorService taskExecutor, List<FutureTask<Long>> taskList) {
        this.id = id;
        this.taskExecutor = taskExecutor;
        this.taskList = taskList;
    }

    @Override
    public void run() {


        long start = System.nanoTime();

        attributes = beClient.getAttributes(columnsList);

        long end = System.nanoTime() - start;

        final AtomicLong before = selectHistogram.putIfAbsent(end / 1000000L, new AtomicLong(1L));
        if (before != null) {
            FutureTask<Long> futureTask = new FutureTask<Long>(new Callable<Long>() {
                public Long call() {
                    return before.incrementAndGet();
                }
            });
            taskList.add(futureTask);
            taskExecutor.execute(futureTask);
        }
    }
}

更新：

我想到了一点可能的改进：与其在ThreadTask 类中告诉taskExecutor 执行futureTask，不如将​​任务的执行推迟到main 方法的末尾.我的意思是：

去掉ThreadTask.run()下面的那一行：

            taskExecutor.execute(futureTask);

并且，在main() 方法中，您有：

        for (FutureTask<Long> futureTask : taskList) {
            futureTask.get(); // doing a job similar to joining threads
        }
        taskExecutor.shutdown();

添加任务的执行，从而有：

        taskExecutor.invokeAll(taskList);
        for (FutureTask<Long> futureTask : taskList) {
            futureTask.get(); // doing a job similar to joining threads
        }
        taskExecutor.shutdown();

（另外，您可以删除ThreadTask 的ExecutorService 字段，因为它将不再使用它。）

这样，在执行基准测试时开销很小（开销是将对象添加到taskList，仅此而已）。

完整更新的代码：

public class BenchMarkTest {

    public static void main(String[] args) {

        ExecutorService executor = Executors.newFixedThreadPool(5);

        List<FutureTask<Long>> taskList = new ArrayList<FutureTask<Long>>();

        try {

            for (int i = 0; i < 3 * 5; i++) {
                executor.submit(new ThreadTask(i, taskList));
            }

            executor.shutdown();
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        } catch (InterruptedException e) {

        }

        int threadNum = 2;
        ExecutorService taskExecutor = Executors.newFixedThreadPool(threadNum);
        taskExecutor.invokeAll(taskList);
        for (FutureTask<Long> futureTask : taskList) {
            futureTask.get(); // doing a job similar to joining threads
        }
        taskExecutor.shutdown();
    }

}

-

class ThreadTask implements Runnable {
    private int id;
    public static ConcurrentHashMap<Long, AtomicLong> selectHistogram = new ConcurrentHashMap<Long, AtomicLong>();

    private List<FutureTask<Long>> taskList;    

    public ThreadTask(int id, List<FutureTask<Long>> taskList) {
        this.id = id;
        this.taskList = taskList;
    }

    @Override
    public void run() {


        long start = System.nanoTime();

        attributes = beClient.getAttributes(columnsList);

        long end = System.nanoTime() - start;

        final AtomicLong before = selectHistogram.putIfAbsent(end / 1000000L, new AtomicLong(1L));
        if (before != null) {
            FutureTask<Long> futureTask = new FutureTask<Long>(new Callable<Long>() {
                public Long call() {
                    return before.incrementAndGet();
                }
            });
            taskList.add(futureTask);
        }
    }
}

Answer 2

我确信递增AtomicLong 比创建Runnable/Callable 对象并将其传递给ExecutorService 所需的时间更少。这会导致你真的成为瓶颈吗？如果您真的想要快速并发增量，请参阅 JDK8 中的 LongAdder。

LongAdders 可以与 ConcurrentHashMap 一起使用，以维护可扩展的频率图（直方图或多重集的一种形式）。例如，要向 ConcurrentHashMap freqs 添加计数，如果不存在则进行初始化，您可以使用 freqs.computeIfAbsent(k -> new LongAdder()).increment();