Неблокирующий потокобезопасный счетчик для JavaFX

Question

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

public void handleSomeButtonClick(){
    if(!dataModel.getSomeList().isEmpty()) {
        boolean unlimited = false;
        int count = 0;
        AtomicInteger successCount = new AtomicInteger(0);

        if(countSelector.getValue().equalsIgnoreCase("Unlimited"))
            unlimited = true;
        else
            count = Integer.parseInt(countSelector.getValue());

        while(unlimited || successCount.get() < count) {
            Task task = getSomeTask();
            taskExecutor.submit(task);
            task.setOnSucceeded(event -> {
                if (task.getValue())
                    log.info("Successfully Completed Task | Total Count: " + successCount.incrementAndGet());
                else
                    log.error("Failed task");
            });
        }
    }
}

Answer 1

Ваш цикл ожидает выполнения определенного количества задач.Это может быть даже бесконечный цикл.

Это не очень хорошая идея:

• Вы блокируете вызывающий поток, который кажется потоком приложения JavaFX.
• Вы не можете контролировать, сколько задач отправлено.count может быть 3, но, поскольку вы только планируете задачи в цикле, 1000 или более задач могут быть созданы и запланированы до завершения первой.

Более того, если вы используете onSucceeded / onFailed вам не нужно использовать AtomicInteger или какой-либо подобный тип синхронизации, поскольку все эти обработчики выполняются в потоке приложения JavaFX.

Ваш код можно переписать так:

private int successCount;

private void scheduleTask(final boolean unlimited) {
    Task task = getSomeTask();
    task.setOnSucceeded(event -> {
        // cannot get a Boolean from a raw task, so I assume the task is successfull iff no exception happens
        successCount++;
        log.info("Successfully Completed Task | Total Count: " + successCount);
        if (unlimited) {
            // submit new task, if the number of tasks is unlimited
            scheduleTask(true);
        }
    });
    // submit new task on failure
    task.setOnFailed(evt -> scheduleTask(unlimited));
    taskExecutor.submit(task);
}

public void handleSomeButtonClick() {
    if(!dataModel.getSomeList().isEmpty()) {
        successCount = 0;
        final boolean unlimited;
        final int count;

        if(countSelector.getValue().equalsIgnoreCase("Unlimited")) {
            unlimited = true;
            count = 4; // set limit of number of tasks submitted to the executor at the same time
        } else {
            count = Integer.parseInt(countSelector.getValue());
            unlimited = false;
        }

        for (int i = 0; i < count; i++) {
            scheduleTask(unlimited);
        }
    }
}

Примечание: В этом коде существует риск многократного нажатия на handleButtonClick до завершения предыдущих задач.Вам следует либо запретить планирование новых задач до завершения старых, либо использовать для счетчика некоторый ссылочный тип, содержащий int, создать этот объект в handleSomeButtonClick и передать этот объект в scheduleTask.

Answer 2

Ваша проблема в блоке future.get () и дождитесь результата.Это будет просто, если вы используете библиотеку Vavr.Потому что он может прикрепить код к своему будущему, который запускается автоматически при успехе или неудаче.Так что вам не нужно ждать.Вот пример, который использует будущее Vavr.

        CheckedFunction0<String> thisIsATask = () -> {
        if ( /*do something*/ ){
            throw new Exception("Hey");
        }
        return "ABC";
    };

    List<Future<String>> futureList = new ArrayList<>();

    for (int x = 0; x < 10; x++) {
        futureList.add(Future.of(getExecutorService(), thisIsATask));
    }

    futureList.forEach((task) -> {
        // This will run if success
        task.onSuccess(s -> s.equals("ABC") ? Platform.runLater(()->UpdateCounter()) : wtf());
        // Your get the exception if it is fail;
        task.onFailure(e -> e.printStackTrace());
        // task.onComplete() will run on any case when complete
    });

Это не блокировка, код в onSucess onFailure или onComplete будет выполняться, когда задача завершится или будет получено исключение.

Примечание: Future.of будет использовать executorService, который вы передаете для запуска каждой задачи в новом потоке, код, который вы предоставляете в onSuccess, будет продолжать выполняться в этом потоке после выполнения задачи, поэтому, если вы вызываете javafx, запомните Platform.runLater ()

Также, если вы хотите запустить что-то, когда все задачи завершены, тогда

    // the code at onComplete will run when tasks all done
    Future<Seq<String>> all = Future.sequence(futureList);
    all.onComplete((i) -> this.btnXYZ.setDisable(false));

Answer 3

Ваш пользовательский интерфейс заблокирован означает, что вы выполняете подсчет (successCount.get ()

if (2) затем выполните весь цикл while в фоновом потоке, обновите пользовательский интерфейс в Platform-> runlater ().

если (1) использовать Future / CompletableFuture или более мощную версию Future в стороннем пакете, например vavr.