Безопасный и эффективный способ ожидания асинхронной задачи

Ответ 1

Я считаю, что замена mutex на CountDownLatch в подходе waitingRoom предотвращает взаимоблокировку.

CountDownLatch latch = new CountDownLatch(1)
taskProcessor.addToWaitingRoom(uniqueIdentifier, latch)
while (!checkResultIsInDatabase())
  // consider timed version
  latch.await()

//TaskProcessor
... Some complicated calculations
if (uniqueIdentifierExistInWaitingRoom(taskUniqueIdentifier))
  getLatchFromWaitingRoom(taskUniqueIdentifier).countDown()

Ответ 2

Кажется, вы ищете какую-то абстракцию будущего/обещания. Взгляните на CompletableFuture, доступный с Java 8.

CompletableFuture<Void> future = CompletableFuture.runAsync(db::yourExpensiveOperation, executor);

// best approach: attach some callback to run when the future is complete, and handle any errors
future.thenRun(this::onSuccess)
        .exceptionally(ex -> logger.error("err", ex));

// if you really need the current thread to block, waiting for the async result:
future.join(); // blocking! returns the result when complete or throws a CompletionException on error

Вы также можете вернуть (значимое) значение из асинхронной операции и передать результат в обратный вызов. Чтобы использовать это, взгляните на supplyAsync(), thenAccept(), thenApply(), whenComplete() и тому подобное.

Вы также можете объединить несколько фьючерсов в одно и многое другое.

Ответ 3

С CompletableFuture и ConcurrentHashMap вы можете достичь этого:

/* Server class, i.e. your TaskProcessor */
// Map of queued tasks (either pending or ongoing)
private static final ConcurrentHashMap<String, CompletableFuture<YourTaskResult>> tasks = new ConcurrentHashMap<>();

// Launch method. By default, CompletableFuture uses ForkJoinPool which implicitly enqueues tasks.
private CompletableFuture<YourTaskResult> launchTask(final String taskId) {
    return tasks.computeIfAbsent(taskId, v -> CompletableFuture // return ongoing task if any, or launch a new one
            .supplyAsync(() -> 
                    doYourThing(taskId)) // get from DB or calculate or whatever
            .whenCompleteAsync((integer, throwable) -> {
                if (throwable != null) {
                    log.error("Failed task: {}", taskId, throwable);
                }
                tasks.remove(taskId);
            })
    );


/* Client class, i.e. your UserThread */
// Usage
YourTaskResult taskResult = taskProcessor.launchTask(taskId).get(); // block until we get a result

Каждый раз, когда пользователь запрашивает результат taskId, он либо:

• ставить новые задачи в очередь, если они первыми taskId этот taskId; или же
• получить результат текущей задачи с идентификатором taskId, если кто-то еще поставил его в очередь.

Это производственный код, который в настоящее время используется сотнями пользователей одновременно.
В нашем приложении пользователи запрашивают любой данный файл через конечную точку REST (каждый пользователь имеет свой собственный поток). Наш taskId - это имена файлов, а наш doYourThing(taskId) извлекает файл из локальной файловой системы или загружает его из doYourThing(taskId) S3, если он не существует.
Очевидно, что мы не хотим загружать один и тот же файл более одного раза. С помощью этого решения, которое я реализовал, любое количество пользователей может запрашивать один и тот же файл в одно и то же или в разное время, и файл будет загружен ровно один раз. Все пользователи, которые запросили его во время загрузки, получат его одновременно с моментом окончания загрузки; все пользователи, которые просят об этом позже, немедленно получат его из локальной файловой системы.

Работает как шарм.

Ответ 4

Что я понял из деталей вопроса is-

Когда UserThread запрашивает результат, есть 3 возможности:

1. Задача уже выполнена, поэтому нет блокировки пользовательского потока и напрямую получен результат из БД.
2. Задача находится в очереди или выполняется, но еще не завершена, поэтому заблокируйте пользовательский поток (до сих пор не должно быть никаких запросов к базе данных) и сразу после завершения задачи (результат задачи должен быть сохранен в БД на этом этапе), разблокируйте пользовательский поток (теперь пользовательский поток может запросить у БД результат)
3. Для заданного uniqueIdentifier, который запрашивал пользователь, не было выполнено ни одной задачи, в этом случае будет пустой результат из db.

Для пунктов 1 и 3, это прямо, не будет никаких блокировок UserThread, просто запросите результат из БД.

Для пункта 2 - я написал простую реализацию TaskProcessor. Здесь я использовал ConcurrentHashMap чтобы сохранить текущие задачи, которые еще не выполнены. Эта карта содержит отображение между UniqueIdentifier и соответствующей задачей. Я использовал метод ConcurrentHashMap computeIfPresent() (представлен в JAVA - 1.8), который гарантирует, что вызов этого метода является потокобезопасным для того же ключа. Вот что говорит Java-документ: Ссылка

Если значение для указанного ключа присутствует, попытка вычислить новое сопоставление, учитывая ключ и его текущее сопоставленное значение. Весь вызов метода выполняется атомарно. Некоторые попытки обновления этой карты другими потоками могут быть заблокированы во время вычислений, поэтому вычисления должны быть короткими и простыми и не должны пытаться обновить какие-либо другие сопоставления этой карты.

Таким образом, при использовании этого метода всякий раз, когда есть запрос пользовательского потока для задачи T1, и если задача T1 находится в очереди или выполняется, но еще не завершена, тогда пользовательский поток будет ожидать выполнения этой задачи. Когда задача T1 будет выполнена, все пользовательские запросы, ожидающие выполнения задачи T1, будут уведомлены, а затем мы удалим задачу T1 из приведенной выше карты.

Ссылки на другие классы, используемые в приведенном ниже коде, присутствуют в этой ссылке.

TaskProcessor.java:

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;

public class TaskProcessor implements ITaskProcessor {

    //This map will contain all the tasks which are in queue and not yet completed
    //If there is scenario where there may be multiple tasks corresponding to same uniqueIdentifier, in that case below map can be modified accordingly to have the list of corresponding tasks which are not completed yet
    private final Map<String, Task> taskInProgresssByUniqueIdentifierMap = new ConcurrentHashMap<>();
    private final int QUEUE_SIZE = 100;
    private final BlockingQueue<Task> taskQueue = new ArrayBlockingQueue<Task>(QUEUE_SIZE);
    private final TaskRunner taskRunner = new TaskRunner();

    private Executor executor;
    private AtomicBoolean isStarted;
    private final DBManager dbManager = new DBManager();

    @Override
    public void start() {
        executor = Executors.newCachedThreadPool();
        while(isStarted.get()) {
            try {
                Task task = taskQueue.take();
                executeTaskInSeperateThread(task);

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    private void executeTaskInSeperateThread(Task task) {
        executor.execute(() -> {
            taskRunner.execute(task, new ITaskProgressListener() {

                @Override
                public void onTaskCompletion(TaskResult taskResult) {
                    task.setCompleted(true);
                    //TODO: we can also propagate the taskResult to waiting users, Implement it if it is required.
                    notifyAllWaitingUsers(task);
                }

                @Override
                public void onTaskFailure(Exception e) {
                    notifyAllWaitingUsers(task);
                }
            });
        });
    }

    private void notifyAllWaitingUsers(Task task) {
        taskInProgresssByUniqueIdentifierMap.computeIfPresent(task.getUniqueIdentifier(), new BiFunction<String, Task, Task>() {
            @Override
            public Task apply(String s, Task task) {
                synchronized (task) {
                    task.notifyAll();
                }
                return null;
            }
        });
    }
    //User thread
    @Override
    public ITaskResult getTaskResult(String uniqueIdentifier) {
        TaskResult result = null;
        Task task = taskInProgresssByUniqueIdentifierMap.computeIfPresent(uniqueIdentifier, new BiFunction<String, Task, Task>() {
            @Override
            public Task apply(String s, Task task) {
                synchronized (task) {
                    try {
                        //
                        task.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return task;
            }
        });
        //If task is null, it means the task was not there in queue, so we direcltly query the db for the task result
        if(task != null && !task.isCompleted()) {
            return null;  // Handle this condition gracefully, If task is not completed, it means there was some exception
        }
        ITaskResult taskResult = getResultFromDB(uniqueIdentifier); // At this point the result must be already saved in DB if the corresponding task has been processed ever.
        return taskResult;
    }

    private ITaskResult getResultFromDB(String uniqueIdentifier) {
        return dbManager.getTaskResult(uniqueIdentifier);
    }

     //Other thread 
    @Override
    public void enqueueTask(Task task) {
        if(isStarted.get()) {
            taskInProgresssByUniqueIdentifierMap.putIfAbsent(task.getUniqueIdentifier(), task);
            taskQueue.offer(task);
        }
    }

    @Override
    public void stop() {
        isStarted.compareAndSet(true, false);
    }
}

Дайте мне знать в комментариях, если у вас есть какие-либо вопросы. Благодарю.