JavaFX 多线程之 TaskServiceScheduledService
Posted Calvin Chan
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了JavaFX 多线程之 TaskServiceScheduledService相关的知识，希望对你有一定的参考价值。
JavaFX 多线程之 Task、Service、ScheduledService
一、开发环境
IntelliJ IDEA 2020.2.3 + JDK 1.8 + JavaFX Scene Builder 2.0
二、javafx.concurrent 包
JavaFX 场景图表示 JavaFX 应用程序的图形用户界面，它不是线程安全的，只能从 UI 线程（也称为 JavaFX 应用程序线程）访问和修改。在 JavaFX 应用程序线程上实现长时间运行的任务不可避免地会使应用程序 UI 无响应。最佳实践是在一个或多个后台线程上执行这些任务，并让 JavaFX 应用程序线程处理用户事件。
如果有特殊要求或需要额外的代码处理能力，通过创建一个 Runnable 对象和一个新线程来实现后台 worker 是一种合适的方法。请注意，在某些时候必须与 JavaFX 应用程序线程通信，无论是结果还是后台任务的进度。
对于大多数情况和大多数开发人员，推荐的方法是使用 javafx.concurrent 包提供的 JavaFX API，它负责与 UI 交互的多线程代码，并确保这种交互发生在正确的线程上。
Java 平台通过 java.util.concurrent 包提供了一整套可用的并发库。 javafx.concurrent 包通过考虑 JavaFX 应用程序线程和 GUI 开发人员面临的其他约束来利用现有 API。
javafx.concurrent 包由 Worker 接口和两个具体实现 Task 和 Service 类组成。 Worker 接口提供了一些 API，这些 API 对于后台 worker 与 UI 进行通信非常有用。 Task 类是 java.util.concurrent.FutureTask 类的一个完全可观察的实现。 Task 类使开发人员能够在 JavaFX 应用程序中实现异步任务。 Service 类执行任务。
WorkerStateEvent 类指定了每当 Worker 实现的状态发生变化时发生的事件。 Task 和 Service 类都实现了 EventTarget 接口，因此支持监听状态事件。
1、Worker 接口
Worker API 文档：Worker (JavaFX 2.2)
Worker 接口定义了一个对象，该对象在一个或多个后台线程上执行某些工作。 Worker 对象的状态可以从 JavaFX 应用程序线程中观察和使用。
Worker 对象的生命周期定义如下。创建时，Worker 对象处于 READY 状态。在计划进行工作后，Worker 对象将转换为 SCHEDULED 状态。之后，当 Worker 对象执行工作时，其状态变为 RUNNING。请注意，即使 Worker 对象在没有被调度的情况下立即启动，它也会首先转换为 SCHEDULED 状态，然后转换为 RUNNING 状态。成功完成的 Worker 对象的状态是 SUCCEEDED，value 属性设置为这个 Worker 对象的结果。否则，如果在 Worker 对象的执行过程中抛出任何异常，其状态将变为 FAILED，并且异常属性被设置为发生的异常的类型。在 Worker 对象结束之前的任何时间，开发人员都可以通过调用取消方法来中断它，该方法将 Worker 对象置于 CANCELED 状态。他们之间的关系如下图。
ScheduledService 对象生命周期的区别可以在 ScheduledService 类部分找到。
可以通过三个不同的属性（例如 totalWork，workDone 和 progress）获得由Worker对象完成的工作进度。
2、Task 类
Task API 文档：Task (JavaFX 2.2)
任务用于实现需要在后台线程上完成的工作逻辑。首先，需要扩展 Task 类。对 Task 类的实现必须覆盖 call 方法才能完成后台工作并返回结果。
call 方法是在后台线程上调用的，因此该方法只能操作可以安全地从后台线程读取和写入的状态。例如，从 call 方法操作活动场景图会引发运行时异常。另一方面，Task 类旨在与 JavaFX GUI 应用程序一起使用，它确保对公共属性、错误或取消的更改通知、事件处理程序和状态的任何更改都发生在 JavaFX 应用程序线程上。在 call 方法中，您可以使用 updateProgress、updateMessage、updateTitle 方法，这些方法更新 JavaFX 应用程序线程上相应属性的值。但是，如果任务被取消，则调用方法的返回值将被忽略。
请注意，Task 类适合 Java 并发库，因为它继承自实现 Runnable 接口的 java.utils.concurrent.FutureTask 类。因此，Task 对象可以在 Java 并发 Executor API 中使用，也可以作为参数传递给线程。您可以使用 FutureTask.run() 方法直接调用 Task 对象，该方法允许从另一个后台线程调用此任务。充分了解Java并发API将有助于了解JavaFX中的并发。
可以通过以下方式之一启动任务：
-
通过以给定任务作为参数启动线程：
Thread th = new Thread(task);
th.setDaemon(true);
th.start(); -
通过使用 ExecutorService API：
ExecutorService.submit(task);
Task 类定义了一个不能重复使用的一次性对象。 如果需要可重用的 Worker 对象，请使用 Service 类。
① 取消 Task
Java 中没有可靠的方法来停止进程中的线程。 但是，只要对任务调用取消，任务就必须停止处理。 任务应该在其工作期间定期检查它是否被调用方法主体中的 isCancelled 方法取消。 下面的示例显示了检查取消的 Task 类的正确实现。
import javafx.concurrent.Task;
Task<Integer> task = new Task<Integer>() {
@Override protected Integer call() throws Exception {
int iterations;
for (iterations = 0; iterations < 100000; iterations++) {
if (isCancelled()) {
break;
}
System.out.println("Iteration " + iterations);
}
return iterations;
}
};
如果任务实现有阻塞调用，例如 Thread.sleep 并且任务在阻塞调用中被取消，则抛出 InterruptedException。 对于这些任务，中断的线程可能是取消任务的信号。 因此，具有阻塞调用的任务必须仔细检查 isCancelled 方法，以确保由于任务取消而引发了 InterruptedException，如下例所示。
import javafx.concurrent.Task;
Task<Integer> task = new Task<Integer>() {
@Override protected Integer call() throws Exception {
int iterations;
for (iterations = 0; iterations < 1000; iterations++) {
if (isCancelled()) {
updateMessage("Cancelled");
break;
}
updateMessage("Iteration " + iterations);
updateProgress(iterations, 1000);
//Block the thread for a short time, but be sure
//to check the InterruptedException for cancellation
try {
Thread.sleep(100);
} catch (InterruptedException interrupted) {
if (isCancelled()) {
updateMessage("Cancelled");
break;
}
}
}
return iterations;
}
};
② 显示后台 Task 的进度
多线程应用程序中的一个典型用例是显示后台任务的进度。 假设有一个从一到一百万计数的后台任务和一个进度条，当计数器在后台运行时，必须更新此进度条上的进度。 下面示例显示了如何更新进度条。
import javafx.concurrent.Task;
Task task = new Task<Void>() {
@Override public Void call() {
static final int max = 1000000;
for (int i=1; i<=max; i++) {
if (isCancelled()) {
break;
}
updateProgress(i, max);
}
return null;
}
};
ProgressBar bar = new ProgressBar();
bar.progressProperty().bind(task.progressProperty());
new Thread(task).start()
首先，通过覆盖 call 方法来创建任务，在其中实现要完成的工作的逻辑并调用 updateProgress 方法，该方法更新任务的 progress、totalWork 和 workDone 属性。 这很重要，因为现在可以使用 progressProperty 方法来检索任务的进度并将进度条绑定到任务的进度。
3、Service 类
Service API 文档：Service (JavaFX 2.2)
Service 类旨在在一个或多个后台线程上执行 Task 对象。服务类方法和状态只能在 JavaFX 应用程序线程上访问。这个类的目的是帮助开发者实现后台线程和 JavaFX Application 线程之间的正确交互。
可以对Service对象进行以下控制：可以根据需要启动，取消和重新启动它。要启动 Service 对象，请使用 Service.start() 方法。
使用 Service 类，可以观察后台工作的状态并可选择取消它。稍后，可以重置服务并重新启动它。因此，可以声明性地定义服务并按需重新启动。
对于需要自动重启的服务，请参见 ScheduledService 类部分。
在实现 Service 类的子类时，务必将输入参数作为子类的属性公开给 Task 对象。
该服务可以通过以下方式之一执行：
- 通过 Executor 对象，如果它是为给定的服务指定的
- 通过守护线程，如果没有指定执行者
- 通过自定义执行程序，例如 ThreadPoolExecutor
下面示例显示了 Service 类的实现，它从任何 URL 读取第一行并将其作为字符串返回。
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.MalformedURLException;
import java.net.URL;
import javafx.application.Application;
import javafx.beans.property.SimpleStringProperty;
import javafx.beans.property.StringProperty;
import javafx.concurrent.Service;
import javafx.concurrent.Task;
import javafx.concurrent.WorkerStateEvent;
import javafx.event.EventHandler;
import javafx.stage.Stage;
public class FirstLineServiceApp extends Application {
@Override
public void start(Stage stage) throws Exception {
FirstLineService service = new FirstLineService();
service.setUrl("https://www.baidu.com/");
service.setOnSucceeded(new EventHandler<WorkerStateEvent>() {
@Override
public void handle(WorkerStateEvent t) {
System.out.println("done:" + t.getSource().getValue());
}
});
service.start();
}
public static void main(String[] args) {
launch();
}
private static class FirstLineService extends Service<String> {
private StringProperty url = new SimpleStringProperty();
public final void setUrl(String value) {
url.set(value);
}
public final String getUrl() {
return url.get();
}
public final StringProperty urlProperty() {
return url;
}
@Override
protected Task<String> createTask() {
return new Task<String>() {
@Override
protected String call()
throws IOException, MalformedURLException {
try ( BufferedReader in = new BufferedReader(
new InputStreamReader(
new URL(getUrl()).openStream;
in = new BufferedReader(
new InputStreamReader(u.openStream()))) {
return in.readLine();
}
}
};
}
}
4、WorkerStateEvent 类和状态转换
WorkerStateEvent API 文档：WorkerStateEvent (JavaFX 2.2)
每当 Worker 实现的状态发生变化时，就会发生一个由 WorkerStateEvent 类定义的适当事件。 例如，当 Task 对象转换到 SUCCEEDED 状态时，发生 WORKER_STATE_SUCCEEDED 事件，调用 onSucceeded 事件处理程序，之后在 JavaFX 应用程序线程上调用成功的受保护的便捷方法。
有 cancelled、failed、running、scheduled、successed 等几种受保护的便捷方法，它们在 Worker 实现转换到相应状态时被调用。 当状态更改以实现应用程序的逻辑时，这些方法可以被 Task 和 Service 类的子类覆盖。 下面示例显示了一个 Task 实现，它更新有关任务成功、取消和失败的状态消息。
import javafx.concurrent.Task;
Task<Integer> task = new Task<Integer>() {
@Override protected Integer call() throws Exception {
int iterations = 0;
for (iterations = 0; iterations < 100000; iterations++) {
if (isCancelled()) {
break;
}
System.out.println("Iteration " + iterations);
}
return iterations;
}
@Override protected void succeeded() {
super.succeeded();
updateMessage("Done!");
}
@Override protected void cancelled() {
super.cancelled();
updateMessage("Cancelled!");
}
@Override protected void failed() {
super.failed();
updateMessage("Failed!");
}
};
5、ScheduledService 类
许多涉及轮询的用例需要自动重启的服务。 为了满足这些需求，对 Service 类进行了扩展以生成 ScheduledService 类。 ScheduledService 类表示在成功执行后自动重启的服务，在特殊情况下，在失败时自动重启。
在创建时，ScheduledService 对象处于 READY 状态。
在调用 ScheduledService.start() 或 ScheduledService.restart() 方法后，ScheduledService 对象在延迟属性指定的持续时间内转换到 SCHEDULED 状态。
在 RUNNING 状态下，ScheduledService 对象执行其任务。
① Task 成功完成
任务完成后，ScheduledService 对象转换到 SUCCEEDED 状态，然后转换到 READY 状态，然后返回到 SCHEDULED 状态。 处于 SCHEDULED 状态的持续时间取决于上次转换到 RUNNING 状态的时间、当前时间以及 period 属性的值，该值定义了两次后续运行之间的最短时间。 如果之前的执行在时间段到期之前完成，则 ScheduledService 对象将保持 SCHEDULED 状态，直到时间段到期。 否则，如果前一次执行的时间超过指定的时间，则 ScheduledService 对象会立即转换为 RUNNING 状态。
② Task 失败
在任务以 FAILED 状态终止的情况下，ScheduledService 对象将重新启动或退出，具体取决于 restartOnFailure、backoffStrategy 和 maximumFailureCount 属性的值。
如果 restartOnFailure 属性为 false，则 ScheduledService 对象转换为 FAILED 状态并退出。在这种情况下，可以手动重新启动失败的 ScheduledService 对象。
如果 restartOnFailure 属性为 true，则 ScheduledService 对象转换到 SCHEDULED 状态并在 cumulativePeriod 属性的持续时间内保持该状态，这是调用 backoffStrategy 属性的结果。使用 cumulativePeriod 属性，可以强制失败的 ScheduledService 对象在下一次运行之前等待更长时间。 ScheduledService 成功完成后，cumulativePeriod 属性将重置为 period 属性的值。当后续失败的数量达到 maximumFailureCount 属性的值时，ScheduledService 对象转换为 FAILED 状态并退出。
在 ScheduledService 对象运行时发生在 delay 和 period 属性上的任何更改都将在下一次迭代中考虑在内。 delay 和 period 属性的默认值设置为 0。
三、代码实现
1、JavaFX 界面
2、创建线程池
JavaFX 多线程之 ThreadPoolExecutor 线程池
3、通过扩展 Task 类创建 Task
MultiThreadTask
import javafx.concurrent.Task;
/**
* Task继承了java.util,concurrency.FutureTask类。
* Task是不可以重复执行的Worker
* 运行方式：1. new Thread(task).start();
* 2.ExecutorService.submit(task);
* 3.其他。
* Task中有许多updateXXX方法用于改变相应的property。setOnXXX方法用于设置EventHandler.
* 继承Task需要覆盖其call()方法，call()方法会在后台线程中被调用。
* 由于java中并没有安全的线程中断机制，所以你必须在call()方法内部对isCancelled()进行判断
*
* @author wxhntmy
*/
public class MultiThreadTask extends Task<Void> {
/**
* 线程名称
*/
private String name;
/**
* 构造函数
* @param name 线程名称
*/
public MultiThreadTask(String name) {
this.name = name;
}
@Override
protected Void call() {
System.out.println(this.toString() + " is running!");
try {
Thread.sleep(100);
} catch (InterruptedException interrupted) {
if (isCancelled()) {
updateMessage("Cancelled");
}
}
updateMessage("Done!");
return null;
}
/**
* 获取任务名称
*
* @return 任务名称
*/
public String getName() {
return name;
}
@Override
public String toString() {
return "MyTask [ name = " + name + " ]";
}
}
启动任务：
MultiThreadTask multiThreadTask = new MultiThreadTask("new task 1");
Task<Integer> task = new Task<Integer>() {
@Override
protected Integer call() throws Exception {
int iterations;
for (iterations = 0; iterations < 10; iterations++) {
if (isCancelled()) {
break;
}
System.out.println("Iteration 2: " + iterations);
}
return iterations;
}
};
executorService.submit(multiThreadTask);
//不建议显式创建线程，请使用线程池。
//new Thread(multiThreadTask).start();
executorService.submit(task);
运行结果：
my-thread-1 has been created
MyTask [ name = new task 1 ] is running!
my-thread-2 has been created
Iteration 2: 0
Iteration 2: 1
Iteration 2: 2
Iteration 2: 3
Iteration 2: 4
Iteration 2: 5
Iteration 2: 6
Iteration 2: 7
Iteration 2: 8
Iteration 2: 9
4、带返回结果的 Task
import javafx.beans.property.ReadOnlyObjectProperty;
import javafx.beans.property.ReadOnlyObjectWrapper;
import javafx.concurrent.Task;
/**
* 返回部分结果的task
*
* @author wxhntmy
*/
public class ReturnResultTask extends Task<Object> {
/**
* 任务名称
*/
private String name;
/**
* 只读对象包装器
*/
private ReadOnlyObjectWrapper<String> results =
new ReadOnlyObjectWrapper<>(this, "Results", "");
/**
* 获取只读对象
* @return 只读对象
*/
public final String getResults() {
return results.get();
}
/**
* 获取只读对象属性
* @return 对象属性
*/
public final ReadOnlyObjectProperty<String> resultsProperty() {
return results.getReadOnlyProperty();
}
/**
* 构造函数
* @param name 任务名称
*/
public ReturnResultTask(String name) {
this.name = name;
}
@Override
protected String call() {
results.set("Results Get");
try {
Thread.sleep(100);
} catch (InterruptedException interrupted) {
if (isCancelled()) {
updateMessage("Cancelled");
}
}
updateMessage("Done!");
return results.getJavaFX 中的多线程会挂起 UI
是否可以将多线程JavaFX AudioClip声音的混合结果记录到磁盘上？
JavaFX实战：模拟电子琴弹奏效果，鼠标弹奏一曲piano送给大家
JavaFX实战：模拟电子琴弹奏效果，鼠标弹奏一曲piano送给大家