同步并发请求以共享慢速操作的结果 [英] Synchronise concurrent requests to share results of a slow operation

问题描述

我有一个Java UI服务，它有一个API方法，可以调用相对较慢的操作（比如〜30secs）。该操作是无参数的，但它对随时间变化（相对缓慢）的外部数据进行操作。这种方法返回最新结果并不重要 - 如果它们是30秒，那是可以接受的。

I have a Java UI service that has an API method that invokes an operation that's relatively slow (say ~30secs). The operation is parameterless, but it operates on external data that does change (relatively slowly) over time. It's not critical for the method to return the most up-to-date results - if they're 30secs old that's acceptable.

最终我需要优化实施慢操作，但作为一个短期修复，我想使操作互斥，这样如果第二个传入请求（在一个单独的线程上）尝试调用该操作而另一个已在进行中，那么第二个一个块直到第一个完成。然后第二个线程使用第一次调用操作的结果 - 即它不会再次尝试运行操作。

Ultimately I need to optimise the implementation of the slow operation, but as a short-term fix, I'd like to make the operation mutually exclusive, such that if a second incoming request (on a separate thread) attempts to invoke the operation while another is already in progress, then the second one blocks until the first one completes. The second thread then uses the results of the first invocation of the operation - i.e. it doesn't attempt to run the operation again.

例如：

class MyService {
    String serviceApiMmethod() {
       // If a second thread attempts to call this method while another is in progress
       // then block here until the first returns and then use those results
       // (allowing it to return immediately without a second call to callSlowOperation).
       return callSlowOperation();
    }
}

Java中首选的一般方法是什么（8 ）。我猜我可以使用CountDownLatch，但目前尚不清楚如何最好地在线程之间共享结果。是否有一个现有的并发原语可以促进这个？

What's the preferred general approach for this in Java (8). I'm guessing I could use a CountDownLatch, but it's not clear how best to share the result across the threads. Is there an existing concurrency primitive that facilitates this?

编辑：我需要清除所有线程消耗后对结果的任何引用（即将其返回给调用者），因为它是一个相对较大的对象，需要尽快进行GC。

I need to clear any references to the result once all threads have consumed it (i.e. returned it to the caller), as it's relatively large object, which needs to be GC'ed as soon as possible.

推荐答案

简单的想法

版本1：

class Foo {
    public String foo() throws Exception {
        synchronized (this) {
            if (counter.incrementAndGet() == 1) {
                future = CompletableFuture.supplyAsync(() -> {
                    try {
                        Thread.sleep(1000 * (ThreadLocalRandom.current().nextInt(3) + 1));
                    } catch (InterruptedException e) {
                    }
                    return "ok" + ThreadLocalRandom.current().nextInt();
                });
            }
        }

        String result = future.get();
        if (counter.decrementAndGet() == 0) {
            future = null;
        }

        return result;
    }

    private AtomicInteger counter = new AtomicInteger();
    private Future<String> future;
}

版本2：与@AleksandrSemyannikov一起

Version 2: together with @AleksandrSemyannikov

public class MyService {
    private AtomicInteger counter = new AtomicInteger();
    private volatile String result;

    public String serviceApiMethod() {
        counter.incrementAndGet();
        try {
            synchronized (this) {
                if (result == null) {
                    result = callSlowOperation();
                }
            }
            return result;
        } finally {
            if (counter.decrementAndGet() == 0) {
                synchronized (this) {
                    if (counter.get() == 0) {
                        result = null;
                    }
                }
            }
        }
    }

    private String callSlowOperation() {
        try {
            Thread.sleep(ThreadLocalRandom.current().nextInt(1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return Thread.currentThread().getName();
    }
}

这篇关于同步并发请求以共享慢速操作的结果的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！