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

Question

В системе у меня есть объект - назовем его TaskProcessor.Он содержит очередь задач, которые выполняются некоторым пулом потоков (ExecutorService + PriorityBlockingQueue). Результат каждой задачи сохраняется в базе данных под некоторым уникальным идентификатором.

Пользователь, который знает этоуникальный идентификатор, может проверить результат этой задачи.Результат может быть в базе данных, но задача также может ожидать выполнения в очереди.В этом случае UserThread должен дождаться завершения задачи.

Кроме того, допустимы следующие предположения:

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

• UserThread и TaskProcess находятся в одном приложении.TaskProcessor содержит пул потоков, а UserThread - просто поток сервлетов.

• UserThread должен быть заблокирован при запросе результата, и результат еще не завершен.UserThread должен быть разблокирован сразу после TaskProcessor завершения задачи (или задач), сгруппированных по уникальному идентификатору

Моя первая попытка (наивная) состояла в том, чтобы проверить результат взацикливайтесь и спите некоторое время:

// UserThread
while(!checkResultIsInDatabase(uniqueIdentifier))
  sleep(someTime)

Но мне это не нравится.Прежде всего, я трачу соединения с базой данных.Более того, если задача будет завершена сразу после сна, то пользователь будет ждать, даже если результат только что появился.

Следующая попытка была основана на ожидании / уведомлении:

//UserThread 
while (!checkResultIsInDatabase())
  taskProcessor.wait()

//TaskProcessor
... some complicated calculations
this.notifyAll()

Но ятоже не нравитсяЕсли больше UserThreads будет использовать TaskProcessor, то они будут без необходимости просыпаться каждый раз, когда будет выполнено какое-то задание, и более того - они будут делать ненужные вызовы базы данных.

Последняя попытка была основана на чем-то, что язвонил waitingRoom:

//UserThread
Object mutex = new Object();
taskProcessor.addToWaitingRoom(uniqueIdentifier, mutex)
while (!checkResultIsInDatabase())
  mutex.wait()

//TaskProcessor
... Some complicated calculations
if (uniqueIdentifierExistInWaitingRoom(taskUniqueIdentifier))
  getMutexFromWaitingRoom(taskUniqueIdentifier).notify()

Но, похоже, это небезопасно.Между проверкой базы данных и wait() задача может быть выполнена (notify() не будет работать, поскольку UserThread еще не вызывал wait()), что может привести к тупику.

Itкажется, что я должен это где-то синхронизировать.Но я боюсь, что это будет не эффективно.Есть ли способ исправить любые мои попытки, сделать их безопасными и эффективными?Или, может быть, есть какой-то другой, лучший способ сделать это?

Answer 1

Вы, похоже, ищете какую-то будущую / обещанную абстракцию. Взгляните на 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() и т. П.

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

Answer 2

Я полагаю, что замена 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()

Answer 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, если кто-то еще поставил его в очередь.

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

Работает как брелок .

Answer 4

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

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

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

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

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

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

Таким образом, при использовании этого метода всякий раз, когда есть запрос пользовательского потока для задачи 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);
    }
}

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