java - 在后台线程中终止长时间运行的任务

Question

我有一个压缩图像的任务，其中使用了很多循环:

private void writeCompressedData() {
    int i, j, r, c, a, b;
    loat[][] inputArray;

    for (r = 0; r < minBlockHeight; r++) {
        for (c = 0; c < minBlockWidth; c++) {
            xpos = c * 8;
            pos = r * 8;
            for (comp = 0; comp < jpegObj.numberOfComponents; comp++) {
                inputArray = (float[][]) jpegObj.components[comp];

                 for (i = 0; i < jpegObj.VsampFactor[comp]; i++) {
                     for (j = 0; j < jpegObj.HsampFactor[comp]; j++) {
                         xblockoffset = j * 8;
                         yblockoffset = i * 8;
                         for (a = 0; a < 8; a++) {
                             for (b = 0; b < 8; b++) {
                                 // Process some data and put to inputArray
                             }
                         }

                         // Encode Huffman block 
                     }
                 }
             }
         }
     }
}

我在这样的普通线程中运行此方法:

new Thread(new Runnable() {
    @Override
    public void run() {
        writeCompressedData();
    }
});

或者在后台工作线程中运行

TaskExecutor.queueRunnable(new Runnable() {
    @Override
    public void run() {
        writeCompressedData();
    }
});

问题是:当接收到无效输入时，此方法有时会出错并导致无限循环。在这种情况下，它将永远运行并且即使设备的屏幕关闭也会损坏 CPU，这会增加设备的温度(如果我使用工作线程，它还会阻塞等待队列中的其他任务)。

我想我需要设置一个超时来终止长时间运行的任务。在普通 Java 线程中实现此目的的最佳方法是什么？ RxJava 支持吗？

---

我知道我需要修复的是“错误”的方法，而不仅仅是终止它。但是对于大应用来说，很难控制其他开发者的代码，我首先要做的就是避免影响用户。

Answer 1

您需要某种形式的协作取消，例如检查一个或多个嵌套循环内的 Thread.currentThread().isInterrupted()。

for (/* ... */) {
    if (Thread.currentThread().isInterrupted()) return;

    for (/* ... */) {

        if (Thread.currentThread().isInterrupted()) return;

        for (/* ... */) {
            // the tightest loop
        }             
    }
}

然后当你运行该方法时，保留Thread/Future并调用interrupt/cancel(true):

backgroundTask = new Thread(() -> method());
backgroundTask.start();
// ...
backgroundTask.interrupt();

backgroundFuture = executorService.submit(() -> method());
// ...
backgroundFuture.cancel(true);

在 RxJava 中，这看起来像这样:

backgroundDisposable = Completable.fromAction(() -> method())
.subscribeOn(Schedulers.io()) // dedicated thread recommended
.observeOn(AndroidSchedulers.mainThread())
.subscribe(() -> { /* done */ }, e -> { /* error */ });

// ...
backgroundDisposable.dispose();