执行器暂停并且不处理挂起的队列

Question

我有一项长期的运行任务，即创建一个位图并保存它并重新创建更多位图，这是我在单个后台线程上所做的

ExecutorService executor = Executors.newSingleThreadExecutor();
Handler handler = new Handler(Looper.getMainLooper());
executor.execute(() -> {...bitmap work ...}

但是处理所有位图需要很长时间，所以我创建了一个线程池来使用多线程来加速任务。

private final int cores = Runtime.getRuntime().availableProcessors();
private final ExecutorService executor = Executors.newFixedThreadPool(cores + 1);
for (int i = 0; i < totalPage; i++) {    
    Runnable runnable = () -> {...bitmap work ...}
    executor.submit(runnable);
}

但每当我使用超过 1 个线程时，它会随机卡在某些任务上（比如 127 个中的 7 个），没有错误或任何它只是不再处理任何任务。我可以在执行者队列中查看待处理任务。但是如果我将线程池更改为使用 1 个线程，它可以正常工作并处理所有任务。

这是完整的实际代码

ExecutorService executor = Executors.newFixedThreadPool(cores + 1);
        List<Future<?>> futureList = new ArrayList<>();
        boolean allDone = false;

        try {
            //Convert pdf to Bitmap
            ParcelFileDescriptor parcelFileDescriptor = ParcelFileDescriptor.open(new File(pdfFileName), ParcelFileDescriptor.MODE_READ_ONLY);
            PdfRenderer pdfRenderer = new PdfRenderer(parcelFileDescriptor);
            int totalPage = pdfRenderer.getPageCount();

            final int[] counter = {1};
            for (int i = 0; i < totalPage; i++) {
                int finalI = i;
                String finalOriginalPdfName = originalPdfName;
                String finalGeneratedPdfName = generatedPdfName;
                Runnable runnable = () -> {
                    //pd.setMessage("Processing page " + (finalI + 1) + " of " + totalPage);
                    PdfRenderer.Page page = pdfRenderer.openPage(finalI);

                    Bitmap pageBitmap = Bitmap.createBitmap((300 * page.getWidth()) / 72, (300 * page.getHeight()) / 72, Bitmap.Config.ARGB_8888);
                    Canvas canvas = new Canvas(pageBitmap);
                    canvas.drawColor(Color.WHITE);
                    canvas.drawBitmap(pageBitmap, 0, 0, null);
                    page.render(pageBitmap, null, null, PdfRenderer.Page.RENDER_MODE_FOR_PRINT);
                    page.close();


                    //Crop bitmaps and temporarily store on app data directory

                    for (int k = 0; k < SlipBoundingBox.Y.length; k++) {
                        for (int j = 0; j < SlipBoundingBox.X.length; j++) {
                            Bitmap slipBitmap = Bitmap.createBitmap(pageBitmap, SlipBoundingBox.X[j], SlipBoundingBox.Y[k], SlipBoundingBox.WIDTH, SlipBoundingBox.HEIGHT);
                            //Filename formation originalPdfName_generatePdfName_pdfPageIndex_x_y.extension
                            File slip = new File(
                                    getExternalFilesDir("slips")
                                            + "/"
                                            + finalOriginalPdfName
                                            + "_"
                                            + finalGeneratedPdfName
                                            + "_"
                                            + finalI +
                                            "_"
                                            + SlipBoundingBox.X[j]
                                            + "_"
                                            + SlipBoundingBox.Y[k]
                                            + "_.jpg");
                            try (FileOutputStream out1 = new FileOutputStream(slip)) {
                                slipBitmap.compress(Bitmap.CompressFormat.JPEG, 100, out1);
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                            slipBitmap.recycle();
                        }
                    }
                    pageBitmap.recycle();
                    pd.setMessage("Processed " + counter[0] + " of " + totalPage + " pages");
                    counter[0]++;
                };
                Future<?> future = executor.submit(runnable);
                Log.d(TAG, "processPdf: " + future.isDone());
                futureList.add(future);

            }
            //Todo close pdfrender on all page processed
            //pdfRenderer.close();

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

Answer 1

非常抱歉无法直接帮助您。我不知道 PdfRenderer 的内部工作原理。我认为三件事可能会造成问题：

PdfRenderer 在其库中某处出现死锁。不太可能。
您的文件 I/O 非常复杂，并行写入多个文件只需花费 100 倍的时间。还记得旧的光盘 (CD) 吗？当你全速阅读一个资源时，一旦有 2 个以上的进程读取，它就会以指数方式变慢（不是快一半，而是 2 个进程慢 20 倍）。我认为这是最有可能的原因。
PdfRenderer 内部某处发生了 I/O 错误。也许它打开了太多的文件句柄或其他什么，导致意外崩溃，只是在原地停止执行。

所以我唯一能给你的就是：分析是否有任何进程在工作（至少是缓慢的）。可能有一些框架，但我不知道。还有很多工具可以在运行time 分析 JVM。这是一个更简单的解决方案：

检查每个线程的状态，以及它们是否完全改变了方法。现在这纯粹是视觉效果：

检查 out/use 下面的代码。运行它as-is
你会看到每个线程不时地task/state变化
如果你喜欢，请将我的文件添加到你的项目中。
调整你的代码（见我的 main() 方法）
reduce/limit 池大小为 3 或 4
控制台输出将直观地显示任何线程是否有任何变化。
如果根本没有变化，那么pdf库就卡住了
如果有重复的变化，但长期没有任何进展，库就会陷入死循环

代码：

package Whosebug;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class SimpleThreadpoolAnalysis {

    static private final long sStartMS = System.currentTimeMillis();

    static public long now() {
        return System.currentTimeMillis() - sStartMS;
    }

    static public void sleep(final long pMS) {
        try {
            Thread.sleep(pMS);
        } catch (final InterruptedException e) { /* */ }
    }



    private final Set<Thread> mWorkingThreads = Collections.synchronizedSet(new HashSet<>());

    private int mPrintStackDepth = 5;

    public void setPrintStackDepth(final int pPrintStackDepth) {
        mPrintStackDepth = pPrintStackDepth;
    }

    public void runChecked(final Runnable pLambda) {
        if (pLambda == null) return;

        final Thread currentThread = Thread.currentThread();
        try {
            System.out.println("SimpleThreadpoolAnalysis.runChecked() adding thread " + currentThread.getName());
            mWorkingThreads.add(currentThread);
            pLambda.run();
        } finally {
            System.out.println("SimpleThreadpoolAnalysis.runChecked() removing thread " + currentThread.getName());
            mWorkingThreads.remove(currentThread);
        }
    }

    public void printMiniStackTraces() {
        System.out.println("Working Threads at " + now());

        for (final Thread t : mWorkingThreads) {
            System.out.println("\tThread " + t.getId() + ": " + t.getName());
            final StackTraceElement[] st = t.getStackTrace();
            for (int i = 0; i < Math.min(st.length, mPrintStackDepth); i++) {
                System.out.println("\t\t" + st[i]);
            }
        }
    }

    public Thread runSupervisorThread(final int pUpdateEveryXMs, final long pDurationMS) {
        System.out.println("Supervisor Thread starting...");
        final Thread t = new Thread(() -> runSupervisorThread_(pUpdateEveryXMs, pDurationMS), "Pool Supervisor Thread");
        t.setDaemon(true);
        t.start();
        return t;
    }
    private void runSupervisorThread_(final int pUpdateEveryXMs, final long pDurationMS) {
        System.out.println("Supervisor Thread starting...");
        final int NUMBER_OF_RUNS = (int) (pDurationMS / pUpdateEveryXMs);
        for (int i = 0; i < NUMBER_OF_RUNS; i++) {
            System.out.flush();
            sleep(pUpdateEveryXMs);
            printMiniStackTraces();
        }
        System.out.println("Supervisor Thread ending...");
    }



    // this is my dummy class. you can remove this once you use it on your code
    static public class TestWorker implements Runnable {
        @Override public void run() {
            while (true) {
                //              final int no =
                subDelegator();
                //              System.out.println("Got  " + no);
            }
        }

        private int subDelegator() {
            SimpleThreadpoolAnalysis.sleep((long) (Math.random() * 1000));

            final int randomIndex = (int) (Math.random() * 10);
            switch (randomIndex) {
                case 0:
                    return run_0();
                case 1:
                    return run_1();
                case 2:
                    return run_2();
                case 3:
                    return run_3();
                case 4:
                    return run_4();
                case 5:
                    return run_5();
                default:
                    return -1;
            }
        }
        private int run_0() {
            SimpleThreadpoolAnalysis.sleep(500);
            return 0;
        }
        private int run_1() {
            SimpleThreadpoolAnalysis.sleep(1000);
            return 1;
        }
        private int run_2() {
            SimpleThreadpoolAnalysis.sleep(2000);
            return 2;
        }
        private int run_3() {
            SimpleThreadpoolAnalysis.sleep(3000);
            return 3;
        }
        private int run_4() {
            SimpleThreadpoolAnalysis.sleep(4000);
            return 4;
        }
        private int run_5() {
            SimpleThreadpoolAnalysis.sleep(5000);
            return 5;
        }
    }



    public static void main(final String[] args) {
        final SimpleThreadpoolAnalysis sta = new SimpleThreadpoolAnalysis();
        sta.runSupervisorThread(100, 60000); // will run for a minute, updating every 100ms
        // this will run a in background thread, so if other threads are done, this will end automatically, too

        final int cores = Runtime.getRuntime().availableProcessors();
        final ExecutorService executor = Executors.newFixedThreadPool(cores + 1);
        final int totalPages = 10;
        for (int i = 0; i < totalPages; i++) {
            // my code:
            //          final Runnable runnable = new TestWorker(); // this would be the normal call
            final Runnable runnable = new TestWorker(); // this is the checked version

            // your code: enable this, comment out my line above
            //final Runnable runnable = () -> { /* your bitmap work */ }; // use this, just like in your code

            final Runnable checkedRunnable = () -> sta.runChecked(runnable); // this is the checked version
            final Future<?> future = executor.submit(checkedRunnable);
            // ... some more of your code ...
        }

    }



}

希望对您有所帮助。

Answer 2

原来问题是我在回收位图之前关闭了 pdf 页面。

page.close();
.....
pageBitmap.recycle();

我在回收位图后移动了page.close()，线程不再挂了

pageBitmap.recycle();
page.close();

执行器暂停并且不处理挂起的队列

Executor pauses and doesn't process pending queues

java

multithreading

android